mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
84 lines
4.1 KiB
Python
84 lines
4.1 KiB
Python
"""Static migration checks; PostgreSQL integration runs in the backend CI."""
|
|
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
|
|
SQL = (Path(__file__).parent / "0001_initial.sql").read_text()
|
|
ASSIGNMENTS_SQL = (Path(__file__).parent / "0002_assignments.sql").read_text()
|
|
QUOTAS_SQL = (Path(__file__).parent / "0007_allocation_quotas.sql").read_text()
|
|
ARENAS_SQL = (Path(__file__).parent / "0008_match_arena_paths.sql").read_text()
|
|
ALLOCATION_ARENAS_SQL = (Path(__file__).parent / "0009_allocation_arena_paths.sql").read_text()
|
|
INITIAL_CONNECT_READY_SQL = (Path(__file__).parent / "0010_initial_connect_ready_at.sql").read_text()
|
|
|
|
|
|
class MigrationTest(unittest.TestCase):
|
|
def test_durable_domains_and_fences_exist(self):
|
|
required_tables = {
|
|
"identities", "sessions", "idempotency_keys", "queue_tickets", "proposals",
|
|
"proposal_participants", "matches", "match_participants",
|
|
"ratings", "seasons", "penalties", "result_receipts", "outbox", "audit_events",
|
|
}
|
|
for table in required_tables:
|
|
self.assertIn(f"CREATE TABLE {table}", SQL)
|
|
self.assertIn("queue_tickets_one_active_per_player", SQL)
|
|
self.assertIn("match_participants_one_active_match", SQL)
|
|
self.assertIn("UNIQUE (aggregate_type, aggregate_id, revision)", SQL)
|
|
self.assertIn("PRIMARY KEY (scope, idempotency_key)", SQL)
|
|
|
|
def test_redis_is_not_a_durable_dependency(self):
|
|
self.assertNotIn("CREATE TABLE redis", SQL.lower())
|
|
self.assertNotIn("redis_id", SQL.lower())
|
|
self.assertIn("CREATE TABLE outbox", SQL)
|
|
self.assertIn("published_at", SQL)
|
|
|
|
def test_no_unbounded_or_client_owned_identity_fields(self):
|
|
self.assertIn("steam_id TEXT NOT NULL UNIQUE", SQL)
|
|
self.assertIn("token_digest BYTEA NOT NULL UNIQUE", SQL)
|
|
self.assertIn("payload JSONB NOT NULL", SQL)
|
|
self.assertNotIn("steam_ticket TEXT", SQL)
|
|
self.assertIn("participation_active BOOLEAN NOT NULL DEFAULT TRUE", SQL)
|
|
self.assertIn("WHERE participation_active", SQL)
|
|
|
|
def test_seasons_are_ranked_only_and_penalties_are_durable(self):
|
|
self.assertIn("CHECK (playlist = 'ranked')", SQL)
|
|
self.assertIn("CREATE TABLE penalties", SQL)
|
|
self.assertIn("CREATE TABLE ranked_season_rollovers", SQL)
|
|
self.assertIn("PRIMARY KEY (player_id, season_id)", SQL)
|
|
self.assertIn("REFERENCES identities(player_id)", SQL)
|
|
self.assertIn("REFERENCES matches(match_id)", SQL)
|
|
|
|
def test_assignments_are_player_scoped_and_expiry_bound(self):
|
|
for fragment in (
|
|
"CREATE TABLE assignments", "PRIMARY KEY (match_id, player_id)",
|
|
"FOREIGN KEY (match_id, player_id)", "UNIQUE (match_id, slot)",
|
|
"join_authorisation TEXT NOT NULL", "expires_at TIMESTAMPTZ NOT NULL",
|
|
"assignments_player_expiry",
|
|
):
|
|
self.assertIn(fragment, ASSIGNMENTS_SQL)
|
|
|
|
def test_allocation_quotas_are_optional_and_region_bound(self):
|
|
for fragment in ("CREATE TABLE allocation_quotas", "region TEXT PRIMARY KEY", "window_seconds", "max_allocations"):
|
|
self.assertIn(fragment, QUOTAS_SQL)
|
|
self.assertIn("region IN ('EU', 'NA')", QUOTAS_SQL)
|
|
|
|
def test_ranked_arena_paths_are_database_enforced_for_new_rows(self):
|
|
for fragment in (
|
|
"proposals_ranked_arena_path", "matches_ranked_arena_path", "NOT VALID",
|
|
"match_arena_path IS NOT NULL", "arena_path IS NOT NULL",
|
|
"res://scenes/arena_01.tscn", "res://scenes/arena_02.tscn", "res://scenes/arena_03.tscn",
|
|
):
|
|
self.assertIn(fragment, ARENAS_SQL)
|
|
|
|
def test_provider_allocation_retains_arena_identity(self):
|
|
self.assertIn("ALTER TABLE allocations", ALLOCATION_ARENAS_SQL)
|
|
self.assertIn("ADD COLUMN arena_path TEXT", ALLOCATION_ARENAS_SQL)
|
|
|
|
def test_initial_connect_window_starts_at_assignment_readiness(self):
|
|
self.assertIn("ADD COLUMN initial_connect_ready_at TIMESTAMPTZ", INITIAL_CONNECT_READY_SQL)
|
|
self.assertIn("state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING')", INITIAL_CONNECT_READY_SQL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|