• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# coding=utf-8
3
4#
5# Copyright (c) 2020-2023 Huawei Device Co., Ltd.
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10#     http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17#
18
19from abc import ABC
20from abc import abstractmethod
21
22from _core.context.life_stage import StageEvent
23from _core.context.life_stage import ILifeStageListener
24from _core.context.result import ExecuteFinished
25from _core.error import ErrorMessage
26from _core.exception import ParamError
27from _core.interface import IScheduler
28from _core.logger import platform_logger
29
30LOG = platform_logger("Abs")
31
32__all__ = ["Sub"]
33
34
35class Sub(IScheduler, ABC):
36    _is_need_auto_retry = False
37    _is_running = True
38    _repeat_index = 1
39
40    def exec_command(self, command, options):
41        if command != "run":
42            raise ParamError(ErrorMessage.Common.Code_0101012.format(command))
43        exec_type = options.exectype
44        if exec_type in self._exec_type_():
45            self._exec_task(options)
46        else:
47            LOG.error(ErrorMessage.Common.Code_0101013.format(exec_type))
48
49    def _register(self):
50        from _core.context.service import ContextImpl
51        from _core.context.service import ActiveState
52        ContextImpl().update_state(ActiveState(self))
53        if not ContextImpl().read_attr("xDevice"):
54            ContextImpl().write_attr("xDevice", True)
55
56    def _exec_task(self, options):
57        """
58        Directly allocates a device and execute a device test.
59        """
60        task = None
61        self.remove_life_stage_listener()
62        self._register()
63        try:
64            Sub._is_running = True
65            self._on_task_prepare_(options)
66            task = self.__discover__(options.__dict__)
67            self.__execute__(task)
68        except (ParamError, ValueError, TypeError, SyntaxError,
69                AttributeError) as exception:
70            LOG.exception(exception, exc_info=False)
71            self._on_task_error_(task, exception)
72        finally:
73            self._on_task_finished_()
74            Sub._is_running = False
75
76    @classmethod
77    @abstractmethod
78    def __max_command_size__(cls) -> int:
79        pass
80
81    @classmethod
82    def terminate_cmd_exec(cls):
83        cls._is_running = False
84        result = cls._call_terminate()
85        if result:
86            return result
87        return None
88
89    @classmethod
90    def is_executing(cls):
91        return cls._is_running
92
93    @classmethod
94    @abstractmethod
95    def _exec_type_(cls) -> list:
96        pass
97
98    def _on_task_prepare_(self, options):
99        pass
100
101    def _on_task_finished_(self):
102        pass
103
104    def _on_task_error_(self, task, exception: Exception):
105        pass
106
107    def _on_execute_finished_(self, task, result: ExecuteFinished):
108        pass
109
110    @abstractmethod
111    def _do_execute_(self, task):
112        pass
113
114    @classmethod
115    @abstractmethod
116    def _call_terminate(cls):
117        pass
118
119    @classmethod
120    @abstractmethod
121    def _reset_environment(cls, environment="", config_file=""):
122        pass
123
124    @abstractmethod
125    def generate_task_report(self, task):
126        pass
127
128    @classmethod
129    def is_need_auto_retry(cls):
130        return cls._is_need_auto_retry
131
132    @classmethod
133    def set_need_auto_retry(cls, is_need: bool):
134        cls._is_need_auto_retry = is_need
135
136    @classmethod
137    def get_repeat_index(cls):
138        return cls._repeat_index
139
140    @classmethod
141    def set_repeat_index(cls, value: int):
142        cls._repeat_index = value
143
144    @classmethod
145    @abstractmethod
146    def notify_stage(cls, stage_event: StageEvent):
147        pass
148
149    @classmethod
150    @abstractmethod
151    def add_life_stage_listener(cls, listener: ILifeStageListener):
152        pass
153
154    @classmethod
155    @abstractmethod
156    def remove_life_stage_listener(cls):
157        pass
158