1import logging
2
3import pytest
4
5from ambassador_diag.diagd import ReconfigStats
6
7
8def assert_checks(r: ReconfigStats, when: float, want_check: bool, want_timers: bool) -> None:
9 got_check = r.needs_check(when)
10 assert got_check == want_check, f"{when}: wanted check {want_check}, got {got_check}"
11
12 got_timers = r.needs_timers(when)
13 assert got_timers == want_timers, f"{when}: wanted timers {want_timers}, got {got_timers}"
14
15
16def test_reconfig_stats():
17 logging.basicConfig(
18 level=logging.DEBUG,
19 format="%(asctime)s ffs %(levelname)s: %(message)s",
20 datefmt="%Y-%m-%d %H:%M:%S",
21 )
22
23 logger = logging.getLogger("ffs")
24 logger.setLevel(logging.DEBUG)
25
26 r = ReconfigStats(
27 logger,
28 max_incr_between_checks=5,
29 max_time_between_checks=20,
30 max_config_between_timers=2,
31 max_time_between_timers=10,
32 )
33
34 r.dump()
35
36 r.mark("complete", 10)
37 assert_checks(r, 11, False, False)
38 assert_checks(r, 12, False, False)
39 r.mark("incremental", 10)
40 assert_checks(r, 14, False, True) # Need timers from outstanding
41 assert_checks(r, 18, False, True)
42 r.mark_timers_logged(20)
43 assert_checks(r, 20, False, False)
44 r.mark("diag", 21)
45 assert_checks(r, 22, False, False)
46 assert_checks(r, 30, False, False)
47 assert_checks(r, 32, True, True)
48 r.mark_checked(False, 32)
49
50 r.mark("incremental", 33)
51 assert_checks(r, 34, False, True)
52 r.mark_timers_logged(34)
53 r.mark("incremental", 35)
54 assert_checks(r, 36, False, False)
55 r.mark("incremental", 37)
56 assert_checks(r, 38, False, True)
57 r.mark_timers_logged(38)
58 r.mark("incremental", 39)
59 assert_checks(r, 40, False, False)
60 r.mark("incremental", 41)
61 assert_checks(r, 42, True, True)
62 r.mark_timers_logged(42)
63 r.mark_checked(True, 42)
64 r.mark("incremental", 43)
65 assert_checks(r, 44, False, False)
66 r.mark("incremental", 45)
67 assert_checks(r, 46, False, True)
68 r.mark("incremental", 47)
69 assert_checks(r, 48, False, True)
70 r.mark("incremental", 49)
71 assert_checks(r, 50, False, True)
72 r.mark("incremental", 51)
73 assert_checks(r, 52, True, True)
74 r.mark_checked(True, 52)
75 assert_checks(r, 55, False, True)
76
77 assert_checks(r, 74, False, True)
78
79 assert_checks(r, 84, False, True)
80
81 r.mark("complete", 100)
82 assert_checks(r, 101, False, True)
83 r.mark("incremental", 102)
84 r.mark_timers_logged(102)
85 assert_checks(r, 103, False, False)
86 r.mark("incremental", 104)
87 assert_checks(r, 105, False, False)
88
89 r.dump()
90
91 assert r.counts["incremental"] == 13
92 assert r.counts["complete"] == 2
93 assert r.incrementals_outstanding == 2
94 assert r.checks == 3
95 assert r.errors == 1
96
97
98if __name__ == "__main__":
99 import sys
100
101 pytest.main(sys.argv)
View as plain text