product-fpol/tests/test_application.py

"""Real C++ service/SQLite integration without requiring a listening socket."""
import concurrent.futures
import copy
import json
import pathlib
import sqlite3
import subprocess
import sys
import tempfile
import unittest

ADMIN = pathlib.Path(sys.argv.pop(1)).resolve()
INGEST = pathlib.Path(sys.argv.pop(1)).resolve()
ROOT = pathlib.Path(__file__).resolve().parents[1]
NOW = 2000000000
PASSWORD = "a sufficiently long password"

class application_tests(unittest.TestCase):
  @classmethod
  def setUpClass(cls):
    cls.base = tempfile.TemporaryDirectory(prefix="fpol_accounts_")
    cls.base_database = pathlib.Path(cls.base.name) / "base.sqlite"
    cls.identities = {}
    def command(*args, stdin=None):
      run = subprocess.run([str(ADMIN), "--database", str(cls.base_database), "--allow-demo", *args], input=stdin, text=True, capture_output=True, timeout=30)
      if run.returncode:
        raise RuntimeError(run.stderr)
      return json.loads(run.stdout.splitlines()[-1])
    command("import", str(ROOT / "data/demo/catalog.json"), str(ROOT / "data/demo/ratings.json"), str(ROOT / "data/real/us-2026-10-04-partial.json"))
    for username in ("alice", "bob", "carol"):
      command("create-user", username, stdin=PASSWORD + "\n")
      cls.identities[username] = command("app-request", "-", stdin=json.dumps({"action": "auth/login", "input": {"username": username, "password": PASSWORD}, "clock": NOW}))
  @classmethod
  def tearDownClass(cls):
    cls.base.cleanup()
  def setUp(self):
    self.temp = tempfile.TemporaryDirectory(prefix="fpol_application_")
    self.addCleanup(self.temp.cleanup)
    self.path = pathlib.Path(self.temp.name)
    self.database = self.path / "test.sqlite"
    with sqlite3.connect(self.base_database) as source, sqlite3.connect(self.database) as destination:
      source.backup(destination)
    self.identities = copy.deepcopy(type(self).identities)
    self.now = NOW
  def command(self, executable, *args, stdin=None):
    result = subprocess.run([str(executable), "--database", str(self.database), *map(str, args)], input=stdin, capture_output=True, text=True, timeout=30)
    self.assertEqual(result.returncode, 0, result.stderr)
    return json.loads(result.stdout.splitlines()[-1])
  def request(self, action, data=None, who="alice", status=200, csrf=True, raw=False):
    identity = self.identities.get(who, {})
    result = self.command(ADMIN, "--allow-demo", "app-request", "-", stdin=json.dumps({"action": action, "input": data or {}, "clock": self.now,
      "session": identity.get("session_token", ""), "csrf": identity.get("payload", {}).get("csrf", "") if csrf else ""}))
    self.assertEqual(result.get("status", 200), status, result)
    return result if raw or not result.get("ok") else result.get("payload", {})
  def league(self, players=("alice", "bob"), maximum=2, demo=True):
    data = {"name": "Friends", "team_name": "Alice team", "max_teams": maximum}
    if demo:
      data["catalog_id"] = "demo-2026"
    league = self.request("league/create", data)["league_id"]
    invite = self.request("league/invite", {"league_id": league})["invite"]
    for player in players[1:]:
      self.request("league/join", {"invite": invite, "team_name": player + " team"}, player)
    return league, invite
  def view(self, league, who="alice"):
    return self.request("league", {"league_id": league}, who)
  def start(self, players=("alice", "bob")):
    league, _ = self.league(players=players, maximum=len(players))
    self.request("draft/start", {"league_id": league})
    return league
  def turn(self, league):
    view = self.view(league)
    who = view["picker"]["username"]
    return who, self.view(league, who)
  def pick(self, league, candidate=None, request_id=None):
    who, view = self.turn(league)
    data = {"league_id": league, "candidate_id": candidate or view["legal_candidate_ids"][0], "pick_number": view["draft"]["pick_number"], "request_id": request_id or f"pick-{view['draft']['pick_number']}"}
    return who, data, self.request("draft/pick", data, who)
  def file(self, name, body):
    path = self.path / name
    path.write_text(json.dumps(body))
    return path
  def ingest(self, payload, configure=False):
    if configure:
      config = {"id": "fixture", "kind": "official-json", "url": "https://example.org/reviewed-test-fixture", "catalog_id": "demo-2026", "races": [{"race_id": "demo-house-01", "role": role} for role in ("counts", "outcome")]}
      self.command(INGEST, "configure", self.file("source.json", config))
    return self.command(INGEST, "replay", "fixture", self.file("payload.json", payload))
  def test_auth_and_csrf_and_logout(self):
    self.request("leagues", who="nobody", status=401)
    self.request("league/create", {}, csrf=False, status=403)
    self.request("auth/login", {"username": "alice", "password": "incorrect password"}, status=401)
    self.request("auth/logout")
    self.request("auth/me", status=401)
  def test_login_rate_limit(self):
    for _ in range(10):
      self.request("auth/login", {"username": "nonexistent", "password": PASSWORD}, status=401)
    self.request("auth/login", {"username": "nonexistent", "password": PASSWORD}, status=429)
    self.now += 901
    self.request("auth/login", {"username": "nonexistent", "password": PASSWORD}, status=401)
  def test_invited_registration_capacity_and_atomicity(self):
    league, invite = self.league(players=("alice",), maximum=2)
    credentials = {"username": "david", "password": PASSWORD, "team_name": "David", "invite": invite}
    registered = self.request("auth/register", credentials, who="nobody", raw=True)
    self.assertEqual(registered["payload"]["admin"], 0)
    self.assertEqual(len(self.view(league)["teams"]), 2)
    credentials["username"] = "edward"
    self.request("auth/register", credentials, who="nobody", status=409)
    with sqlite3.connect(self.database) as database:
      self.assertEqual(database.execute("SELECT COUNT(*) FROM users WHERE username='edward'").fetchone()[0], 0)
      self.assertNotIn(invite, str(database.execute("SELECT * FROM invites").fetchall()))
  def test_invite_expiry_and_revocation(self):
    league, invite = self.league(players=("alice",))
    self.request("league/revoke-invites", {"league_id": league})
    self.request("league/join", {"invite": invite, "team_name": "Bob"}, "bob", status=404)
    invite = self.request("league/invite", {"league_id": league})["invite"]
    self.now += 7 * 86400
    self.request("league/join", {"invite": invite, "team_name": "Bob"}, "bob", status=404)
  def test_isolation_and_commissioner_permissions(self):
    league, _ = self.league()
    self.request("league", {"league_id": league}, "carol", status=403)
    self.request("standings", {"league_id": league}, "carol", status=403)
    self.request("candidates", {"league_id": league}, "carol", status=403)
    self.request("draft/start", {"league_id": league}, "bob", status=403)
    self.request("league/invite", {"league_id": league}, "bob", status=403)
    self.assertEqual(self.request("leagues", who="carol"), [])
    self.request("draft/queue", {"league_id": league}, status=400)
  def test_reset_single_use_expiry_and_session_revocation(self):
    reset = self.command(ADMIN, "reset-link", "alice")
    # Operator reset timestamps use wall time, not the diagnostic clock.
    self.now = reset["expires_at"] - 1
    self.request("auth/reset", {"reset_token": reset["reset_token"], "password": PASSWORD + " new"}, who="nobody")
    self.now = NOW
    self.request("auth/me", status=401)
    self.request("auth/reset", {"reset_token": reset["reset_token"], "password": PASSWORD}, who="nobody", status=400)
    reset = self.command(ADMIN, "reset-link", "bob")
    self.now = reset["expires_at"]
    self.request("auth/reset", {"reset_token": reset["reset_token"], "password": PASSWORD}, who="nobody", status=400)
  def test_password_change_invalidates_old_session(self):
    result = self.request("auth/password", {"current_password": PASSWORD, "password": PASSWORD + " updated"}, raw=True)
    self.request("auth/me", status=401)
    self.identities["alice"] = result
    self.assertEqual(self.request("auth/me")["username"], "alice")
  def test_real_leagues_exist_before_data_ready(self):
    league, _ = self.league(demo=False)
    view = self.view(league)
    self.assertFalse(view["readiness"]["ready"])
    self.assertEqual(len(view["teams"]), 2)
    self.request("draft/start", {"league_id": league}, status=409)
    self.assertEqual(self.request("candidates", {"limit": 100})["total"], 115)
    self.assertEqual(len(self.request("candidates", {"offset": 100, "limit": 100})["items"]), 15)
  def test_private_queue_timeout_and_snake(self):
    league = self.start()
    who, view = self.turn(league)
    candidate = view["legal_candidate_ids"][-1]
    self.request("draft/queue", {"league_id": league, "candidates": [candidate]}, who)
    other = "alice" if who == "bob" else "bob"
    self.assertNotIn(candidate, json.dumps(self.view(league, other)["teams"]))
    self.now += 120
    self.request("tick")
    view = self.view(league)
    self.assertEqual(view["picks"][0]["candidate_id"], candidate)
    self.assertEqual(view["picks"][0]["automatic"], 1)
    second = view["picker"]["id"]
    self.pick(league)
    self.assertEqual(self.view(league)["picker"]["id"], second)
  def test_pause_restart_and_frozen_settings(self):
    league = self.start()
    self.now += 50
    self.request("draft/pause", {"league_id": league})
    self.assertEqual(self.view(league)["draft"]["remaining"], 70)
    self.now += 1000
    self.request("tick")
    self.assertEqual(len(self.view(league)["picks"]), 0)
    self.request("draft/resume", {"league_id": league})
    self.assertEqual(self.view(league)["draft"]["deadline"], self.now + 70)
    self.request("restart")
    view = self.view(league)
    self.assertEqual(view["league"]["status"], "paused")
    self.assertEqual(view["draft"]["remaining"], 120)
    self.assertEqual(json.loads(view["league"]["rules_json"])["version"], "v2-november-share")
    self.request("league/settings", {"league_id": league, "name": "Changed", "catalog_id": "demo-2026"}, status=409)
  def test_idempotent_concurrent_picks_and_timeout_boundary(self):
    league = self.start()
    who, view = self.turn(league)
    data = {"league_id": league, "candidate_id": view["legal_candidate_ids"][0], "pick_number": 0, "request_id": "same-request"}
    with concurrent.futures.ThreadPoolExecutor(2) as pool:
      results = list(pool.map(lambda _: self.request("draft/pick", data, who), range(2)))
    self.assertEqual(results[0], results[1])
    self.assertEqual(len(self.view(league)["picks"]), 1)
    data["candidate_id"] = view["legal_candidate_ids"][1]
    self.request("draft/pick", data, who, status=409)
    who, view = self.turn(league)
    self.now = view["draft"]["deadline"]
    self.request("draft/pick", {"league_id": league, "candidate_id": view["legal_candidate_ids"][0], "pick_number": 1, "request_id": "expired"}, who, status=409)
    self.request("tick")
    self.assertEqual(len(self.view(league)["picks"]), 2)
  def test_full_draft_capacity_and_league_exclusivity(self):
    league = self.start(players=("alice", "bob", "carol"))
    for _ in range(36):
      self.now += 120
      self.request("tick")
    view = self.view(league)
    self.assertEqual(view["league"]["status"], "drafted")
    self.assertEqual(len({p["candidate_id"] for p in view["picks"]}), 36)
    for team in view["teams"]:
      self.assertEqual(len({p["slot"] for p in view["picks"] if p["team_id"] == team["id"]}), 12)
    second = self.start()
    self.pick(second, view["picks"][0]["candidate_id"])
    self.assertEqual(len(self.view(second)["picks"]), 1)
  def test_ingest_corrections_runoff_retraction_and_outage(self):
    league = self.start()
    self.pick(league, "demo-house-01-1")
    counts = {"race_id": "demo-house-01", "kind": "counts", "round": "november_first", "source_time": "2026-11-04T01:00:00Z", "total_votes": 1000, "votes": [{"candidate_id": "demo-house-01-1", "votes": 400}]}
    outcome = {"race_id": "demo-house-01", "kind": "outcome", "round": "general", "source_time": "2026-11-04T01:00:00Z", "status": "pending_round", "winner_id": None}
    payload = {"schema_version": 1, "observations": [counts, outcome]}
    result = self.ingest(payload, configure=True)
    self.assertFalse(result["verified"])
    def score():
      return next(row["roster"][0]["points"] for row in self.request("standings", {"league_id": league})["teams"] if row["roster"])
    self.assertEqual(score(), 40)
    outcome.update(round="runoff", source_time="2026-12-06T01:00:00Z", status="called", winner_id="demo-house-01-1")
    self.ingest({"schema_version": 1, "observations": [outcome]})
    self.assertEqual(score(), 90)
    outcome.update(source_time="2026-12-06T02:00:00Z", status="reporting", winner_id=None)
    self.ingest({"schema_version": 1, "observations": [outcome]})
    self.assertEqual(score(), 40)
    counts.update(source_time="2026-11-04T03:00:00Z", votes=[{"candidate_id": "demo-house-01-1", "votes": 450}])
    self.ingest({"schema_version": 1, "observations": [counts]})
    self.assertEqual(score(), 45)
    counts["round"] = "runoff"
    self.assertTrue(self.ingest({"schema_version": 1, "observations": [counts]})["quarantined"])
    self.assertEqual(score(), 45)
    counts["round"] = "november_first"
    counts["source_time"] = "2026-11-04T00:00:00Z"
    counts["votes"][0]["votes"] = 100
    self.ingest({"schema_version": 1, "observations": [counts]})
    self.assertEqual(score(), 45)
  def test_twelve_teams_preserve_party_capacity_and_finish(self):
    league, invite = self.league(players=("alice", "bob", "carol"), maximum=12)
    for index in range(9):
      name = f"player_{index}"
      self.identities[name] = self.request("auth/register", {"username": name, "password": PASSWORD, "team_name": name, "invite": invite}, who="nobody", raw=True)
    self.request("draft/start", {"league_id": league})
    self.pick(league, "demo-house-01-1")
    # Once someone has their D slot, a second D would consume another team's only required supply.
    with sqlite3.connect(self.database) as database:
      first = database.execute("SELECT team_id FROM picks WHERE number=0 AND league_id=?", (league,)).fetchone()[0]
      database.execute("UPDATE drafts SET pick_number=23 WHERE league_id=?", (league,))
    who, view = self.turn(league)
    self.assertEqual(view["picker"]["id"], first)
    self.assertNotIn("demo-house-02-1", view["legal_candidate_ids"])
    with sqlite3.connect(self.database) as database:
      database.execute("UPDATE drafts SET pick_number=1 WHERE league_id=?", (league,))
    for _ in range(143):
      self.now += 120
      self.request("tick")
    view = self.view(league)
    self.assertEqual(view["league"]["status"], "drafted")
    self.assertEqual(len(view["picks"]), 144)
    self.assertEqual(len({row["candidate_id"] for row in view["picks"]}), 144)
  def test_openamerica_mapping_and_fusion_lines(self):
    config = {"id": "openamerica-test", "kind": "openamerica", "url": "https://openamerica.io/elections/results.json?year=2026&office=House", "office": "House", "catalog_id": "demo-2026", "counts_until": "2099-01-01T00:00:00Z", "races": [{"race_id": "demo-house-01", "role": "counts", "provider_id": "MI-01", "candidates": {"Test Candidate|D": "demo-house-01-1", "Test Candidate|W": "demo-house-01-1", "Write-in|": None}}]}
    self.command(INGEST, "configure", self.file("oa-config.json", config))
    # Synthetic transport fixture. It cannot verify a real feed.
    payload = {"api": 1, "year": 2026, "office": "House", "stale": False, "as_of": "2026-11-04T01:00:00+00:00", "races": {"MI-01": {"total": 100, "cands": [{"n": "Test Candidate", "p": "D", "v": 60}, {"n": "Test Candidate", "p": "W", "v": 10}]}}}
    result = self.command(INGEST, "replay", "openamerica-test", self.file("oa-payload.json", payload))
    self.assertEqual(result["changed"], 1)
    self.assertFalse(result["verified"])
    with sqlite3.connect(self.database) as database:
      normalized = json.loads(database.execute("SELECT body FROM observations").fetchone()[0])
      self.assertEqual(normalized["votes"], [{"candidate_id": "demo-house-01-1", "votes": 70}])
    payload["races"]["MI-01"]["cands"].append({"n": "Unknown", "p": "R", "v": 20})
    self.assertTrue(self.command(INGEST, "replay", "openamerica-test", self.file("oa-payload.json", payload))["quarantined"])
    activation = subprocess.run([str(INGEST), "--database", str(self.database), "activate", "openamerica-test"], capture_output=True, text=True)
    self.assertNotEqual(activation.returncode, 0)
  def test_ratings_import_permission_immutability_and_freeze(self):
    league, _ = self.league()
    ratings = json.loads((ROOT / "data/demo/ratings.json").read_text())
    ratings["id"] = "commissioner-ratings"
    data = {"league_id": league, "document": ratings}
    self.request("league/ratings", data, "bob", status=403)
    self.request("league/ratings", data)
    self.request("league/ratings", data)
    self.assertEqual(self.view(league)["league"]["ratings_id"], ratings["id"])
    ratings["entries"][0]["rating"] = "toss_up"
    self.request("league/ratings", data, status=409)
    self.request("draft/start", {"league_id": league})
    ratings["id"] = "too-late"
    self.request("league/ratings", data, status=409)
    self.request("league/name", {"league_id": league, "name": "Renamed after start"})
    self.assertEqual(self.view(league)["league"]["name"], "Renamed after start")
  def test_one_verified_race_cannot_activate_whole_feed(self):
    self.ingest({"schema_version": 1, "observations": []}, configure=True)
    # Fixture-only network evidence, to exercise the coverage gate without sockets.
    with sqlite3.connect(self.database) as database:
      database.execute("UPDATE sources SET verified=1,last_error=NULL WHERE id='fixture'")
      database.execute("INSERT INTO source_checks VALUES('fixture','demo-house-01','counts',?)", (self.now,))
    result = subprocess.run([str(INGEST), "--database", str(self.database), "activate", "fixture"], capture_output=True, text=True)
    self.assertNotEqual(result.returncode, 0)
    with sqlite3.connect(self.database) as database:
      self.assertEqual(database.execute("SELECT COUNT(*) FROM coverage").fetchone()[0], 0)
  def test_candidate_watch_deduplicates_and_detects_return_to_previous_payload(self):
    config = {"id": "watch", "kind": "candidate-watch", "url": "https://example.org/official-candidates", "catalog_id": "demo-2026", "races": []}
    self.command(INGEST, "configure", self.file("watch.json", config))
    first = self.file("first.json", {"version": 1})
    second = self.file("second.json", {"version": 2})
    self.command(INGEST, "replay", "watch", first)
    with sqlite3.connect(self.database) as database:
      database.execute("UPDATE sources SET last_changed=123,etag='external-etag' WHERE id='watch'")
    self.command(INGEST, "replay", "watch", first)
    with sqlite3.connect(self.database) as database:
      self.assertEqual(database.execute("SELECT last_changed FROM sources WHERE id='watch'").fetchone()[0], 123)
    self.command(INGEST, "replay", "watch", second)
    with sqlite3.connect(self.database) as database:
      database.execute("UPDATE sources SET last_changed=456 WHERE id='watch'")
    self.command(INGEST, "replay", "watch", first)
    with sqlite3.connect(self.database) as database:
      self.assertGreater(database.execute("SELECT last_changed FROM sources WHERE id='watch'").fetchone()[0], 456)
      self.assertEqual(database.execute("SELECT COUNT(*) FROM source_payloads").fetchone()[0], 2)
      self.assertEqual(database.execute("SELECT verified FROM sources WHERE id='watch'").fetchone()[0], 0)
  def test_catalog_documents_survive_migration_and_backup(self):
    backup = self.path / "backup.sqlite"
    self.command(ADMIN, "backup", backup)
    with sqlite3.connect(backup) as database:
      self.assertEqual(database.execute("PRAGMA integrity_check").fetchone()[0], "ok")
      self.assertEqual(database.execute("SELECT COUNT(*) FROM documents").fetchone()[0], 3)
      hashes = database.execute("SELECT password_hash FROM users").fetchall()
      self.assertEqual(len(set(row[0] for row in hashes)), 3)
      self.assertTrue(all(row[0].startswith("scrypt$131072$8$1$") for row in hashes))

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