diff options
Diffstat (limited to 'tests/test_res.py')
| -rw-r--r-- | tests/test_res.py | 68 |
1 files changed, 64 insertions, 4 deletions
diff --git a/tests/test_res.py b/tests/test_res.py index e4dfb1c..5f9ac92 100644 --- a/tests/test_res.py +++ b/tests/test_res.py @@ -566,10 +566,6 @@ class Units(unittest.TestCase): "alias claude='systemd-run --user --scope --quiet --slice=agents.slice claude'") -if __name__ == "__main__": - unittest.main() - - class Throttle(unittest.TestCase): def test_psi(self): text = "some avg10=41.50 avg60=33.20 avg300=12.00 total=99\nfull avg10=30.00 avg60=25.00 avg300=9.00 total=88\n" @@ -806,3 +802,67 @@ class TaskFit(unittest.TestCase): self.assertEqual(R.batch_fit(4, 59, 30), 1) self.assertEqual(R.batch_fit(4, 29, 30), 0) self.assertEqual(R.batch_fit(4, 0, 30), 0) + + +class Coalesce(unittest.TestCase): + """Batch r-740/744/757: one queued gate per lock on the newest commit (ruling 2026-10-07: the queue coalesces).""" + # history a ← b ← c (c newest); x on a side branch of a + ANC = {("a", "b"), ("a", "c"), ("b", "c"), ("a", "x")} + + def anc(self, x, y): + return (x, y) in self.ANC + + def gate(self, id, commit, at, lock="gate@/p"): + e = queued(id, 10.0, at=at) + e.lock, e.commit = lock, commit + return e + + def test_descendant_supersedes_queued_ancestors(self): + g1, g2 = self.gate("r-740", "a", T(12)), self.gate("r-744", "b", T(13)) + run = running("r-730", title="gate z") + run.lock, run.commit = "gate@/p", "z" + led = R.Ledger(entries=[run, g1, g2]) + new = self.gate("r-757", "c", T(14)) + cov, gone = R.coalesce(led, new, self.anc) + self.assertEqual((cov, [e.id for e in gone]), (None, ["r-740", "r-744"])) # running r-730 untouched + led.entries.append(new) + R.supersede(new, gone, NOW) + self.assertEqual([(e.state, e.superseded_by, e.why) for e in (g1, g2)], + [("done", "r-757", "superseded by r-757")] * 2) + self.assertEqual((new.covers, new.queued), (["a", "b"], T(12))) # keeps the oldest turn + self.assertEqual([e.id for e in R._queue(led)], ["r-757"]) + self.assertEqual(run.state, "running") + + def test_ancestor_is_covered_not_queued(self): + g = self.gate("r-757", "c", T(14)) + led = R.Ledger(entries=[g]) + for c in ("b", "c"): + cov, gone = R.coalesce(led, self.gate("r-760", c, T(15)), self.anc) + self.assertIs(cov, g) + self.assertEqual(gone, []) + R.cover(g, "b") + R.cover(g, "c") # own commit not listed + self.assertEqual(g.covers, ["b"]) + + def test_unrelated_other_lock_or_no_commit_kept(self): + led = R.Ledger(entries=[self.gate("r-1", "x", T(12)), self.gate("r-2", "a", T(12), lock="gate@/q"), + self.gate("r-3", "", T(12))]) + self.assertEqual(R.coalesce(led, self.gate("r-9", "c", T(14)), self.anc), (None, [])) + self.assertEqual(R.coalesce(led, self.gate("r-9", "", T(14)), self.anc), (None, [])) + + def test_red_coalesced_gate_names_range(self): + e = running("r-757", title="gate c") + e.commit, e.covers = "c" * 40, ["a" * 40, "b" * 40] + e.state, e.rc, e.ended = "done", 1, T(14, 30) + line = R.done_line(e) + self.assertIn("covers aaaaaaa,bbbbbbb", line) + self.assertIn("aaaaaaa^..ccccccc", line) + self.assertIn("bisect", line) + e.rc = 0 + self.assertNotIn("bisect", R.done_line(e)) + e.covers = [] + self.assertNotIn("covers", R.done_line(e)) + + +if __name__ == "__main__": + unittest.main() |