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