• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# coding=utf-8
3
4#
5# Copyright (c) 2020-2021 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
21from enum import Enum
22
23__all__ = ["LifeCycle", "IDevice", "IDriver", "IListener", "IShellReceiver",
24           "IParser", "ITestKit", "IScheduler", "IDeviceManager", "IReporter"]
25
26
27class LifeCycle(Enum):
28    TestTask = "TestTask"
29    TestSuite = "TestSuite"
30    TestCase = "TestCase"
31    TestSuites = "TestSuites"
32
33
34def _check_methods(class_info, *methods):
35    mro = class_info.__mro__
36    for method in methods:
37        for cls in mro:
38            if method in cls.__dict__:
39                if cls.__dict__[method] is None:
40                    return NotImplemented
41                break
42        else:
43            return NotImplemented
44    return True
45
46
47class IDeviceManager(ABC):
48    """
49    Class managing the set of different types of devices for testing
50    """
51    __slots__ = ()
52    support_labels = []
53
54    @abstractmethod
55    def apply_device(self, device_option, timeout=10):
56        pass
57
58    @abstractmethod
59    def release_device(self, device):
60        pass
61
62    @classmethod
63    def __subclasshook__(cls, class_info):
64        if cls is IDevice:
65            return _check_methods(class_info, "__serial__")
66        return NotImplemented
67
68    @abstractmethod
69    def init_environment(self, environment, user_config_file):
70        pass
71
72    @abstractmethod
73    def env_stop(self):
74        pass
75
76    @abstractmethod
77    def list_devices(self):
78        pass
79
80
81class IDevice(ABC):
82    """
83    IDevice provides an reliable and slightly higher level API to access
84    devices
85    """
86    __slots__ = ()
87    extend_value = {}
88
89    @abstractmethod
90    def __set_serial__(self, device_sn=""):
91        pass
92
93    @abstractmethod
94    def __get_serial__(self):
95        pass
96
97    @classmethod
98    def __subclasshook__(cls, class_info):
99        if cls is IDevice:
100            return _check_methods(class_info, "__serial__")
101        return NotImplemented
102
103    @abstractmethod
104    def get(self, key=None, default=None):
105        if not key:
106            return default
107        value = getattr(self, key, None)
108        if value:
109            return value
110        else:
111            return self.extend_value.get(key, default)
112
113
114class IDriver(ABC):
115    """
116    A test driver runs the tests and reports results to a listener.
117    """
118    __slots__ = ()
119
120    @classmethod
121    def __check_failed__(cls, msg):
122        raise ValueError(msg)
123
124    @abstractmethod
125    def __check_environment__(self, device_options):
126        """
127        check environment correct or not.
128        you should return False when check failed.
129        :param device_options:
130        """
131
132    @abstractmethod
133    def __check_config__(self, config):
134        """
135        check config correct or not.
136        you should raise exception when check failed.
137        :param config:
138        """
139        self.__check_failed__("Not implementation for __check_config__")
140
141    @abstractmethod
142    def __execute__(self, request):
143        """
144        Execute tests according to the request.
145        """
146
147    @abstractmethod
148    def __result__(self):
149        """
150        Return tests execution result
151        """
152
153    @classmethod
154    def __subclasshook__(cls, class_info):
155        if cls is IDriver:
156            return _check_methods(class_info, "__check_config__",
157                                  "__execute__")
158        return NotImplemented
159
160
161class IScheduler(ABC):
162    """
163    A scheduler to run jobs parallel.
164    """
165    __slots__ = ()
166
167    @abstractmethod
168    def __discover__(self, args):
169        """
170        Discover tests according to request, and return root TestDescriptor.
171        """
172
173    @abstractmethod
174    def __execute__(self, request):
175        """
176        Execute tests according to the request.
177        """
178
179    @classmethod
180    @abstractmethod
181    def __allocate_environment__(cls, options, test_driver):
182        """
183        allocate_environment according to the request.
184        """
185
186    @classmethod
187    @abstractmethod
188    def __free_environment__(cls, environment):
189        """
190        free environment to the request.
191        """
192
193    @classmethod
194    def __subclasshook__(cls, class_info):
195        if cls is IScheduler:
196            return _check_methods(class_info, "__discover__", "__execute__")
197        return NotImplemented
198
199
200class IListener(ABC):
201    """
202    Listener to be notified of test execution events by TestDriver, as
203    following sequence:
204    __started__(TestTask)
205    __started__(TestSuite)
206    __started__(TestCase)
207    [__skipped__(TestCase)]
208    [__failed__(TestCase)]
209    __ended__(TestCase)
210    ...
211    [__failed__(TestSuite)]
212    __ended__(TestSuite)
213    ...
214    [__failed__(TestTask)]
215    __ended__(TestTask)
216    """
217    __slots__ = ()
218
219    @abstractmethod
220    def __started__(self, lifecycle, result):
221        """
222        Called when the execution of the TestCase or TestTask has started,
223        before any test has been executed.
224        """
225
226    @abstractmethod
227    def __ended__(self, lifecycle, result, **kwargs):
228        """
229        Called when the execution of the TestCase or TestTask has finished,
230        after all tests have been executed.
231        """
232
233    @abstractmethod
234    def __skipped__(self, lifecycle, result):
235        """
236        Called when the execution of the TestCase or TestTask has been skipped.
237        """
238
239    @abstractmethod
240    def __failed__(self, lifecycle, result):
241        """
242        Called when the execution of the TestCase or TestTask has been skipped.
243        """
244
245    @classmethod
246    def __subclasshook__(cls, class_info):
247        if cls is IListener:
248            return _check_methods(class_info, "__started__", "__ended__",
249                                  "__skipped__", "__failed__")
250        return NotImplemented
251
252
253class IShellReceiver(ABC):
254    """
255    read the output from shell out.
256    """
257    __slots__ = ()
258
259    @abstractmethod
260    def __read__(self, output):
261        pass
262
263    @abstractmethod
264    def __error__(self, message):
265        pass
266
267    @abstractmethod
268    def __done__(self, result_code, message):
269        pass
270
271    @classmethod
272    def __subclasshook__(cls, class_info):
273        if cls is IShellReceiver:
274            return _check_methods(class_info, "__read__", "__error__",
275                                  "__done__")
276        return NotImplemented
277
278
279class IParser(ABC):
280    """
281    A parser to parse the output of testcases.
282    """
283    __slots__ = ()
284
285    @abstractmethod
286    def __process__(self, lines):
287        pass
288
289    @abstractmethod
290    def __done__(self):
291        pass
292
293    @classmethod
294    def __subclasshook__(cls, class_info):
295        if cls is IParser:
296            return _check_methods(class_info, "__process__", "__done__")
297        return NotImplemented
298
299
300class ITestKit(ABC):
301    """
302    A test kit running on the host.
303    """
304    __slots__ = ()
305
306    @classmethod
307    def __check_failed__(cls, msg):
308        raise ValueError(msg)
309
310    @abstractmethod
311    def __check_config__(self, config):
312        """
313        check config correct or not.
314        you should raise exception when check failed.
315        :param config:
316        """
317        self.__check_failed__("Not implementation for __check_config__")
318
319    @abstractmethod
320    def __setup__(self, device, **kwargs):
321        pass
322
323    @abstractmethod
324    def __teardown__(self, device):
325        pass
326
327    @classmethod
328    def __subclasshook__(cls, class_info):
329        if cls is ITestKit:
330            return _check_methods(class_info, "__check_config__", "__setup__",
331                                  "__teardown__")
332        return NotImplemented
333
334
335class IReporter(ABC):
336    """
337    A reporter to generate reports
338    """
339    __slots__ = ()
340
341    @abstractmethod
342    def __generate_reports__(self, report_path, **kwargs):
343        pass
344
345    @classmethod
346    def __subclasshook__(cls, class_info):
347        if cls is IReporter:
348            return _check_methods(class_info, "__generate_reports__")
349        return NotImplemented
350