• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# coding=utf-8
3
4#
5# Copyright (c) 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#
18from devicetest.core.constants import RunResult
19from devicetest.core.error_message import ErrorMessage
20from devicetest.core.exception import DeviceTestError
21from devicetest.core.report import ReportHandler
22from devicetest.log.logger import DeviceTestLog as log
23from xdevice import SuiteReporter
24
25
26class UploadResultHandler:
27
28    def __init__(self, report_path):
29        self.__log = log
30        self.__test_runner = None
31        self.__report_path = None
32        self.__upload_suitereporter_lock = False
33        self.report_handler = ReportHandler(report_path)
34
35    def set_test_runner(self, _log, test_runner):
36        self.__log = _log
37        self.__test_runner = test_runner
38        self.__log.info("finish set test runner.")
39
40    def get_error_msg(self, test_runner, is_cur_case_error=False):
41        if test_runner.project.record.get_is_manual_stop_status():
42            error_msg = ErrorMessage.Error_01300.Topic if is_cur_case_error \
43                else ErrorMessage.Error_01301.Topic
44
45        else:
46            error_msg = ErrorMessage.Error_01400.Topic if is_cur_case_error \
47                else ErrorMessage.Error_01404.Topic
48        return error_msg
49
50    def flash_os_test_results(self, test_runner, test_results):
51        cur_case_error_msg = self.get_error_msg(test_runner,
52                                                is_cur_case_error=True)
53        if not test_results:
54            test_result = test_runner.record_cls_result(
55                test_runner.project.execute_case_name,
56                error=cur_case_error_msg)
57            test_results.append(test_result)
58        else:
59            if test_results[-1].get("result").strip() == RunResult.FAILED \
60                    and not test_results[0].get("error").strip():
61                test_results[0]["error"] = cur_case_error_msg
62        return test_results
63
64    def upload_suite_reporter(self, test_results=None):
65        report_result_tuple = self.report_handler.generate_test_report(
66            self.__test_runner, test_results)
67        SuiteReporter.append_report_result(report_result_tuple)
68        self.__log.debug("result tuple:{}".format(report_result_tuple))
69        self.__log.info("upload suitereporter success.")
70
71    def upload_suitereporter(self, is_stoped=False):
72        try:
73            self.upload_suite_reporter()
74        except DeviceTestError as err:
75            log.error(err)
76        except Exception:
77            self.__log.error(ErrorMessage.Error_01427.Message.en,
78                             error_no=ErrorMessage.Error_01427.Code,
79                             is_traceback=True)
80