Something went wrong. Try again.
bayes for days
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439"""What `sds watch` claims, tested without the browser.
Three things here can be wrong without anybody noticing, and each one is worsethan a crash:
- **the tail** re-reading the file, or losing a record, or choking on the half written line it caught mid-flush. The first makes the viewer the reason a match is slow; the second and third make it wrong about a match nobody can replay.- **the reduction** disagreeing with `sds view` about the same log. Two pages that each decide for themselves what regret means is how a bad call and a bad menu get read as each other, which is the one thing the viewer exists for.- **the enemy layer** losing the half of the map the log only recently carried at all: where an enemy stood, the set the menu was priced against, and the worst member of that set. A frame that quietly drops these draws a board with no opponent on it and looks finished.- **the cut** dropping the chosen proposal. A map missing the hex the bot actually picked looks exactly like a map that is complete.
The markup is not tested. A page that renders the wrong number correctly iswhat these tests are for, and asserting on tags would not see it."""
import jsonimport tempfileimport unittestfrom pathlib import Path
from sds import viewerfrom sds.live import LogTail, Session, make_frame, open_session, stripfrom tests.test_viewer import WEIGHTS, decision, phi, proposal, record
def hexed(label, dealt, x, y, steps=0): """A proposal that ends somewhere, which is what the map needs.""" made = proposal(label, dealt=dealt) made["end"] = {"x": x, "y": y} made["action"] = {"kind": "move", "steps": [{}] * steps} return made
def written(directory: Path, tag: str, seat: str, records: list[dict]) -> Path: path = directory / f"{tag}-{seat}{viewer.DECISION_SUFFIX}" path.write_text("".join(json.dumps(r) + "\n" for r in records)) return path
class TestTail(unittest.TestCase): def test_a_half_written_line_is_held_back_and_completed_later(self): # The whole reason this reads bytes rather than lines. `SdsClient` # flushes after every record, so a reader routinely arrives inside one. with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "m-North.decisions.jsonl" first = json.dumps(record(1, 1)) second = json.dumps(record(2, 1)) path.write_text(first + "\n" + second[:20]) tail = LogTail(path, "North") self.assertEqual(len(tail.poll()), 1) self.assertEqual(tail.skipped, 0) with path.open("a") as handle: handle.write(second[20:] + "\n") got = tail.poll() self.assertEqual(len(got), 1) self.assertEqual(got[0]["seq"], 2) self.assertEqual(tail.skipped, 0)
def test_nothing_is_read_twice(self): with tempfile.TemporaryDirectory() as tmp: path = written(Path(tmp), "m", "North", [record(1, 1), record(2, 1)]) tail = LogTail(path, "North") self.assertEqual(len(tail.poll()), 2) self.assertEqual(tail.offset, path.stat().st_size) self.assertEqual(tail.poll(), []) self.assertEqual(tail.poll(), [])
def test_a_file_that_got_shorter_is_read_from_the_start(self): # A replay overwriting the run directory. Continuing from a stale # offset would read from the middle of a different match. with tempfile.TemporaryDirectory() as tmp: path = written(Path(tmp), "m", "North", [record(1, 1), record(2, 1)]) tail = LogTail(path, "North") tail.poll() written(Path(tmp), "m", "North", [record(9, 4)]) got = tail.poll() self.assertEqual([r["seq"] for r in got], [9])
def test_a_corrupt_line_is_counted_rather_than_raised(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "m-North.decisions.jsonl" path.write_text('{"not json\n' + json.dumps(record(2, 1)) + "\n") tail = LogTail(path, "North") self.assertEqual(len(tail.poll()), 1) self.assertEqual(tail.skipped, 1)
def test_a_missing_file_is_not_an_error(self): with tempfile.TemporaryDirectory() as tmp: self.assertEqual(LogTail(Path(tmp) / "gone.jsonl", "North").poll(), [])
class TestEnemy(unittest.TestCase): """The enemy half of a frame: ground truth, the priced set, the worst case.
All three are new fields, and every one of them reads as "no enemy" when it goes missing rather than as an error - which is the failure mode these assertions exist for. """
def frame(self, **kw): threats = [ { "id": 3, "from": {"x": 9, "y": 4}, "facing": 2, "done": False, "hexes": [{"x": 9, "y": 4}, {"x": 9, "y": 5}], "hexes_total": 7, } ] proposals = [hexed("close", 20.0, 6, 3, steps=2), hexed("hold", 5.0, 4, 4)] proposals[0]["feared"] = {"3": {"at": {"x": 9, "y": 5}, "take": 12.5}} made = decision(proposals=proposals, chose=0) made["threats"] = threats line = record(1, 2, rationale=[made], **kw) line["positions"] = [ { "id": 1, "name": "Ours", "friendly": True, "x": 6, "y": 3, "facing": 0, "done": True, "destroyed": False, }, { "id": 3, "name": "Theirs", "friendly": False, "x": 9, "y": 4, "facing": 2, "done": False, "destroyed": False, }, ] return make_frame(line, 0, None, set())
def test_the_priced_set_survives_with_the_size_it_was_cut_from(self): # `hexes_total` is what stops the page claiming a smaller envelope is # the whole one. It must never be inferred from `len(hexes)`. threat = self.frame()["forces"][0]["threats"][0] self.assertEqual(threat["from"], {"x": 9, "y": 4}) self.assertEqual(threat["facing"], 2) self.assertEqual(len(threat["hexes"]), 2) self.assertEqual(threat["hexes_total"], 7)
def test_the_worst_case_is_lifted_off_the_chosen_proposal(self): # Off the chosen one only: it is the order that was actually given, and # a worst case taken from a proposal nobody took prices nothing. unit = self.frame()["forces"][0]["units"][0] self.assertEqual(unit["feared"]["3"]["at"], {"x": 9, "y": 5}) self.assertAlmostEqual(unit["feared"]["3"]["take"], 12.5)
def test_ground_truth_travels_beside_the_belief(self): frame = self.frame() enemy = [u for u in frame["positions"] if not u["friendly"]] self.assertEqual([(u["id"], u["x"], u["y"]) for u in enemy], [(3, 9, 4)])
def test_a_log_without_any_of_it_still_makes_a_frame(self): # Every log written before these fields existed, and every replay of # one. An empty enemy layer, not an exception. line = record(1, 2, rationale=[decision(proposals=[hexed("hold", 1.0, 4, 4)])]) frame = make_frame(line, 0, None, set()) self.assertEqual(frame["positions"], []) self.assertEqual(frame["forces"][0]["threats"], []) self.assertEqual(frame["forces"][0]["units"][0]["feared"], {})
class TestFiring(unittest.TestCase): """The FIRING panel's half of a frame.
The reduction is `sds.firingaudit`'s and is tested there. What is tested here is that a firing row carries it at all, that nothing else does, and that a frame past the detail window drops it - a stale volley left on a trimmed frame would be read as this decision's. """
def plan(self): return { "x": 4, "y": 7, "facing": 2, "secondaryFacing": 2, "twist": 3, "arsenal": [ { "weapon": 3, "name": "PPC", "location": "RA", "rearMounted": False, "usable": True, "locationDestroyed": False, "longRange": 18, "bears": [9], "bearsTwisted": [9], } ], "offered": [ { "weapon": 3, "name": "PPC", "target": 9, "range": 5, "toHit": 7, "damage": 10.0, "heat": 10, "rung": 1, "alpha": 1, "declared": True, } ], "enemies": [{"unit": 9, "x": 6, "y": 8, "facing": 0, "range": 5}], }
def test_a_firing_row_carries_the_whole_arsenal_accounted_for(self): line = record(1, 2, phase="FIRING") line["firing"] = self.plan() firing = make_frame(line, 0, None, set())["firing"] self.assertEqual(firing["declared"], 1) self.assertEqual(firing["twist"], 3) self.assertEqual([w["status"] for w in firing["weapons"]], ["fired"])
def test_a_row_that_is_not_a_firing_row_has_no_volley(self): self.assertIsNone(make_frame(record(1, 2), 0, None, set())["firing"])
def test_a_trimmed_frame_drops_the_volley_with_the_rest_of_the_detail(self): line = record(1, 2, phase="FIRING") line["firing"] = self.plan() self.assertIsNone(strip(make_frame(line, 0, None, set()))["firing"])
class TestFrames(unittest.TestCase): def frame(self, proposals, chose=0, **kw): line = record(1, 2, rationale=[decision(proposals=proposals, chose=chose)], **kw) return make_frame(line, 0, None, set())
def test_a_proposal_carries_the_hex_it_ends_in(self): frame = self.frame([hexed("close", 20.0, 6, 3, steps=2), hexed("hold", 5.0, 4, 4)]) ends = [p["end"] for p in frame["forces"][0]["units"][0]["proposals"]] self.assertEqual(ends, [{"x": 6, "y": 3}, {"x": 4, "y": 4}])
def test_a_proposal_with_no_steps_is_where_the_unit_stands(self): # The only place the log says where a unit is. The map draws the move # as a move because of this, rather than as a bare destination. frame = self.frame([hexed("hold", 5.0, 4, 4), hexed("close", 20.0, 6, 3, steps=3)]) steps = [p["steps"] for p in frame["forces"][0]["units"][0]["proposals"]] self.assertEqual(steps, [0, 3])
def test_regret_is_the_viewer_s_regret(self): # Taking the worse of two: the "lance chose badly" case, and it must # come out the same number `sds view` prints for the same record. better = hexed("close", 20.0, 6, 3) worse = hexed("hold", 5.0, 4, 4) frame = self.frame([better, worse], chose=1) unit = frame["forces"][0]["units"][0] direct = viewer.unit_view( {"unit": 12, "role": "SNIPER", "proposals": [better, worse]}, {"unit": 12, "proposal": 1}, WEIGHTS, ) self.assertAlmostEqual(unit["regret"], round(direct["regret"], 4)) self.assertGreater(unit["regret"], 0.0)
def test_a_menu_of_nothing_shows_as_no_regret_and_no_value(self): flat = [hexed("hold", 0.0, 4, 4), hexed("also hold", 0.0, 4, 4)] unit = self.frame(flat)["forces"][0]["units"][0] self.assertEqual(unit["regret"], 0.0) self.assertLessEqual(unit["best_value"], 0.01)
def test_the_cut_keeps_the_chosen_proposal_and_says_how_many_it_dropped(self): many = [hexed(f"p{i}", float(i), i % 8, i // 8) for i in range(40)] line = record(1, 2, rationale=[decision(proposals=many, chose=0)]) # Proposal 0 is the worst of the forty, so a naive "best 5" drops it. unit = make_frame(line, 0, None, set(), cap=5)["forces"][0]["units"][0] self.assertEqual(unit["proposals_total"], 40) self.assertEqual(len(unit["proposals"]), 5) self.assertTrue(any(p["chosen"] for p in unit["proposals"]))
def test_the_turn_block_survives(self): line = record(1, 3) line["turn"] = { "chose": 2, "differed": True, "eligible": [2, 3], "replan": {"reason": "round 3 began"}, "replans_this_round": 1, "considered": [{"unit": 2, "label": "close", "value": 3.9}], "value": 3.9, } frame = make_frame(line, 0, None, set()) self.assertTrue(frame["turn"]["differed"]) self.assertEqual(frame["turn"]["replan"], "round 3 began") self.assertEqual(frame["turn"]["considered"][0]["label"], "close")
def test_only_the_first_sighting_of_a_round_s_plan_is_a_new_plan(self): # The bot plans once per force per round and repeats the rationale on # every reply. A page that redrew on each echo would hide the moment # the plan actually changed. seen = set() line = record(1, 2, rationale=[decision()]) self.assertTrue(make_frame(line, 0, None, seen)["plan_new"]) self.assertFalse(make_frame(line, 1, None, seen)["plan_new"])
class TestMenu(unittest.TestCase): def line_with_candidates(self): line = record(1, 2) line["candidates"] = [ {"label": "one gun", "phi": {"expected_damage": 0.2, "heat_incurred": 0.1}}, {"label": "alpha", "phi": {"expected_damage": 0.6, "heat_incurred": 0.5}}, ] for candidate in line["candidates"]: candidate["value"] = sum(WEIGHTS[n] * v for n, v in candidate["phi"].items()) line["chosen"] = 1 return line
def test_the_chosen_is_broken_out_against_the_runner_up(self): menu = make_frame(self.line_with_candidates(), 0, WEIGHTS, set())["menu"] self.assertEqual(menu["chosen_label"], "alpha") self.assertEqual(menu["runner_label"], "one gun") self.assertEqual(menu["mismatch"], 0.0) # Biggest mover first, and its sign is the sign of the difference. self.assertEqual(menu["rows"][0]["name"], "expected_damage") self.assertGreater(menu["rows"][0]["delta"], 0.0) self.assertLess(menu["rows"][1]["delta"], 0.0)
def test_the_wrong_weights_are_reported_rather_than_drawn_over(self): menu = make_frame(self.line_with_candidates(), 0, {"expected_damage": 1.0}, set())["menu"] self.assertGreater(menu["mismatch"], 0.0)
def test_no_weights_means_no_decomposition(self): self.assertIsNone(make_frame(self.line_with_candidates(), 0, None, set())["menu"])
class TestSession(unittest.TestCase): def session(self, tmp: Path, records: list[dict]) -> Session: written(tmp, "m", "North", records) return open_session(tmp)
def test_state_returns_only_what_is_new(self): with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) session = self.session(tmp, [record(1, 1), record(2, 1)]) first = session.state(0) self.assertEqual(first["total"], 2) self.assertEqual(len(first["frames"]), 2) self.assertIsNotNone(first["board"]) # The board is sent once. Re-sending it every second would be most # of the payload of a page that is meant to be cheap to poll. self.assertIsNone(session.state(2)["board"]) self.assertEqual(session.state(2)["frames"], [])
def test_appended_records_arrive_without_the_file_being_re_read(self): with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) session = self.session(tmp, [record(1, 1)]) session.state(0) before = session.tails[0].offset with session.tails[0].path.open("a") as handle: handle.write(json.dumps(record(2, 1)) + "\n") session._last_poll = 0.0 state = session.state(1) self.assertEqual(state["total"], 2) self.assertEqual([f["seq"] for f in state["frames"]], [2]) self.assertGreater(session.tails[0].offset, before)
def test_counters_are_the_viewer_s_counters(self): with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) rows = [record(1, 1), record(2, 1, outcome="defaulted"), record(3, 2, phase="FIRING")] session = self.session(tmp, rows) state = session.state(0) seat = state["seats"]["Davion"] self.assertEqual(seat["totals"], viewer.totals(viewer.phase_counters(rows))) self.assertEqual(seat["counters"]["MOVEMENT"]["defaulted"], 1)
def test_stance_changes_are_reported_from_the_plans_only(self): with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) rows = [ record(1, 1, rationale=[decision(dominant="hold")]), record(2, 1, rationale=[decision(dominant="hold")]), record(3, 2, rationale=[decision(dominant="close", because="honour broken")]), ] session = self.session(tmp, rows) events = session.state(0)["seats"]["Davion"]["events"] self.assertEqual([e["kind"] for e in events], ["opened", "latch"]) self.assertEqual(events[1]["to"], "close") self.assertEqual(events[1]["because"], "honour broken")
def test_old_frames_lose_their_detail_and_keep_their_headline(self): with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) written(tmp, "m", "North", [record(i, 1, rationale=[decision()]) for i in range(1, 8)]) session = open_session(tmp, keep=3) session.state(0) self.assertTrue(session.frames[0]["trimmed"]) self.assertEqual(session.frames[0]["forces"], []) self.assertEqual(session.frames[0]["seq"], 1) self.assertNotIn("trimmed", session.frames[-1]) self.assertTrue(session.frames[-1]["forces"])
def test_an_empty_directory_is_refused_with_a_reason(self): with tempfile.TemporaryDirectory() as raw, self.assertRaises(viewer.ViewerError): open_session(Path(raw))
def test_a_seat_is_read_off_the_filename_and_the_records_agree(self): with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) written(tmp, "m", "sds_North", [record(1, 1, seat="sds North")]) session = open_session(tmp) self.assertEqual(session.tails[0].seat, "sds_North") self.assertEqual(list(session.state(0)["seats"]), ["sds North"])
class TestPhi(unittest.TestCase): def test_feature_values_reach_the_frame_for_ranking(self): # Ranking a map by one feature is the whole point of the metric picker, # so `phi` has to survive the reduction with its names intact. made = proposal("close", features=phi(expected_damage=0.5, heat_incurred=0.25)) made["end"] = {"x": 1, "y": 1} frame = make_frame(record(1, 1, rationale=[decision(proposals=[made])]), 0, None, set()) got = frame["forces"][0]["units"][0]["proposals"][0]["phi"] self.assertEqual(got, {"expected_damage": 0.5, "heat_incurred": 0.25})
if __name__ == "__main__": unittest.main()