• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import os
2import time
3
4from test.support import MS_WINDOWS
5from .results import TestResults
6from .runtests import RunTests
7from .utils import print_warning
8
9if MS_WINDOWS:
10    from .win_utils import WindowsLoadTracker
11
12
13class Logger:
14    def __init__(self, results: TestResults, quiet: bool, pgo: bool):
15        self.start_time = time.perf_counter()
16        self.test_count_text = ''
17        self.test_count_width = 3
18        self.win_load_tracker: WindowsLoadTracker | None = None
19        self._results: TestResults = results
20        self._quiet: bool = quiet
21        self._pgo: bool = pgo
22
23    def log(self, line: str = '') -> None:
24        empty = not line
25
26        # add the system load prefix: "load avg: 1.80 "
27        load_avg = self.get_load_avg()
28        if load_avg is not None:
29            line = f"load avg: {load_avg:.2f} {line}"
30
31        # add the timestamp prefix:  "0:01:05 "
32        log_time = time.perf_counter() - self.start_time
33
34        mins, secs = divmod(int(log_time), 60)
35        hours, mins = divmod(mins, 60)
36        formatted_log_time = "%d:%02d:%02d" % (hours, mins, secs)
37
38        line = f"{formatted_log_time} {line}"
39        if empty:
40            line = line[:-1]
41
42        print(line, flush=True)
43
44    def get_load_avg(self) -> float | None:
45        if hasattr(os, 'getloadavg'):
46            try:
47                return os.getloadavg()[0]
48            except OSError:
49                pass
50        if self.win_load_tracker is not None:
51            return self.win_load_tracker.getloadavg()
52        return None
53
54    def display_progress(self, test_index: int, text: str) -> None:
55        if self._quiet:
56            return
57        results = self._results
58
59        # "[ 51/405/1] test_tcl passed"
60        line = f"{test_index:{self.test_count_width}}{self.test_count_text}"
61        fails = len(results.bad) + len(results.env_changed)
62        if fails and not self._pgo:
63            line = f"{line}/{fails}"
64        self.log(f"[{line}] {text}")
65
66    def set_tests(self, runtests: RunTests) -> None:
67        if runtests.forever:
68            self.test_count_text = ''
69            self.test_count_width = 3
70        else:
71            self.test_count_text = '/{}'.format(len(runtests.tests))
72            self.test_count_width = len(self.test_count_text) - 1
73
74    def start_load_tracker(self) -> None:
75        if not MS_WINDOWS:
76            return
77
78        try:
79            self.win_load_tracker = WindowsLoadTracker()
80        except PermissionError as error:
81            # Standard accounts may not have access to the performance
82            # counters.
83            print_warning(f'Failed to create WindowsLoadTracker: {error}')
84
85    def stop_load_tracker(self) -> None:
86        if self.win_load_tracker is None:
87            return
88        self.win_load_tracker.close()
89        self.win_load_tracker = None
90