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