verify_schema.py

"""Validate the reference DDL only; does not implement or certify the API."""
from pathlib import Path
import sqlite3
import tempfile
import threading
import unittest


SCHEMA = Path(__file__).with_name("schema.sql").read_text() + "\n" + Path(__file__).with_name("multiplayer-schema.sql").read_text()


def seed(conn):
    for player in ("a", "b"):
        conn.execute("INSERT INTO users(id,kind,display_name,created_at,updated_at,last_active_at) VALUES (?, 'guest', ?, 1, 1, 1)", (player, player))
        conn.execute("INSERT INTO devices VALUES (?, ?, ?, 'Web', 'web', 1, 1)", (f"dev-{player}", player, f"install-{player}"))
    conn.execute("INSERT INTO rule_revisions VALUES ('r1', '{}', ?, 1, 1)", ("1" * 64,))
    conn.execute("INSERT INTO content_revisions VALUES ('classic/easy', 'c1', 'classic', '{}', ?)", ("2" * 64,))
    conn.execute("INSERT INTO game_runs VALUES ('a', 'run-a', 'run-a', 'classic', 'classic/easy', 'c1', 'r1', '{}', 1)")
    conn.commit()


def add_reward(conn, row_id, owner="a", run="run-a", key="run/run-a/terminal"):
    conn.execute("INSERT INTO reward_ledger VALUES (?, ?, ?, ?, 'run', 5000, 'r1', 'client_reported', '{}', 1)", (row_id, owner, key, run))


class SchemaContract(unittest.TestCase):
    def setUp(self):
        self.db = sqlite3.connect(":memory:")
        self.db.executescript(SCHEMA)
        seed(self.db)

    def tearDown(self):
        self.db.close()

    def reject(self, sql, args=()):
        with self.assertRaises(sqlite3.IntegrityError):
            self.db.execute(sql, args)

    def test_schema_integrity_and_foreign_keys(self):
        self.assertEqual(self.db.execute("PRAGMA integrity_check").fetchone()[0], "ok")
        self.assertEqual(self.db.execute("PRAGMA foreign_keys").fetchone()[0], 1)
        self.assertEqual(self.db.execute("PRAGMA foreign_key_check").fetchall(), [])

    def test_account_unique_and_normalized(self):
        self.db.execute("INSERT INTO auth_identities VALUES ('i1','a','password','local','player_a',1)")
        self.reject("INSERT INTO auth_identities VALUES ('i2','b','password','local','player_a',1)")
        self.reject("INSERT INTO auth_identities VALUES ('i3','b','password','local','Player_B',1)")
        self.reject("INSERT INTO auth_identities VALUES ('i4','b','password','local','a!',1)")

    def test_password_cannot_attach_to_sso_identity(self):
        self.db.execute("INSERT INTO auth_identities VALUES ('sso','a','oidc','issuer','subject',1)")
        self.reject("INSERT INTO password_credentials VALUES ('sso','$argon2id$example',1)")

    def test_cross_player_device_session_rejected(self):
        self.reject("INSERT INTO sessions VALUES ('s','a','dev-b',?,?,1,1,1,2,3,NULL)", ("3"*64, "4"*64))

    def test_cross_player_reward_run_rejected(self):
        with self.assertRaises(sqlite3.IntegrityError):
            add_reward(self.db, "reward-wrong", owner="b")

    def test_personal_replay_has_no_cloud_table_or_upload_purpose(self):
        self.assertIsNone(self.db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='replays'").fetchone())
        self.reject("INSERT INTO blobs VALUES ('a','blob-a',?,'identity','replay',1,1,?,1,NULL)", ("5"*64, b"x"))
        self.reject("INSERT INTO blobs VALUES ('a','blob-a',?,'identity','evidence',1,1,?,1,NULL)", ("5"*64, b"x"))

    def test_cross_player_merge_session_rejected(self):
        self.db.execute("INSERT INTO sessions VALUES ('session-b','b','dev-b',?,?,1,1,1,2,3,NULL)", ("3"*64, "4"*64))
        self.reject("INSERT INTO merge_tickets VALUES ('ticket','a','session-b',?,10,NULL,NULL)", ("5"*64,))

    def test_terminal_unique_and_immutable(self):
        self.db.execute("INSERT INTO run_results VALUES ('a','run-a','won','{}',?,'client_reported',1)", ("6"*64,))
        self.reject("INSERT INTO run_results VALUES ('a','run-a','lost','{}',?,'client_reported',2)", ("7"*64,))
        self.reject("UPDATE run_results SET status='lost' WHERE user_id='a'")

    def test_award_unique_and_immutable(self):
        add_reward(self.db, "reward-1")
        with self.assertRaises(sqlite3.IntegrityError):
            add_reward(self.db, "reward-2")
        self.reject("UPDATE reward_ledger SET delta_units=10000")

    def test_invalid_json_and_negative_totals_rejected(self):
        self.reject("INSERT INTO profile_settings VALUES ('a','theme','not-json',1,1)")
        self.reject("INSERT INTO player_summary VALUES ('a',-1,0,0,'{}',1,1)")
        self.reject("INSERT INTO player_summary VALUES ('a',0,1,2,'{}',1,1)")

    def test_blob_length_enforced(self):
        self.reject("INSERT INTO blobs VALUES ('a','blob-a',?,'identity','legacy_import',2,2,?,1,NULL)", ("5"*64, b"x"))

    def test_reward_and_receipt_rollback_together(self):
        self.db.execute("BEGIN IMMEDIATE")
        try:
            add_reward(self.db, "reward-rollback")
            self.db.execute("INSERT INTO sync_operations VALUES ('a','op','dev-a',?,'invalid-json',1)", ("8"*64,))
        except sqlite3.IntegrityError:
            self.db.rollback()
        else:
            self.fail("invalid receipt must fail")
        self.assertEqual(self.db.execute("SELECT count(*) FROM reward_ledger").fetchone()[0], 0)

    def test_cas_and_tombstone_shape(self):
        self.db.execute("INSERT INTO save_slots VALUES ('a','slot','run-a','branch','dev-a',1,'{}',1,1,NULL)")
        update = "UPDATE save_slots SET snapshot_json=?,version=version+1 WHERE user_id='a' AND slot_id='slot' AND version=? AND deleted_at IS NULL"
        self.assertEqual(self.db.execute(update, ('{"elapsed":1}', 1)).rowcount, 1)
        self.assertEqual(self.db.execute(update, ('{"elapsed":2}', 1)).rowcount, 0)
        self.reject("UPDATE save_slots SET deleted_at=2 WHERE slot_id='slot'")
        self.db.execute("UPDATE save_slots SET deleted_at=2,snapshot_json=NULL,version=version+1 WHERE slot_id='slot'")
        self.assertEqual(self.db.execute(update, ('{}', 3)).rowcount, 0)

    def test_seq_not_reused_after_cleanup(self):
        statement = "INSERT INTO sync_changes(user_id,entity_type,entity_id,entity_version,change_json,created_at) VALUES ('a','summary','self',1,'{}',1)"
        first = self.db.execute(statement).lastrowid
        self.db.execute("DELETE FROM sync_changes")
        second = self.db.execute(statement).lastrowid
        self.assertGreater(second, first)

    def test_two_connections_cannot_duplicate_reward(self):
        with tempfile.TemporaryDirectory(prefix="minesweeper-schema-") as directory:
            path = str(Path(directory) / "test.sqlite")
            setup = sqlite3.connect(path)
            setup.executescript(SCHEMA)
            setup.execute("PRAGMA journal_mode=WAL")
            seed(setup)
            setup.close()
            start = threading.Barrier(2)
            results = []

            def attempt(row_id):
                conn = sqlite3.connect(path, timeout=5)
                conn.execute("PRAGMA foreign_keys=ON")
                try:
                    start.wait(timeout=5)
                    conn.execute("BEGIN IMMEDIATE")
                    add_reward(conn, row_id)
                    conn.commit()
                    results.append("accepted")
                except sqlite3.IntegrityError:
                    conn.rollback()
                    results.append("duplicate")
                except Exception as error:
                    results.append(type(error).__name__)
                finally:
                    conn.close()

            threads = [threading.Thread(target=attempt, args=(str(i),)) for i in range(2)]
            for thread in threads:
                thread.start()
            for thread in threads:
                thread.join(timeout=10)
                self.assertFalse(thread.is_alive())
            self.assertCountEqual(results, ["accepted", "duplicate"])
            check = sqlite3.connect(path)
            self.assertEqual(check.execute("SELECT count(*),sum(delta_units) FROM reward_ledger").fetchone(), (1, 5000))
            check.close()


class MultiplayerContract(unittest.TestCase):
    def setUp(self):
        self.db = sqlite3.connect(":memory:")
        self.db.executescript(SCHEMA)
        seed(self.db)
        self.db.execute("UPDATE users SET kind='registered'")
        # Deterministic UTC clock for SQL cutoff guards; does not test a real scheduler.
        self.clock_ms = 1500
        self.db.create_function("strftime", 2, lambda fmt, value: str(self.clock_ms // 1000))

    def tearDown(self):
        self.db.close()

    def reject(self, sql, args=()):
        with self.assertRaises(sqlite3.IntegrityError):
            self.db.execute(sql, args)

    def game(self, key, kind="daily_ranked", owner="a", tier="easy"):
        self.db.execute("INSERT INTO online_games(id,kind,status,difficulty,rules_revision,rules_snapshot_json,created_at) VALUES (?,?,'waiting',?,'r1','{}',1)", (key,kind,tier))
        self.db.execute("INSERT INTO online_seats(game_id,user_id,seat_no,control_device_id,connection_state,eligibility,remaining_safe_at_start,private_state_json) VALUES (?,?,1,?,'connected','active',10,'{}')", (key,owner,f"dev-{owner}"))

    def daily(self, key="daily-a", finished=1500, damage=0, eligible=1, tier="easy", confirmed=1600):
        self.game(key,tier=tier)
        self.db.execute("INSERT OR IGNORE INTO daily_boards VALUES ('2026-09-13',1000,2000,3000,'open',0,NULL,'r1')")
        self.db.execute("INSERT OR IGNORE INTO daily_account_state(user_id,challenge_date,version) VALUES ('a','2026-09-13',1)")
        self.db.execute("INSERT OR IGNORE INTO daily_challenges VALUES ('2026-09-13',?,'classic/easy','c1','r1')", (tier,))
        self.db.execute("INSERT INTO daily_attempts(game_id,user_id,challenge_date,tier,entry_kind,started_at,finished_at,finished_order,elapsed_ms,initial_health,damage_count,health_lost,status,verification,ranking_eligible,record_complete,confirmed_at) VALUES (?,'a','2026-09-13',?,'ranked',1100,?,1,400,3,?,?,'won','valid',?,1,?)", (key,tier,finished,damage,damage,eligible,confirmed))

    def segment(self, key, game):
        self.db.execute("INSERT INTO restricted_segments VALUES (?,?,1,3000,'gzip',?,1,3000,1,?,'valid','available',1600)",
                        (key,game,f"minesweeper:dev:v1:replay:{game}:body:{key}","1"*64))

    def submission(self, key="sub", state="awaiting_upload", owner="a"):
        self.db.execute("INSERT INTO daily_score_submissions VALUES (?,?,'2026-09-13','overall','e','m','x',1200,1500,1,0,?,?,1500,1900,NULL)", (key,owner,state,"a"*64))

    def three(self):
        for key,tier in (("e","easy"),("m","medium"),("x","expert")):
            self.daily(key,tier=tier)

    def grant(self, key="up", submission="sub", owner="a", game="e", expires=1900):
        self.db.execute("INSERT INTO daily_replay_uploads(id,submission_id,user_id,game_id,codec,compressed_sha256,content_sha256,compressed_bytes,uncompressed_bytes,redis_key,state,expires_at) VALUES (?,?,?,?,'gzip',?,?,1,1,?,'granted',?)",(key,submission,owner,game,"a"*64,"b"*64,f"minesweeper:dev:v1:replay:{game}:upload:{key}",expires))

    def ranking(self):
        self.three()
        self.db.execute("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',1,'publish',1600)")
        self.db.execute("INSERT INTO daily_rankings VALUES ('2026-09-13','overall',1,'a',1,1200,1500,1,'e','m','x')")
        self.db.execute("UPDATE daily_boards SET generation=1")

    def test_live_tables_game_kind_and_broadcast_scope_absent(self):
        self.assertEqual(self.db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'live_%'").fetchall(),[])
        with self.assertRaises(sqlite3.IntegrityError):
            self.game("unsupported-live","live_classic")
        self.reject("INSERT INTO account_restrictions VALUES ('a','broadcast',NULL,'removed','{}',1)")
        self.reject("INSERT INTO moderation_cases VALUES ('case','a','live','removed','report','open','removed',NULL,1,NULL)")

    def test_watch_requires_replay_and_has_no_live_target(self):
        columns={row[1] for row in self.db.execute("PRAGMA table_info(watch_sessions)")}
        self.assertNotIn("live_session_id",columns)
        self.reject("INSERT INTO watch_sessions(id,anonymous_id_hash,permission_epoch,state,created_at,expires_at) VALUES ('watch',?,1,'waiting',1,2)",("1"*64,))

    def test_no_pvp_tables_or_game_kind(self):
        self.assertEqual(self.db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'pvp_%'").fetchall(),[])
        with self.assertRaises(sqlite3.IntegrityError):
            self.game("unsupported","pvp")

    def test_one_formal_attempt_per_account(self):
        self.game("first")
        self.game("second")
        self.db.execute("INSERT INTO active_competitions VALUES ('a','first','daily_ranked',1)")
        self.reject("INSERT INTO active_competitions VALUES ('a','second','daily_ranked',1)")

    def test_guest_cannot_acquire_competitive_slot(self):
        self.game("first")
        self.db.execute("UPDATE users SET kind='guest' WHERE id='a'")
        self.reject("INSERT INTO active_competitions VALUES ('a','first','daily_ranked',1)")

    def test_second_seat_not_supported(self):
        self.game("daily")
        self.reject("INSERT INTO online_seats(game_id,user_id,seat_no,control_device_id,connection_state,eligibility,remaining_safe_at_start,private_state_json) VALUES ('daily','b',2,'dev-b','connected','active',10,'{}')")

    def test_damaged_attempt_cannot_be_best(self):
        self.daily(damage=1,eligible=0)
        self.reject("INSERT INTO daily_bests VALUES ('a','2026-09-13','easy','daily-a')")
        self.reject("UPDATE daily_attempts SET ranking_eligible=1 WHERE game_id='daily-a'")

    def test_finish_and_validation_both_must_precede_cutoff(self):
        self.daily("late-finish",finished=2000)
        self.reject("INSERT INTO daily_bests VALUES ('a','2026-09-13','easy','late-finish')")
        self.daily("late-verification",confirmed=2000)
        self.reject("INSERT INTO daily_bests VALUES ('a','2026-09-13','easy','late-verification')")

    def test_best_cannot_replace_with_damaged_attempt(self):
        self.daily("good")
        self.db.execute("INSERT INTO daily_bests VALUES ('a','2026-09-13','easy','good')")
        self.daily("bad",damage=1,eligible=0)
        self.reject("UPDATE daily_bests SET game_id='bad'")

    def test_today_view_cannot_reuse_same_tier(self):
        self.daily()
        self.reject("INSERT INTO daily_view_entitlements VALUES ('a','2026-09-13','daily-a','daily-a','daily-a',1600,NULL,1)")

    def test_results_append_only(self):
        self.game("g")
        self.reject("INSERT INTO online_results VALUES ('g',1,'invalid','a','incident','{}',2,2)")
        self.db.execute("INSERT INTO online_results VALUES ('g',1,'invalid',NULL,'incident','{}',2,2)")
        self.reject("UPDATE online_results SET kind='completion'")

    def test_command_sequence_cannot_repeat_under_new_id(self):
        self.game("g")
        self.db.execute("INSERT INTO online_commands VALUES ('g','a','c1',1,1,?,2,'{}')",("1"*64,))
        self.reject("INSERT INTO online_commands VALUES ('g','a','c2',1,1,?,2,'{}')",("1"*64,))

    def test_watch_cannot_use_another_accounts_session(self):
        self.daily()
        self.segment("segment","daily-a")
        self.db.execute("INSERT INTO sessions VALUES ('auth-b','b','dev-b',?,?,1,1,1,2,3,NULL)",("3"*64,"4"*64))
        self.reject("INSERT INTO watch_sessions(id,user_id,auth_session_id,replay_segment_id,permission_epoch,state,created_at,expires_at) VALUES ('watch','a','auth-b','segment',1,'waiting',1,2)")

    def test_published_puzzle_and_mapping_immutable(self):
        self.daily()
        self.db.execute("INSERT INTO daily_puzzles VALUES ('2026-09-13','easy','g1','v1','{}',?,1,'{}',1,2)",("1"*64,))
        self.reject("UPDATE daily_puzzles SET private_map_json='[]'")
        self.reject("UPDATE daily_challenges SET rules_revision='r1'")

    def test_upload_requires_candidate_and_correct_owner(self):
        self.three()
        self.submission(state="not_candidate")
        with self.assertRaises(sqlite3.IntegrityError):
            self.grant()
        self.db.execute("UPDATE daily_score_submissions SET status='awaiting_upload'")
        with self.assertRaises(sqlite3.IntegrityError):
            self.grant(owner="b")
        self.grant()

    def test_upload_must_belong_to_selected_attempts(self):
        self.three()
        self.daily("other")
        self.submission()
        with self.assertRaises(sqlite3.IntegrityError):
            self.grant(game="other")

    def test_upload_expiry_and_body_size(self):
        self.three()
        self.submission()
        with self.assertRaises(sqlite3.IntegrityError):
            self.grant(expires=2001)
        self.grant()
        self.db.execute("UPDATE daily_replay_uploads SET compressed_bytes=1048576,uncompressed_bytes=8388608")
        self.reject("UPDATE daily_replay_uploads SET compressed_bytes=1048577")
        self.reject("UPDATE daily_replay_uploads SET redis_key=''")
        self.reject("UPDATE daily_replay_uploads SET uncompressed_bytes=8388609")
        self.reject("UPDATE daily_replay_uploads SET state='valid',received_at=1600")
        self.clock_ms=2000
        with self.assertRaises(sqlite3.IntegrityError):
            self.grant(key="late",game="m")

    def test_pending_combination_unique(self):
        self.three()
        self.submission()
        with self.assertRaises(sqlite3.IntegrityError):
            self.submission(key="another")

    def test_candidate_does_not_create_ranking(self):
        self.three()
        self.submission()
        self.grant()
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_rankings").fetchone()[0],0)

    def test_rank_snapshot_cannot_be_rewritten(self):
        self.ranking()
        self.reject("UPDATE daily_rankings SET total_ms=1")
        self.reject("DELETE FROM daily_rankings")
        self.reject("UPDATE daily_rank_revisions SET created_at=1999")
        self.reject("DELETE FROM daily_rank_revisions")

    def test_published_generation_cannot_receive_extra_rows(self):
        self.ranking()
        self.db.execute("DELETE FROM daily_rank_annotations")
        self.reject("INSERT INTO daily_rankings VALUES ('2026-09-13','overall',1,'b',2,1200,1500,1,'e','m','x')")
        self.reject("UPDATE daily_attempts SET elapsed_ms=1 WHERE game_id='e'")
        self.reject("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',3,'skip',1700)")

    def test_deadline_enforced_even_if_freeze_worker_is_late(self):
        self.ranking()
        self.clock_ms=2000
        self.reject("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',2,'late',1900)")
        self.reject("UPDATE daily_boards SET ends_at=4000")
        self.reject("UPDATE daily_boards SET generation=2")
        self.reject("INSERT INTO daily_bests VALUES ('a','2026-09-13','easy','e')")
        self.reject("INSERT INTO daily_rankings VALUES ('2026-09-13','overall',1,'b',2,1200,1500,1,'e','m','x')")

    def test_freeze_is_permanent_but_annotation_allowed(self):
        self.ranking()
        self.reject("UPDATE daily_boards SET status='frozen',frozen_at=2000")
        self.clock_ms=2000
        self.db.execute("UPDATE daily_boards SET status='frozen',frozen_at=2000")
        self.reject("UPDATE daily_boards SET status='open',frozen_at=NULL")
        self.db.execute("INSERT INTO daily_rank_annotations VALUES ('note','2026-09-13',1,'overall',1,'disqualified','verified violation',2100)")
        self.assertEqual(self.db.execute("SELECT rank,total_ms FROM daily_rankings").fetchone(),(1,1200))

    def test_historical_replay_can_have_anonymous_watch_without_entitlement(self):
        # SQL allows anonymous playback; date, Top10 and expiry checks belong to API.
        self.daily()
        self.segment("segment","daily-a")
        self.db.execute("INSERT INTO watch_sessions(id,anonymous_id_hash,replay_segment_id,permission_epoch,state,created_at,expires_at) VALUES ('anon',?,'segment',1,'prepared',2100,2200)",("2"*64,))
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_view_entitlements").fetchone()[0],0)
        self.assertEqual(self.db.execute("SELECT count(*) FROM watch_exposures").fetchone()[0],0)

    def board_row(self, board, generation=1):
        ids = {"easy": ("e",None,None), "medium": (None,"m",None),
               "expert": (None,None,"x"), "overall": ("e","m","x")}[board]
        self.db.execute("INSERT INTO daily_rankings VALUES ('2026-09-13',?,?,'a',1,?,1500,1,?,?,?)",
                        (board,generation,1200 if board=="overall" else 400,*ids))

    def board_submission(self, board, key):
        ids = {"easy": ("e",None,None), "medium": (None,"m",None),
               "expert": (None,None,"x"), "overall": ("e","m","x")}[board]
        self.db.execute("INSERT INTO daily_score_submissions VALUES (?,'a','2026-09-13',?,?,?,?,?,1500,1,0,'awaiting_upload',?,1500,1900,NULL)",
                        (key,board,*ids,1200 if board=="overall" else 400,"a"*64))

    def four_rankings(self):
        self.three()
        self.db.execute("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',1,'four boards',1600)")
        for board in ("easy","medium","expert","overall"):
            self.board_row(board)
        self.db.execute("UPDATE daily_boards SET generation=1")

    def test_single_tier_candidate_needs_only_its_own_replay(self):
        self.daily("e")
        self.board_submission("easy","single")
        self.grant(submission="single")
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_attempts").fetchone()[0],1)
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_rankings").fetchone()[0],0)

    def test_one_pending_per_board_not_per_date(self):
        self.three()
        for board in ("easy","medium","expert","overall"):
            self.board_submission(board,board)
        with self.assertRaises(sqlite3.IntegrityError):
            self.board_submission("easy","duplicate")
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_score_submissions").fetchone()[0],4)

    def test_same_user_can_rank_first_on_all_four_boards(self):
        self.four_rankings()
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_rankings WHERE user_id='a' AND rank=1").fetchone()[0],4)

    def test_single_and_overall_attempt_shapes_are_distinct(self):
        self.three()
        self.db.execute("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',1,'draft',1600)")
        self.reject("INSERT INTO daily_rankings VALUES ('2026-09-13','overall',1,'a',1,400,1500,1,'e',NULL,NULL)")
        self.reject("INSERT INTO daily_rankings VALUES ('2026-09-13','easy',1,'a',1,1200,1500,1,'e','m','x')")
        self.reject("INSERT INTO daily_rankings VALUES ('2026-09-13','easy',1,'a',1,400,1500,1,NULL,'m',NULL)")
        self.board_submission("easy","single")
        self.reject("UPDATE daily_score_submissions SET board_kind='overall' WHERE id='single'")

    def test_one_replay_is_shared_between_single_and_overall(self):
        self.four_rankings()
        self.segment("seg","e")
        with self.assertRaises(sqlite3.IntegrityError):
            self.segment("copy","e")
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_rankings r JOIN restricted_segments s ON r.easy_game_id=s.game_id WHERE s.id='seg'").fetchone()[0],2)

    def test_failed_four_board_publication_preserves_previous_snapshot(self):
        self.four_rankings()
        self.db.commit()
        with self.assertRaises(sqlite3.IntegrityError):
            with self.db:
                self.db.execute("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',2,'update',1700)")
                self.board_row("easy",2)
                self.board_row("easy",2)  # duplicate in same board fails whole transaction
                self.db.execute("UPDATE daily_boards SET generation=2")
        self.assertEqual(self.db.execute("SELECT generation FROM daily_boards").fetchone()[0],1)
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_rankings WHERE generation=1").fetchone()[0],4)
        self.assertEqual(self.db.execute("SELECT count(*) FROM daily_rank_revisions WHERE generation=2").fetchone()[0],0)

    def test_freeze_and_annotations_do_not_mix_board_identities(self):
        self.four_rankings()
        self.clock_ms=2000
        self.db.execute("UPDATE daily_boards SET status='frozen',frozen_at=2000")
        self.db.execute("INSERT INTO daily_rank_annotations VALUES ('note','2026-09-13',1,'easy',1,'disqualified','violation',2100)")
        rows=self.db.execute("SELECT r.board_kind, a.id FROM daily_rankings r LEFT JOIN daily_rank_annotations a USING(challenge_date,generation,board_kind,rank) ORDER BY r.board_kind").fetchall()
        self.assertEqual(dict(rows),{"easy":"note","medium":None,"expert":None,"overall":None})
        # Service must add overall annotation when the same invalid attempt affects it.
        self.reject("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',2,'late',2100)")

    def test_overall_can_reference_attempts_from_different_devices(self):
        self.three()
        self.db.execute("INSERT INTO devices VALUES ('dev-a2','a','install-a2','Tablet','web',1,1)")
        self.db.execute("UPDATE online_seats SET control_device_id='dev-a2' WHERE game_id='m'")
        self.db.execute("UPDATE daily_attempts SET record_device_id='dev-a2',record_manifest_json='{}' WHERE game_id='m'")
        self.reject("UPDATE daily_attempts SET record_device_id='dev-b' WHERE game_id='m'")
        self.reject("UPDATE daily_attempts SET record_manifest_json='invalid' WHERE game_id='m'")
        self.submission()
        self.db.execute("INSERT INTO daily_rank_revisions VALUES ('2026-09-13',1,'multi device',1600)")
        self.board_row("overall")
        self.assertEqual(self.db.execute("PRAGMA foreign_key_check").fetchall(),[])

    def test_replay_tables_have_only_metadata_not_binary_payloads(self):
        for table in ("daily_replay_uploads","restricted_segments","restricted_chunks"):
            columns={row[1]:row[2] for row in self.db.execute(f"PRAGMA table_info({table})")}
            self.assertIn("redis_key",columns)
            self.assertNotIn("BLOB",columns.values())
            self.assertNotIn("compressed_content",columns)
            self.assertNotIn("projected_content",columns)

    def test_redis_record_metadata_expiry_and_chunk_key(self):
        self.daily()
        self.segment("segment","daily-a")
        self.db.execute("UPDATE restricted_segments SET compressed_bytes=1048576,uncompressed_bytes=8388608")
        self.reject("UPDATE restricted_segments SET compressed_bytes=1048577")
        self.reject("UPDATE restricted_segments SET uncompressed_bytes=8388609")
        self.reject("UPDATE restricted_segments SET storage_expires_at=4000")
        self.reject("UPDATE restricted_segments SET storage_expires_at=1600")
        self.db.execute("INSERT INTO restricted_chunks VALUES ('segment',0,0,400,'minesweeper:dev:v1:replay:chunk',20,2000,?)",("1"*64,))
        self.reject("INSERT INTO restricted_chunks VALUES ('segment',1,400,800,'minesweeper:dev:v1:replay:chunk',20,2000,?)",("1"*64,))
        self.reject("UPDATE restricted_chunks SET byte_length=0")

    def test_integrity_of_combined_database(self):
        self.ranking()
        self.submission()
        self.grant()
        self.assertEqual(self.db.execute("PRAGMA foreign_key_check").fetchall(),[])
        self.assertEqual(self.db.execute("PRAGMA integrity_check").fetchone()[0],"ok")


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