khodo/product-fpol
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Setup

Browse Source
This commit is contained in:
khodowany
2026-10-04 21:15:21 -05:00
parent c6b4ba1b43
commit 2759337d16
33 changed files with 10510 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_all.py
+11
View File
@@ -0,0 +1,11 @@
"""One suite entry point for the shared Modulus/Conan check target."""
import pathlib
import subprocess
import sys
ROOT = pathlib.Path(__file__).resolve().parent
status = 0
for script, executable in (("test_core.py", sys.argv[1]), ("test_server.py", sys.argv[2])):
result = subprocess.run([sys.executable, str(ROOT / script), executable], check=False)
status = status or result.returncode
sys.exit(status)
tests/test_core.py
+241
View File
@@ -0,0 +1,241 @@
"""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], 1)
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()
tests/test_server.py
+27 -3
View File
@@ -18,7 +18,7 @@ def running_server(*arguments, stop_signal=signal.SIGTERM):
with tempfile.TemporaryDirectory(prefix="fpol_test_") as directory:
log_path = pathlib.Path(directory) / "server.log"
with log_path.open("w") as log:
process = subprocess.Popen([str(SERVER), "--port", "0", *map(str, arguments)], stdout=log, stderr=log)
process = subprocess.Popen([str(SERVER), "--port", "0", "--database", str(pathlib.Path(directory) / "league.sqlite"), *map(str, arguments)], stdout=log, stderr=log)
try:
deadline = time.monotonic() + 10
while time.monotonic() < deadline:
@@ -44,10 +44,10 @@ def running_server(*arguments, stop_signal=signal.SIGTERM):
if return_code != 0:
raise AssertionError(f"Server exited with {return_code}:\n{log_path.read_text()}")
def request(port, path, method="GET"):
def request(port, path, method="GET", body=None):
connection = http.client.HTTPConnection("127.0.0.1", port, timeout=3)
try:
connection.request(method, path)
connection.request(method, path, body=json.dumps(body) if body is not None else None, headers={"Content-Type": "application/json"})
response = connection.getresponse()
return response.status, response.getheader("Content-Type"), response.read()
finally:
@@ -66,6 +66,30 @@ class server_tests(unittest.TestCase):
self.assertEqual(status, 404)
self.assertEqual(json.loads(body)["code"], "not_found")
self.assertEqual(request(port, "/api/health", "POST")[0], 404)
def test_catalog_rules_and_validation_routes(self):
with running_server() as port:
self.assertEqual(json.loads(request(port, "/api/catalog")[2])["payload"], {"documents": []})
self.assertEqual(len(json.loads(request(port, "/api/rules")[2])["payload"]["slots"]), 12)
status, _, body = request(port, "/api/capacity", "POST", {"catalog_id": "missing", "players": 12})
self.assertEqual(status, 400)
self.assertEqual(json.loads(body)["code"], "invalid_argument")
self.assertEqual(request(port, "/api/score-preview", "POST", {})[0], 400)
self.assertEqual(request(port, "/api/capacity")[0], 404)
def test_seeded_catalog_capacity_and_scoring(self):
demo = pathlib.Path(__file__).resolve().parents[1] / "data" / "demo"
with tempfile.TemporaryDirectory(prefix="fpol_api_") as directory:
database = pathlib.Path(directory) / "league.sqlite"
result = subprocess.run([str(SERVER.with_name("fpol_admin")), "--database", str(database), "import", str(demo / "catalog.json"), str(demo / "ratings.json"), str(demo / "results-3.json")], capture_output=True, text=True, timeout=10)
self.assertEqual(result.returncode, 0, result.stderr)
with running_server("--database", database) as port:
catalog = json.loads(request(port, "/api/catalog")[2])["payload"]
self.assertEqual(len(catalog["documents"]), 3)
status, _, body = request(port, "/api/capacity", "POST", json.loads((demo / "capacity.json").read_text()))
self.assertEqual(status, 200)
self.assertTrue(json.loads(body)["payload"]["sufficient_candidates"])
status, _, body = request(port, "/api/score-preview", "POST", json.loads((demo / "preview.json").read_text()))
self.assertEqual(status, 200)
self.assertEqual(json.loads(body)["payload"]["total"], 528.24)
def test_static_assets_and_route_boundaries(self):
with tempfile.TemporaryDirectory(prefix="fpol_web_") as directory:
root = pathlib.Path(directory)