1#
2# SPDX-License-Identifier: MIT
3#
4
5import sys
6import os
7import re
8
9# A parser that can be used to identify weather a line is a test result or a section statement.
10class PtestParser(object):
11    def __init__(self):
12        self.results = {}
13        self.sections = {}
14
15    def parse(self, logfile):
16        test_regex = {}
17        test_regex['PASSED'] = re.compile(r"^PASS:(.+)")
18        test_regex['FAILED'] = re.compile(r"^FAIL:([^(]+)")
19        test_regex['SKIPPED'] = re.compile(r"^SKIP:(.+)")
20
21        section_regex = {}
22        section_regex['begin'] = re.compile(r"^BEGIN: .*/(.+)/ptest")
23        section_regex['end'] = re.compile(r"^END: .*/(.+)/ptest")
24        section_regex['duration'] = re.compile(r"^DURATION: (.+)")
25        section_regex['exitcode'] = re.compile(r"^ERROR: Exit status is (.+)")
26        section_regex['timeout'] = re.compile(r"^TIMEOUT: .*/(.+)/ptest")
27
28        # Cache markers so we don't take the re.search() hit all the time.
29        markers = ("PASS:", "FAIL:", "SKIP:", "BEGIN:", "END:", "DURATION:", "ERROR: Exit", "TIMEOUT:")
30
31        def newsection():
32            return { 'name': "No-section", 'log': [] }
33
34        current_section = newsection()
35
36        with open(logfile, errors='replace') as f:
37            for line in f:
38                if not line.startswith(markers):
39                    current_section['log'].append(line)
40                    continue
41
42                result = section_regex['begin'].search(line)
43                if result:
44                    current_section['name'] = result.group(1)
45                    continue
46
47                result = section_regex['end'].search(line)
48                if result:
49                    if current_section['name'] != result.group(1):
50                        bb.warn("Ptest END log section mismatch %s vs. %s" % (current_section['name'], result.group(1)))
51                    if current_section['name'] in self.sections:
52                        bb.warn("Ptest duplicate section for %s" % (current_section['name']))
53                    self.sections[current_section['name']] = current_section
54                    del self.sections[current_section['name']]['name']
55                    current_section = newsection()
56                    continue
57
58                result = section_regex['timeout'].search(line)
59                if result:
60                    if current_section['name'] != result.group(1):
61                        bb.warn("Ptest TIMEOUT log section mismatch %s vs. %s" % (current_section['name'], result.group(1)))
62                    current_section['timeout'] = True
63                    continue
64
65                for t in ['duration', 'exitcode']:
66                    result = section_regex[t].search(line)
67                    if result:
68                        current_section[t] = result.group(1)
69                        continue
70
71                current_section['log'].append(line)
72
73                for t in test_regex:
74                    result = test_regex[t].search(line)
75                    if result:
76                        if current_section['name'] not in self.results:
77                            self.results[current_section['name']] = {}
78                        self.results[current_section['name']][result.group(1).strip()] = t
79
80        # Python performance for repeatedly joining long strings is poor, do it all at once at the end.
81        # For 2.1 million lines in a log this reduces 18 hours to 12s.
82        for section in self.sections:
83            self.sections[section]['log'] = "".join(self.sections[section]['log'])
84
85        return self.results, self.sections
86
87    # Log the results as files. The file name is the section name and the contents are the tests in that section.
88    def results_as_files(self, target_dir):
89        if not os.path.exists(target_dir):
90            raise Exception("Target directory does not exist: %s" % target_dir)
91
92        for section in self.results:
93            prefix = 'No-section'
94            if section:
95                prefix = section
96            section_file = os.path.join(target_dir, prefix)
97            # purge the file contents if it exists
98            with open(section_file, 'w') as f:
99                for test_name in sorted(self.results[section]):
100                    status = self.results[section][test_name]
101                    f.write(status + ": " + test_name + "\n")
102
103
104# ltp log parsing
105class LtpParser(object):
106    def __init__(self):
107        self.results = {}
108        self.section = {'duration': "", 'log': ""}
109
110    def parse(self, logfile):
111        test_regex = {}
112        test_regex['PASSED'] = re.compile(r"PASS")
113        test_regex['FAILED'] = re.compile(r"FAIL")
114        test_regex['SKIPPED'] = re.compile(r"SKIP")
115
116        with open(logfile, errors='replace') as f:
117            for line in f:
118                for t in test_regex:
119                    result = test_regex[t].search(line)
120                    if result:
121                        self.results[line.split()[0].strip()] = t
122
123        for test in self.results:
124            result = self.results[test]
125            self.section['log'] = self.section['log'] + ("%s: %s\n" % (result.strip()[:-2], test.strip()))
126
127        return self.results, self.section
128
129
130# ltp Compliance log parsing
131class LtpComplianceParser(object):
132    def __init__(self):
133        self.results = {}
134        self.section = {'duration': "", 'log': ""}
135
136    def parse(self, logfile):
137        test_regex = {}
138        test_regex['PASSED'] = re.compile(r"^PASS")
139        test_regex['FAILED'] = re.compile(r"^FAIL")
140        test_regex['SKIPPED'] = re.compile(r"(?:UNTESTED)|(?:UNSUPPORTED)")
141
142        section_regex = {}
143        section_regex['test'] = re.compile(r"^Testing")
144
145        with open(logfile, errors='replace') as f:
146            for line in f:
147                result = section_regex['test'].search(line)
148                if result:
149                    self.name = ""
150                    self.name = line.split()[1].strip()
151                    self.results[self.name] = "PASSED"
152                    failed = 0
153
154                failed_result = test_regex['FAILED'].search(line)
155                if failed_result:
156                    failed = line.split()[1].strip()
157                    if int(failed) > 0:
158                        self.results[self.name] = "FAILED"
159
160        for test in self.results:
161            result = self.results[test]
162            self.section['log'] = self.section['log'] + ("%s: %s\n" % (result.strip()[:-2], test.strip()))
163
164        return self.results, self.section
165