• 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#
18
19from ohos.parser import *
20
21__all__ = ["OHRustTestParser"]
22
23LOG = platform_logger("Parser")
24
25
26@Plugin(type=Plugin.PARSER, id=CommonParserType.oh_rust)
27class OHRustTestParser(IParser):
28
29    def __init__(self):
30        self.test_pattern = "test (?:tests::)?(.+) ... (ok|FAILED)"
31        self.stout_pattern = "---- tests::(.+) stdout ----"
32        self.running_pattern = "running (\\d+) test|tests"
33        self.test_result_pattern = \
34            "test result: (ok|FAILED)\\..+finished in (.+)s"
35        self.suite_name = ""
36        self.result_list = list()
37        self.stdout_list = list()
38        self.failures_stdout = list()
39        self.cur_fail_case = ""
40        self.state_machine = StateRecorder()
41        self.listeners = []
42
43    def get_listeners(self):
44        return self.listeners
45
46    def __process__(self, lines):
47        for line in lines:
48            LOG.debug(line)
49            self.parse(line)
50
51    def __done__(self):
52        self.handle_suite_end()
53
54    def parse(self, line):
55        if line.startswith("running"):
56            matcher = re.match(self.running_pattern, line)
57            if not (matcher and matcher.group(1)):
58                return
59            self.handle_suite_start(matcher)
60        elif line.startswith("test result:"):
61            matcher = re.match(self.test_result_pattern, line)
62            if not (matcher and matcher.group(2)):
63                return
64            self.handle_case_lifecycle(matcher)
65
66        elif "..." in line:
67            matcher = re.match(self.test_pattern, line)
68            if not (matcher and matcher.group(1) and matcher.group(2)):
69                return
70            self.collect_case(matcher)
71        elif line.startswith("---- tests::"):
72            matcher = re.match(self.stout_pattern, line)
73            if not (matcher and matcher.group(1)):
74                return
75            self.cur_fail_case = matcher.group(1)
76        else:
77            if self.cur_fail_case:
78                self.handle_stdout(line)
79
80    def handle_case_lifecycle(self, matcher):
81        cost_time = matcher.group(2)
82        for test_result in self.result_list:
83            if test_result.code == ResultCode.FAILED.value:
84                if self.stdout_list and \
85                        self.stdout_list[0][0] == test_result.test_name:
86                    test_result.stacktrace = self.stdout_list[0][1]
87                    self.stdout_list.pop(0)
88            test_result.current = self.state_machine.running_test_index + 1
89            for listener in self.get_listeners():
90                test_result = copy.copy(test_result)
91                listener.__started__(LifeCycle.TestCase, test_result)
92            for listener in self.get_listeners():
93                result = copy.copy(test_result)
94                listener.__ended__(LifeCycle.TestCase, result)
95        test_suite = self.state_machine.suite()
96        test_suite.run_time = float(cost_time) * 1000
97
98    def handle_stdout(self, line):
99        if line.strip():
100            self.failures_stdout.append(line.strip())
101        else:
102            self.stdout_list.append((self.cur_fail_case,
103                                     " ".join(self.failures_stdout)))
104            self.cur_fail_case = ""
105            self.failures_stdout.clear()
106
107    def collect_case(self, matcher):
108        test_result = self.state_machine.test(reset=True)
109        test_result.test_class = self.suite_name
110        test_result.test_name = matcher.group(1)
111        test_result.code = ResultCode.PASSED.value if \
112            matcher.group(2) == "ok" else ResultCode.FAILED.value
113        self.result_list.append(test_result)
114
115    def handle_suite_start(self, matcher):
116        self.state_machine.suite(reset=True)
117        test_suite = self.state_machine.suite()
118        test_suite.suite_name = self.suite_name
119        test_suite.test_num = int(matcher.group(1))
120
121        for listener in self.get_listeners():
122            suite_report = copy.copy(test_suite)
123            listener.__started__(LifeCycle.TestSuite, suite_report)
124
125    def handle_suite_end(self):
126        suite_result = self.state_machine.suite()
127        suite_result.run_time += suite_result.run_time
128        suite_result.is_completed = True
129        for listener in self.get_listeners():
130            suite = copy.copy(suite_result)
131            listener.__ended__(LifeCycle.TestSuite, suite, suite_report=True)
132