• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# coding=utf-8
3
4#
5# Copyright (c) 2020-2023 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
19import os
20from abc import ABC
21from abc import abstractmethod
22from xml.etree import ElementTree
23
24from _core.constants import LifeStage
25from _core.report.reporter_helper import ReportConstant
26from _core.logger import platform_logger
27from _core.testkit.json_parser import JsonParser
28
29LOG = platform_logger("LifeStage")
30
31class StageEvent(ABC):
32    def __init__(self):
33        self._event_type: str = self._stage_name()
34        self._data = dict()
35
36    @abstractmethod
37    def _covert_data(self):
38        pass
39
40    def get_data(self):
41        if not self._data:
42            self._data.update(self._get_data())
43        return self._data
44
45    def _get_data(self):
46        data = self._covert_data()
47        data.update({"type": self._event_type})
48        return data
49
50    @abstractmethod
51    def _stage_name(self) -> str:
52        pass
53
54
55
56class TaskStart(StageEvent):
57
58    def __init__(self, action: str):
59        super().__init__()
60        self.action: str = action
61
62    def _stage_name(self) -> str:
63        return LifeStage.task_start
64
65    def _covert_data(self) -> dict:
66        from xdevice import Variables
67        data = {
68            "id": Variables.task_id,
69            "name": Variables.task_name,
70            "command": self.action,
71        }
72        return data
73
74
75class TaskEnd(StageEvent):
76
77    def __init__(self, report_path, unavailable, error_msg):
78        super().__init__()
79        self._report_path: str = report_path
80        self._unavailable: int = unavailable
81        self._error_msg = error_msg
82
83    def _stage_name(self) -> str:
84        return LifeStage.task_end
85
86    def _covert_data(self):
87        from xdevice import Variables
88        summary_data_report = os.path.join(self._report_path, ReportConstant.summary_data_report)
89        if not os.path.exists(summary_data_report):
90            LOG.error("Call lifecycle error, summary report {} not exists".format(self._report_path))
91            passed = failures = blocked = 0
92            unavailable = self._unavailable
93        else:
94            task_element = ElementTree.parse(summary_data_report).getroot()
95            total_tests = int(task_element.get(ReportConstant.tests, 0))
96            failures = int(task_element.get(ReportConstant.failures, 0))
97            blocked = int(task_element.get(ReportConstant.disabled, 0))
98            ignored = int(task_element.get(ReportConstant.ignored, 0))
99            unavailable = int(task_element.get(ReportConstant.unavailable, 0))
100            passed = total_tests - failures - blocked - ignored
101
102        data = {
103            "id": Variables.task_id,
104            "name": Variables.task_name,
105            "passed": passed,
106            "failures": failures,
107            "blocked": blocked,
108            "unavailable": unavailable,
109            "error": self._error_msg,
110        }
111        return data
112
113
114class CaseStart(StageEvent):
115
116    def __init__(self, case_name, driver):
117        super().__init__()
118        self._case_name = case_name
119        self.driver = driver
120        self.driver_name = ""
121        self.device_options = dict()
122        self._get_device_options()
123
124    def _get_device_options(self):
125        case_json = self.driver[1].source
126        json_config = None
127        try:
128            if case_json.config_file:
129                json_config = JsonParser(case_json.config_file)
130            # To be compatible with old code
131            elif case_json.source_string and (
132                    # Importing is_config_str() in _core.util will cause circular import error
133                    "{" in case_json.source_string and "}" in case_json.source_string
134            ):
135                json_config = JsonParser(case_json.source_string)
136            else:
137                LOG.warning('No json config file or the json config file is empty!')
138        except Exception as e:
139            LOG.error(f'Error loading json config file because {e}')
140        if json_config:
141            self.device_options = json_config.get_environment()
142            self.driver_name = json_config.get_driver_type()
143
144    def _stage_name(self) -> str:
145        return LifeStage.case_start
146
147    def _covert_data(self) -> dict:
148        from xdevice import Variables
149        data = {
150            "id": Variables.task_id,
151            "driver": self.driver_name,
152            "devices": self.device_options,
153            "name": self._case_name
154        }
155        return data
156
157
158class CaseEnd(StageEvent):
159
160    def __init__(self, case_name, case_result, error_msg=""):
161        super().__init__()
162        self._case_name = case_name
163        self._case_result = case_result
164        self._error_msg = error_msg
165
166    def _stage_name(self) -> str:
167        return LifeStage.case_end
168
169    def _covert_data(self) -> dict:
170        from xdevice import Variables
171        data = {
172            "id": Variables.task_id,
173            "name": self._case_name,
174            "case_result": self._case_result,
175            "error_msg": self._error_msg,
176        }
177        return data
178
179
180class ILifeStageListener(ABC):
181    __slots__ = ()
182
183    @abstractmethod
184    def __on_event__(self, stage_event: StageEvent):
185        pass
186