File: stats.py

package info (click to toggle)
libreswan 5.2-2.2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 81,632 kB
  • sloc: ansic: 129,988; sh: 32,018; xml: 20,646; python: 10,303; makefile: 3,022; javascript: 1,506; sed: 574; yacc: 511; perl: 264; awk: 52
file content (97 lines) | stat: -rw-r--r-- 4,045 bytes parent folder | download | duplicates (2)
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
#!/usr/bin/env python3

# Collect test statistics
#
# Copyright (C) 2015-2016 Andrew Cagney <cagney@gnu.org>
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.  See <https://www.gnu.org/licenses/gpl2.txt>.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
# for more details.

from collections import defaultdict
import threading

from fab import logutil

class Results:

    def __init__(self):
        # Use a default dict so no need to worry about initializing
        # values to zero.
        self.counts = defaultdict(lambda: defaultdict(set))
        self.lock = threading.Lock()

    def _add(self, *keys_value, domain=None):
        keys = keys_value[0:-1]
        value = keys_value[-1]
        with self.lock:
            key = "/".join(keys)
            # Include the "None" domain in the set so that it is always
            # non-empty - makes iterating over it is easier.  See
            # log_details.
            self.counts[key][value].add(domain)

    def log_summary(self, log, header=None, footer=None, prefix=""):
        with self.lock:
            if len(self.counts):
                header and log(header)
                for key, values in sorted(self.counts.items()):
                    log("%s%s: %d", prefix, key, len(values))
                footer and log(footer)

    def log_details(self, log, header=None, footer=None, prefix=""):
        with self.lock:
            if len(self.counts):
                header and log(header)
                for key, values in sorted(self.counts.items()):
                    # First invert value:domain creating domain:value
                    table = defaultdict(set)
                    for value, domains in sorted(values.items()):
                        for domain in domains:
                            table[domain].add(value)
                    # Second log key[/domain]: value ...
                    for domain, values in sorted(table.items()):
                        line = ""
                        for value in sorted(values):
                            line += " " + value
                        if domain:
                            log("%s%s/%s:%s", prefix, key, domain, line)
                        else:
                            log("%s%s:%s", prefix, key, line)
                footer and log(footer)

    def _count_result(self, result):
        self._add("total",                                              result.test.name)
        self._add("total",                                 str(result), result.test.name)
        self._add("tests", result.test.status, "results", str(result), result.test.name)
        # details
        for issue in result.issues:
            for domain in result.issues[issue]:
                self._add("tests", result.test.status, "errors", issue, result.test.name, domain=domain)

    def _count_previous(self, result, previous):
        self._add("stats", previous.test.status,
                   result, "previous=" + str(previous), previous.test.name)

    def add_ignored(self, test, reason):
        """The test has been excluded from the test run"""
        self._add("total", test.name)
        self._add("total", "ignored", test.name)
        self._add("tests", test.status, "results", "untested", test.name)
        self._add("stats", test.status, "ignored", reason, test.name)

    def add_skipped(self, result):
        """The test wasn't run; log the previous result"""
        self._count_result(result)
        self._count_previous("skipped", result)

    def add_result(self, result, old_result=None):
        self._count_result(result)
        if old_result:
            self._count_previous(str(result), old_result)