Lab: test an operational analysis toolkit

cd .local/advanced-design
python3 --version

import unittest
from bisect import bisect_left
from random import Random

from complexity import unique_in_order
from events import summarize
from graph_walk import bfs_distances, dfs_reachable
from dependency_order import dependency_order
from topk import top_k_counts
from time_search import lower_bound, interval_indices
from rate_window import SlidingWindow


class ToolkitTests(unittest.TestCase):
    def test_ordered_deduplication(self):
        self.assertEqual(unique_in_order(["b", "a", "b"]), ["b", "a"])
        self.assertEqual(unique_in_order([]), [])

    def test_deliveries_and_distinct_events(self):
        rows = [{"id": "1", "kind": "timeout"}] * 2
        self.assertEqual(summarize(rows), ({"timeout": 2}, {"timeout": 1}, 1))
        self.assertEqual(summarize([]), ({}, {}, 0))
        with self.assertRaises(ValueError):
            summarize([{"id": "", "kind": "timeout"}])
        with self.assertRaises(KeyError):
            summarize([{"id": "1"}])

    def test_cycle_does_not_loop_forever(self):
        graph = {"a": ["b", "c"], "b": ["a", "c"], "c": [], "z": []}
        self.assertEqual(bfs_distances(graph, "a"), {"a": 0, "b": 1, "c": 1})
        self.assertEqual(dfs_reachable(graph, "a"), {"a", "b", "c"})
        with self.assertRaises(ValueError):
            bfs_distances({"a": ["missing"]}, "a")

    def test_precedence_property(self):
        graph = {"db": [], "api": ["db", "db"], "check": ["api"], "docs": []}
        order = dependency_order(graph)
        self.assertEqual(set(order), set(graph))
        position = {task: i for i, task in enumerate(order)}
        self.assertTrue(all(position[p] < position[t]
                            for t, parents in graph.items() for p in parents))
        self.assertEqual(dependency_order({}), [])
        for invalid in ({"a": ["a"]}, {"a": ["b"], "b": ["a"], "c": ["b"]},
                        {"a": ["missing"]}):
            with self.subTest(invalid=invalid), self.assertRaises(ValueError):
                dependency_order(invalid)

    def test_heap_against_full_sort(self):
        rng = Random(41)
        for size in range(20):
            counts = {f"event-{i}": rng.randrange(6) for i in range(size)}
            ranked = sorted(((count, name) for name, count in counts.items()), reverse=True)
            for k in (0, 1, 3, size + 2):
                expected = [(name, count) for count, name in ranked[:k]]
                self.assertEqual(top_k_counts(counts, k), expected)
        with self.assertRaises(ValueError):
            top_k_counts({"timeout": -1}, 1)

    def test_search_against_library_and_linear_filter(self):
        rng = Random(42)
        for size in range(20):
            values = sorted(rng.randrange(10) for _ in range(size))
            for target in range(-1, 12):
                self.assertEqual(lower_bound(values, target), bisect_left(values, target))
            for start, end in ((0, 5), (3, 3), (-1, 11), (7, 9)):
                left, right = interval_indices(values, start, end)
                self.assertEqual(values[left:right], [v for v in values if start <= v < end])
        with self.assertRaises(ValueError):
            interval_indices([], 2, 1)

    def test_window_boundaries(self):
        limiter = SlidingWindow(2, 60)
        self.assertEqual([limiter.allow(t) for t in (0, 1, 2, 60, 60, 61)],
                         [True, True, False, True, False, True])
        self.assertLessEqual(len(limiter.accepted), 2)
        with self.assertRaises(ValueError):
            limiter.allow(59)
        with self.assertRaises(ValueError):
            limiter.allow(float("nan"))

    def test_independent_replicas_have_independent_limits(self):
        replicas = [SlidingWindow(2, 60), SlidingWindow(2, 60)]
        accepted = sum(replica.allow(0) for replica in replicas for _ in range(2))
        self.assertEqual(accepted, 4)  # Not a shared global limit of two.


if __name__ == "__main__":
    unittest.main()

python3 -m unittest -v test_toolkit