• 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 an 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    @classmethod
105    def __subclasshook__(cls, class_info):
106        if cls is IDevice:
107            return _check_methods(class_info, "__serial__")
108        return NotImplemented
109
110    @abstractmethod
111    def get(self, key=None, default=None):
112        if not key:
113            return default
114        value = getattr(self, key, None)
115        if value:
116            return value
117        else:
118            return self.extend_value.get(key, default)
119
120
121class IDriver(ABC):
122    """
123    A test driver runs the tests and reports results to a listener.
124    """
125    __slots__ = ()
126
127    @classmethod
128    def __check_failed__(cls, msg):
129        raise ValueError(msg)
130
131    @abstractmethod
132    def __check_environment__(self, device_options):
133        """
134        Check environment correct or not.
135        You should return False when check failed.
136        :param device_options:
137        """
138
139    @abstractmethod
140    def __check_config__(self, config):
141        """
142        Check config correct or not.
143        You should raise exception when check failed.
144        :param config:
145        """
146        self.__check_failed__("Not implementation for __check_config__")
147
148    @abstractmethod
149    def __execute__(self, request):
150        """
151        Execute tests according to the request.
152        """
153
154    @classmethod
155    def __dry_run_execute__(self, request):
156        """
157        Dry run tests according to the request.
158        """
159        pass
160
161    @abstractmethod
162    def __result__(self):
163        """
164        Return tests execution result
165        """
166
167    @classmethod
168    def __subclasshook__(cls, class_info):
169        if cls is IDriver:
170            return _check_methods(class_info, "__check_config__",
171                                  "__execute__")
172        return NotImplemented
173
174
175class IScheduler(ABC):
176    """
177    A scheduler to run jobs parallel.
178    """
179    __slots__ = ()
180
181    @abstractmethod
182    def __discover__(self, args):
183        """
184        Discover tests according to request, and return root TestDescriptor.
185        """
186
187    @abstractmethod
188    def __execute__(self, request):
189        """
190        Execute tests according to the request.
191        """
192
193    @classmethod
194    @abstractmethod
195    def __allocate_environment__(cls, options, test_driver):
196        """
197        Allocate environment according to the request.
198        """
199
200    @classmethod
201    @abstractmethod
202    def __free_environment__(cls, environment):
203        """
204        Free environment to the request.
205        """
206
207    @classmethod
208    def __subclasshook__(cls, class_info):
209        if cls is IScheduler:
210            return _check_methods(class_info, "__discover__", "__execute__")
211        return NotImplemented
212
213
214class IListener(ABC):
215    """
216    Listener to be notified of test execution events by TestDriver, as
217    following sequence:
218    __started__(TestTask)
219    __started__(TestSuite)
220    __started__(TestCase)
221    [__skipped__(TestCase)]
222    [__failed__(TestCase)]
223    __ended__(TestCase)
224    ...
225    [__failed__(TestSuite)]
226    __ended__(TestSuite)
227    ...
228    [__failed__(TestTask)]
229    __ended__(TestTask)
230    """
231    __slots__ = ()
232
233    @abstractmethod
234    def __started__(self, lifecycle, result):
235        """
236        Called when the execution of the TestCase or TestTask has started,
237        before any test has been executed.
238        """
239
240    @abstractmethod
241    def __ended__(self, lifecycle, result, **kwargs):
242        """
243        Called when the execution of the TestCase or TestTask has finished,
244        after all tests have been executed.
245        """
246
247    @abstractmethod
248    def __skipped__(self, lifecycle, result):
249        """
250        Called when the execution of the TestCase or TestTask has been skipped.
251        """
252
253    @abstractmethod
254    def __failed__(self, lifecycle, result):
255        """
256        Called when the execution of the TestCase or TestTask has been skipped.
257        """
258
259    @classmethod
260    def __subclasshook__(cls, class_info):
261        if cls is IListener:
262            return _check_methods(class_info, "__started__", "__ended__",
263                                  "__skipped__", "__failed__")
264        return NotImplemented
265
266
267class IShellReceiver(ABC):
268    """
269    Read the output from shell out.
270    """
271    __slots__ = ()
272
273    @abstractmethod
274    def __read__(self, output):
275        pass
276
277    @abstractmethod
278    def __error__(self, message):
279        pass
280
281    @abstractmethod
282    def __done__(self, result_code, message):
283        pass
284
285    @classmethod
286    def __subclasshook__(cls, class_info):
287        if cls is IShellReceiver:
288            return _check_methods(class_info, "__read__", "__error__",
289                                  "__done__")
290        return NotImplemented
291
292
293class IParser(ABC):
294    """
295    A parser to parse the output of testcases.
296    """
297    __slots__ = ()
298
299    @abstractmethod
300    def __process__(self, lines):
301        pass
302
303    @abstractmethod
304    def __done__(self):
305        pass
306
307    @classmethod
308    def __subclasshook__(cls, class_info):
309        if cls is IParser:
310            return _check_methods(class_info, "__process__", "__done__")
311        return NotImplemented
312
313
314class ITestKit(ABC):
315    """
316    A test kit running on the host.
317    """
318    __slots__ = ()
319
320    @classmethod
321    def __check_failed__(cls, msg):
322        raise ValueError(msg)
323
324    @abstractmethod
325    def __check_config__(self, config):
326        """
327        Check config correct or not.
328        You should raise exception when check failed.
329        :param config:
330        """
331        self.__check_failed__("Not implementation for __check_config__")
332
333    @abstractmethod
334    def __setup__(self, device, **kwargs):
335        pass
336
337    @abstractmethod
338    def __teardown__(self, device):
339        pass
340
341    @classmethod
342    def __subclasshook__(cls, class_info):
343        if cls is ITestKit:
344            return _check_methods(class_info, "__check_config__", "__setup__",
345                                  "__teardown__")
346        return NotImplemented
347
348
349class IReporter(ABC):
350    """
351    A reporter to generate reports
352    """
353    __slots__ = ()
354
355    @abstractmethod
356    def __generate_reports__(self, report_path, **kwargs):
357        pass
358
359    @classmethod
360    def __subclasshook__(cls, class_info):
361        if cls is IReporter:
362            return _check_methods(class_info, "__generate_reports__")
363        return NotImplemented
364
365
366class IFilter(ABC):
367    """
368    A filter is used to filter xml node and selector on the manager
369    """
370    __slots__ = ()
371
372    @abstractmethod
373    def __filter_xml_node__(self, node):
374        pass
375
376    @abstractmethod
377    def __filter_selector__(self, selector):
378        pass
379