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