"""Exercise the production C++ domain and Noto persistence without HTTP sockets.""" import copy import contextlib import json import pathlib import sqlite3 import subprocess import sys import tempfile import unittest ADMIN = pathlib.Path(sys.argv.pop(1)).resolve() DEMO = pathlib.Path(__file__).resolve().parents[1] / "data" / "demo" class core_tests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(prefix="fpol_core_") self.addCleanup(self.temporary.cleanup) self.root = pathlib.Path(self.temporary.name) self.database = self.root / "league.sqlite" self.counter = 0 self.catalog = json.loads((DEMO / "catalog.json").read_text()) self.ratings = json.loads((DEMO / "ratings.json").read_text()) self.results = json.loads((DEMO / "results-3.json").read_text()) self.request = json.loads((DEMO / "preview.json").read_text()) self.run_admin("import", DEMO / "catalog.json", DEMO / "ratings.json", *sorted(DEMO.glob("results-*.json"))) def run_admin(self, *arguments, succeeds=True): result = subprocess.run([str(ADMIN), "--database", str(self.database), *map(str, arguments)], capture_output=True, text=True, timeout=10) if succeeds: self.assertEqual(result.returncode, 0, result.stderr) return [json.loads(line) for line in result.stdout.splitlines()][-1] self.assertNotEqual(result.returncode, 0, result.stdout) return result.stderr def file(self, value): self.counter += 1 path = self.root / f"input-{self.counter}.json" path.write_text(json.dumps(value)) return path def preview(self, request=None): return self.run_admin("score-preview", self.file(request or self.request)) def scenario(self, votes, rating="toss_up", winner=None, status="reporting", total=None): race_id = "demo-house-01" ratings = copy.deepcopy(self.ratings) ratings["id"] = f"ratings-{self.counter}" ratings["entries"] = [{"race_id": race_id, "rating": rating}] results = copy.deepcopy(self.results) results["id"] = f"results-{self.counter}" results["entries"] = [{"race_id": race_id, "round": "general", "status": status, "total_votes": sum(votes) if total is None else total, "winner_id": f"{race_id}-{winner}" if winner else None, "votes": [{"candidate_id": f"{race_id}-{index + 1}", "votes": count} for index, count in enumerate(votes)]}] self.run_admin("import", self.file(ratings), self.file(results)) request = {"catalog_id": self.catalog["id"], "ratings_id": ratings["id"], "results_id": results["id"], "roster": [{"slot": "house:democrat", "candidate_id": f"{race_id}-1"}]} return self.preview(request)["scores"][0] def test_rules_and_twelve_slot_roster(self): rules = self.run_admin("rules") self.assertEqual(len(rules["slots"]), 12) result = self.preview() self.assertTrue(result["complete_roster"]) self.assertEqual(len(result["scores"]), 12) self.assertEqual(result["missing_scores"], 0) # Each office: 151.50 toss-up winner + 15 safe loser + 7.70 + 1.88. self.assertEqual(result["total"], 528.24) self.assertEqual(result["baseline_subtotal"], 373.5) def test_capacity_all_player_counts(self): for players in range(2, 13): result = self.run_admin("capacity", self.file({"catalog_id": self.catalog["id"], "players": players})) self.assertTrue(result["sufficient_candidates"]) self.assertFalse(result["candidate_supply_ready"]) for players in (0, 1, 13, 2.5, "12"): self.run_admin("capacity", self.file({"catalog_id": self.catalog["id"], "players": players}), succeeds=False) def test_party_and_flex_capacity_are_both_required(self): catalog = copy.deepcopy(self.catalog) catalog["id"] = "short-catalog" # Exactly 12 D, 12 R, 12 Other per office: every party passes, Flex cannot. catalog["candidates"] = [candidate for candidate in catalog["candidates"] if not candidate["id"].endswith("-4")] self.run_admin("import", self.file(catalog)) result = self.run_admin("capacity", self.file({"catalog_id": catalog["id"], "players": 12})) self.assertFalse(result["sufficient_candidates"]) for office in result["offices"]: for requirement in office["requirements"]: self.assertEqual(requirement["shortfall"], 12 if requirement["group"] == "total" else 0) def test_safe_tossup_winner_and_loser_examples(self): self.assertEqual(self.scenario([800, 200, 0, 0], "safe", 1, "called")["points"], 130) self.assertEqual(self.scenario([510, 490, 0, 0], winner=1, status="called")["points"], 151.5) self.assertEqual(self.scenario([490, 510, 0, 0], winner=2, status="called")["points"], 73.5) def test_lead_is_only_hypothetical_until_called(self): row = self.scenario([510, 490, 0, 0]) self.assertEqual(row["points"], 76.5) self.assertEqual(row["win_bonus"], 0) self.assertEqual(row["leader_preview_points"], 151.5) row = self.scenario([500, 500, 0, 0]) self.assertIsNone(row["leader_preview_points"]) self.assertFalse(row["leader_preview_available"]) def test_rounding_is_per_candidate_half_up(self): row = self.scenario([1, 31, 0, 0], "safe") self.assertEqual(row["vote_share"], 3.125) self.assertEqual(row["points"], 3.13) row = self.scenario([1, 2, 0, 0]) self.assertEqual(row["points"], 50) def test_unaccounted_votes_do_not_create_a_leader_preview(self): row = self.scenario([400, 100, 0, 0], total=1000) self.assertEqual(row["points"], 60) self.assertIsNone(row["leader_preview_points"]) self.assertFalse(row["leader_preview_available"]) def test_older_snapshot_import_keeps_observation_order(self): self.results["id"] = "late-import-of-old-results" self.results["as_of"] = "2026-11-04T01:30:00Z" self.run_admin("import", self.file(self.results)) snapshots = [document["id"] for document in self.run_admin("catalog")["documents"] if document["kind"] == "results"] self.assertEqual(snapshots, ["demo-results-0", "demo-results-1", self.results["id"], "demo-results-2", "demo-results-3", "demo-results-4"]) def test_more_own_votes_never_reduce_points(self): previous = -1 for votes in (1, 2, 49, 50, 51, 100, 999, 999999000): row = self.scenario([votes, 1000, 0, 0]) self.assertGreaterEqual(row["points"], previous) previous = row["points"] def test_missing_votes_and_ratings_are_unknown(self): self.request["results_id"] = "demo-results-0" result = self.preview() self.assertIsNone(result["total"]) self.assertEqual(result["missing_scores"], 12) self.assertTrue(all(row["points"] is None for row in result["scores"])) self.ratings["id"] = "no-ratings" self.ratings["entries"] = [] self.run_admin("import", self.file(self.ratings)) self.request["ratings_id"] = "no-ratings" self.assertEqual(self.preview()["missing_scores"], 12) def test_missing_candidate_votes_not_assumed_zero(self): self.results["id"] = "missing-candidate" self.results["entries"][3]["votes"] = self.results["entries"][3]["votes"][1:] self.run_admin("import", self.file(self.results)) self.request["results_id"] = self.results["id"] self.assertEqual(self.preview()["missing_scores"], 1) def test_partial_roster_has_no_complete_total(self): self.request["roster"] = self.request["roster"][:1] result = self.preview() self.assertTrue(result["valid"]) self.assertFalse(result["complete_roster"]) self.assertIsNone(result["total"]) self.assertEqual(result["known_subtotal"], 151.5) def test_duplicates_and_ineligible_slots_rejected(self): invalid = [ [self.request["roster"][0], self.request["roster"][0]], [{"slot": "house:flex", "candidate_id": "demo-house-04-1"}, self.request["roster"][0]], [{"slot": "senate:democrat", "candidate_id": "demo-house-04-1"}], [{"slot": "house:republican", "candidate_id": "demo-house-04-1"}], [{"slot": "house:other", "candidate_id": "unknown"}], ] for roster in invalid: request = dict(self.request, roster=roster) result = self.preview(request) self.assertFalse(result["valid"]) self.assertIsNone(result["total"]) def test_opposing_candidates_allowed_but_each_candidate_only_once(self): self.request["roster"] = [{"slot": "house:democrat", "candidate_id": "demo-house-04-1"}, {"slot": "house:republican", "candidate_id": "demo-house-04-2"}] self.assertTrue(self.preview()["valid"]) def test_corrections_are_reversible_and_history_survives_reopen(self): original = self.preview() self.request["results_id"] = "demo-results-4" corrected = self.preview() self.assertLess(corrected["total"], original["total"]) self.request["results_id"] = "demo-results-3" self.assertEqual(self.preview(), original) with contextlib.closing(sqlite3.connect(self.database)) as database: self.assertEqual(database.execute("SELECT COUNT(*) FROM documents").fetchone()[0], 7) self.assertEqual(database.execute("SELECT COUNT(*) FROM _noto_schema_migrations").fetchone()[0], 2) def test_idempotency_and_frozen_ratings(self): self.assertFalse(self.run_admin("import", DEMO / "ratings.json")["imported"]) self.ratings["entries"][0]["rating"] = "toss_up" self.assertIn("immutable", self.run_admin("import", self.file(self.ratings), succeeds=False)) # Object-key order does not change document identity. reordered = dict(reversed(list(self.catalog.items()))) self.assertFalse(self.run_admin("import", self.file(reordered))["imported"]) def test_rounds_never_combined_and_pending_round_unscored(self): self.results["id"] = "bad-round" self.results["entries"][0]["round"] = "runoff" self.assertIn("round", self.run_admin("import", self.file(self.results), succeeds=False)) row = self.scenario([600, 400, 0, 0], status="pending_round") self.assertIsNone(row["points"]) self.assertIn("resolution", row["reason"]) def test_invalid_result_imports_leave_no_partial_document(self): entries = [ {"total_votes": -1}, {"total_votes": 1000000001}, {"total_votes": 1}, {"total_votes": None}, {"status": "final"}, {"winner_id": "demo-senate-01-1"}, {"winner_id": None}, {"status": "reporting"}, {"votes": [{"candidate_id": "demo-house-01-1", "votes": -1}]}, {"votes": [{"candidate_id": "demo-house-01-1", "votes": 1.5}]}, {"votes": [{"candidate_id": "unknown", "votes": 1}]}, ] for changes in entries: result = copy.deepcopy(self.results) result["id"] = f"bad-result-{self.counter}" result["entries"][0].update(changes) self.run_admin("import", self.file(result), succeeds=False) self.assertEqual(len(self.run_admin("catalog")["documents"]), 7) def test_duplicate_races_votes_and_unknown_ratings_rejected(self): self.results["id"] = "duplicate-race" self.results["entries"].append(self.results["entries"][0]) self.run_admin("import", self.file(self.results), succeeds=False) self.results["entries"].pop() self.results["entries"][0]["votes"].append(self.results["entries"][0]["votes"][0]) self.assertIn("Duplicate", self.run_admin("import", self.file(self.results), succeeds=False)) self.ratings["id"] = "bad-rating" self.ratings["entries"][0]["rating"] = "certain" self.run_admin("import", self.file(self.ratings), succeeds=False) def test_fusion_lines_do_not_duplicate_candidates(self): self.catalog["id"] = "fusion" self.catalog["candidates"][0]["ballot_lines"].append("Another ballot party") self.run_admin("import", self.file(self.catalog)) capacity = self.run_admin("capacity", self.file({"catalog_id": "fusion", "players": 12})) self.assertEqual(capacity["offices"][0]["requirements"][0]["available"], 12) self.catalog["id"] = "duplicate-candidate" self.catalog["candidates"].append(self.catalog["candidates"][0]) self.run_admin("import", self.file(self.catalog), succeeds=False) def test_disqualified_candidates_and_ambiguous_provider_ids_rejected(self): self.catalog["id"] = "not-qualified" self.catalog["candidates"][0]["qualified"] = False self.run_admin("import", self.file(self.catalog), succeeds=False) self.catalog["candidates"][0]["qualified"] = True self.catalog["candidates"][1]["provider_ids"] = self.catalog["candidates"][0]["provider_ids"] self.assertIn("provider", self.run_admin("import", self.file(self.catalog), succeeds=False)) def test_demo_and_real_snapshots_cannot_mix(self): self.ratings["id"] = "mixed-mode" self.ratings["mode"] = "real" self.ratings["source"]["url"] = "https://example.org/ratings" self.assertIn("mismatch", self.run_admin("import", self.file(self.ratings), succeeds=False)) def test_timestamp_validation(self): for stamp in ("yesterday", "2026-02-30T00:00:00Z", "2026-10-01T24:00:00Z", "2026-10-01T00:00:00-05:00"): self.catalog["as_of"] = stamp self.run_admin("import", self.file(self.catalog), succeeds=False) def test_malformed_requests_and_unknown_snapshots(self): for request in ({}, dict(self.request, catalog_id="missing"), dict(self.request, results_id="missing"), dict(self.request, ratings_id="demo-results-1"), dict(self.request, roster="invalid")): self.run_admin("score-preview", self.file(request), succeeds=False) def test_missing_catalog_dependency(self): self.ratings["catalog_id"] = "not-imported" self.run_admin("import", self.file(self.ratings), succeeds=False) if __name__ == "__main__": unittest.main()