mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
59 lines
2.6 KiB
Python
59 lines
2.6 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()
|
|
|
|
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|