summaryrefslogtreecommitdiffstats
path: root/meta/lib/oeqa/utils/logparser.py
blob: cc6d18d94a0236b5a6688cda49414d6eabb8e392 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
#!/usr/bin/env python
#
# SPDX-License-Identifier: MIT
#

import sys
import os
import re

# A parser that can be used to identify weather a line is a test result or a section statement.
class PtestParser(object):
    def __init__(self):
        self.results = {}
        self.sections = {}

    def parse(self, logfile):
        test_regex = {}
        test_regex['PASSED'] = re.compile(r"^PASS:(.+)")
        test_regex['FAILED'] = re.compile(r"^FAIL:(.+)")
        test_regex['SKIPPED'] = re.compile(r"^SKIP:(.+)")

        section_regex = {}
        section_regex['begin'] = re.compile(r"^BEGIN: .*/(.+)/ptest")
        section_regex['end'] = re.compile(r"^END: .*/(.+)/ptest")
        section_regex['duration'] = re.compile(r"^DURATION: (.+)")
        section_regex['exitcode'] = re.compile(r"^ERROR: Exit status is (.+)")
        section_regex['timeout'] = re.compile(r"^TIMEOUT: .*/(.+)/ptest")

        def newsection():
            return { 'name': "No-section", 'log': "" }

        current_section = newsection()

        with open(logfile, errors='replace') as f:
            for line in f:
                result = section_regex['begin'].search(line)
                if result:
                    current_section['name'] = result.group(1)
                    continue

                result = section_regex['end'].search(line)
                if result:
                    if current_section['name'] != result.group(1):
                        bb.warn("Ptest END log section mismatch %s vs. %s" % (current_section['name'], result.group(1)))
                    if current_section['name'] in self.sections:
                        bb.warn("Ptest duplicate section for %s" % (current_section['name']))
                    self.sections[current_section['name']] = current_section
                    del self.sections[current_section['name']]['name']
                    current_section = newsection()
                    continue

                result = section_regex['timeout'].search(line)
                if result:
                    if current_section['name'] != result.group(1):
                        bb.warn("Ptest TIMEOUT log section mismatch %s vs. %s" % (current_section['name'], result.group(1)))
                    current_section['timeout'] = True
                    continue

                for t in ['duration', 'exitcode']:
                    result = section_regex[t].search(line)
                    if result:
                        current_section[t] = result.group(1)
                        continue

                current_section['log'] = current_section['log'] + line 

                for t in test_regex:
                    result = test_regex[t].search(line)
                    if result:
                        if current_section['name'] not in self.results:
                            self.results[current_section['name']] = {}
                        self.results[current_section['name']][result.group(1)] = t

        return self.results, self.sections

    # Log the results as files. The file name is the section name and the contents are the tests in that section.
    def results_as_files(self, target_dir):
        if not os.path.exists(target_dir):
            raise Exception("Target directory does not exist: %s" % target_dir)

        for section in self.results:
            prefix = 'No-section'
            if section:
                prefix = section
            section_file = os.path.join(target_dir, prefix)
            # purge the file contents if it exists
            with open(section_file, 'w') as f:
                for test_name in sorted(self.results[section]):
                    status = self.results[section][test_name]
                    f.write(status + ": " + test_name + "\n")


# ltp log parsing
class LtpParser(object):
    def __init__(self):
        self.results = {}
        self.section = {'duration': "", 'log': ""}

    def parse(self, logfile):
        test_regex = {}
        test_regex['PASSED'] = re.compile(r"PASS")
        test_regex['FAILED'] = re.compile(r"FAIL")
        test_regex['SKIPPED'] = re.compile(r"SKIP")

        with open(logfile, errors='replace') as f:
            for line in f:
                for t in test_regex:
                    result = test_regex[t].search(line)
                    if result:
                        self.results[line.split()[0].strip()] = t

        for test in self.results:
            result = self.results[test]
            self.section['log'] = self.section['log'] + ("%s: %s\n" % (result.strip()[:-2], test.strip()))

        return self.results, self.section


# ltp Compliance log parsing
class LtpComplianceParser(object):
    def __init__(self):
        self.results = {}
        self.section = {'duration': "", 'log': ""}

    def parse(self, logfile):
        test_regex = {}
        test_regex['PASSED'] = re.compile(r"^PASS")
        test_regex['FAILED'] = re.compile(r"^FAIL")
        test_regex['SKIPPED'] = re.compile(r"(?:UNTESTED)|(?:UNSUPPORTED)")

        section_regex = {}
        section_regex['test'] = re.compile(r"^Testing")

        with open(logfile, errors='replace') as f:
            for line in f:
                result = section_regex['test'].search(line)
                if result:
                    self.name = ""
                    self.name = line.split()[1].strip()
                    self.results[self.name] = "PASSED"
                    failed = 0

                failed_result = test_regex['FAILED'].search(line)
                if failed_result:
                    failed = line.split()[1].strip()
                    if int(failed) > 0:
                        self.results[self.name] = "FAILED"

        for test in self.results:
            result = self.results[test]
            self.section['log'] = self.section['log'] + ("%s: %s\n" % (result.strip()[:-2], test.strip()))

        return self.results, self.section