#!/usr/bin/env python3 """ Adversarial Empirical Challenge Script for Milestone 1: Progression Benchmarks. Stress-tests SQLite schema, constraints, idempotency, boundary handling, and in-memory lifecycle. """ from __future__ import annotations import os import sys import sqlite3 import tempfile from typing import Any, Dict, List, Tuple # Add workspace root to sys.path WORKSPACE_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) if WORKSPACE_ROOT not in sys.path: sys.path.insert(0, WORKSPACE_ROOT) from server.world.game_design_matrix_service import GameDesignMatrixService from server.world.game_design_matrix_types import ProgressionBenchmarkRow from server.world.game_design_matrix_schema import SQL_CREATE_TABLES, SQL_PRAGMA_SETUP class EmpiricalChallengeRunner: def __init__(self) -> None: self.passed: int = 0 self.failed: int = 0 self.findings: List[str] = [] def record_pass(self, test_name: str) -> None: self.passed += 1 print(f" [PASS] {test_name}") def record_fail(self, test_name: str, reason: str) -> None: self.failed += 1 msg = f" [FAIL] {test_name}: {reason}" self.findings.append(msg) print(msg) # ------------------------------------------------------------------------- # 1. Boundary Inputs & Rejection Testing # ------------------------------------------------------------------------- def test_boundary_inputs_query_rejection(self) -> None: print("\n--- 1. Boundary Inputs & Query Rejection ---") service = GameDesignMatrixService(":memory:") service.seed_canonical_data(force=True) # 1.1 Out-of-bounds levels query for invalid_lvl in [0, 101, -1, 999, -999, 1000]: res = service.get_level_progression_benchmark(invalid_lvl) if res is None: self.record_pass(f"Query level {invalid_lvl} safely rejected (returned None)") else: self.record_fail(f"Query level {invalid_lvl}", f"Expected None, got {res}") # 1.2 Non-integer / edge query inputs for edge_val in [1.5, "101", "invalid_level", None]: # type: ignore[list-item] try: res = service.get_level_progression_benchmark(edge_val) # type: ignore[arg-type] if res is None: self.record_pass(f"Query invalid type {edge_val!r} returned None safely") elif edge_val == "101": self.record_fail("Query string '101'", f"Should return None, got {res}") except Exception as e: self.record_pass(f"Query invalid type {edge_val!r} raised exception safely: {type(e).__name__}") # 1.3 Database level CHECK constraint (level BETWEEN 1 AND 100) with service._get_connection() as conn: cur = conn.cursor() for bad_lvl in [0, 101, -1, 999]: try: cur.execute(""" INSERT INTO progression_benchmarks ( level, target_exp, exp_to_next_level, cumulative_exp, player_base_hp, player_benchmark_dps, monster_base_hp, monster_base_dps, max_affix_tier_allowed, death_penalty_ratio, level_gap_safe_range, level_gap_penalty_exp, monster_benchmark_exp ) VALUES (?, 0, 100, 0, 100.0, 20.0, 80.0, 12.0, 15, 0.0, 5, 0.60, 25) """, (bad_lvl,)) self.record_fail(f"Insert level {bad_lvl} CHECK", "Failed to raise IntegrityError") except sqlite3.IntegrityError: self.record_pass(f"Insert level {bad_lvl} rejected by SQLite CHECK (level BETWEEN 1 AND 100)") service.close() # ------------------------------------------------------------------------- # 2. Idempotency & Repeated Seeder Runs # ------------------------------------------------------------------------- def test_seeder_idempotency_and_no_corruption(self) -> None: print("\n--- 2. Seeder Idempotency & Multiple Runs ---") service = GameDesignMatrixService(":memory:") # First seed service.seed_canonical_data(force=True) initial_snapshot: Dict[int, Tuple[Any, ...]] = {} with service._get_connection() as conn: cur = conn.cursor() cur.execute("SELECT * FROM progression_benchmarks ORDER BY level ASC") for row in cur.fetchall(): initial_snapshot[row["level"]] = tuple(row) if len(initial_snapshot) == 100: self.record_pass("Initial seed generated exactly 100 levels") else: self.record_fail("Initial seed count", f"Expected 100, got {len(initial_snapshot)}") # Run seeder 5 additional times with force=True for i in range(1, 6): service.seed_canonical_data(force=True) with service._get_connection() as conn: cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM progression_benchmarks") count = cur.fetchone()[0] if count != 100: self.record_fail(f"Repeated seed force=True (run {i})", f"Row count drifted to {count}") break # Check for zero drift/corruption across all rows mismatch_found = False with service._get_connection() as conn: cur = conn.cursor() cur.execute("SELECT * FROM progression_benchmarks ORDER BY level ASC") for row in cur.fetchall(): if tuple(row) != initial_snapshot.get(row["level"]): mismatch_found = True break if not mismatch_found: self.record_pass("5 repeated force=True seeds produced identical zero-drift rows") else: self.record_fail("Repeated seed corruption", "Row values drifted after repeated seed") # Test force=False idempotency guard res = service.seed_canonical_data(force=False) if res.get("status") == "already_seeded": self.record_pass("force=False properly short-circuits with 'already_seeded'") else: self.record_fail("force=False short-circuit", f"Expected already_seeded, got {res}") # Integrity report after all seeds report = service.validate_game_design_integrity() if report.is_valid and len(report.violations) == 0: self.record_pass("Matrix integrity remains 100% valid after 6 total seeds") else: self.record_fail("Post-seed validation", f"Violations found: {report.violations}") service.close() # ------------------------------------------------------------------------- # 3. Column Types & SQLite CHECK Constraints # ------------------------------------------------------------------------- def test_schema_constraints_and_column_checks(self) -> None: print("\n--- 3. Column Types & SQLite CHECK Constraints ---") service = GameDesignMatrixService(":memory:") service.seed_canonical_data(force=True) constraint_tests = [ ("negative exp_to_next_level (-1)", "exp_to_next_level", -1), ("negative cumulative_exp (-50)", "cumulative_exp", -50), ("negative target_exp (-10)", "target_exp", -10), ("death_penalty_ratio > 1.0 (1.25)", "death_penalty_ratio", 1.25), ("death_penalty_ratio < 0.0 (-0.05)", "death_penalty_ratio", -0.05), ("player_base_hp <= 0 (0.0)", "player_base_hp", 0.0), ("player_benchmark_dps <= 0 (-5.0)", "player_benchmark_dps", -5.0), ("monster_base_hp <= 0 (0.0)", "monster_base_hp", 0.0), ("monster_base_dps <= 0 (0.0)", "monster_base_dps", 0.0), ("max_affix_tier_allowed < 1 (0)", "max_affix_tier_allowed", 0), ("max_affix_tier_allowed > 15 (16)", "max_affix_tier_allowed", 16), ("negative level_gap_safe_range (-1)", "level_gap_safe_range", -1), ("non-positive level_gap_penalty_exp (0.0)", "level_gap_penalty_exp", 0.0), ("non-positive monster_benchmark_exp (0)", "monster_benchmark_exp", 0), ] with service._get_connection() as conn: cur = conn.cursor() for desc, col, bad_val in constraint_tests: try: cur.execute(f"UPDATE progression_benchmarks SET {col} = ? WHERE level = 50", (bad_val,)) self.record_fail(f"CHECK constraint: {desc}", "Failed to raise IntegrityError") except sqlite3.IntegrityError: self.record_pass(f"CHECK constraint enforced: {desc} rejected") # Test NOT NULL constraint on key columns for not_null_col in ["target_exp", "exp_to_next_level", "cumulative_exp", "player_base_hp"]: try: cur.execute(f"UPDATE progression_benchmarks SET {not_null_col} = NULL WHERE level = 50") self.record_fail(f"NOT NULL on {not_null_col}", "Failed to raise IntegrityError") except sqlite3.IntegrityError: self.record_pass(f"NOT NULL enforced: {not_null_col} cannot be NULL") # Test UNIQUE / Primary Key constraint on level try: cur.execute(""" INSERT INTO progression_benchmarks ( level, target_exp, exp_to_next_level, cumulative_exp, player_base_hp, player_benchmark_dps, monster_base_hp, monster_base_dps, max_affix_tier_allowed, death_penalty_ratio, level_gap_safe_range, level_gap_penalty_exp, monster_benchmark_exp ) VALUES (50, 0, 100, 0, 100.0, 20.0, 80.0, 12.0, 15, 0.0, 5, 0.60, 25) """) self.record_fail("PK UNIQUE on level", "Duplicate level 50 insert succeeded without error") except sqlite3.IntegrityError: self.record_pass("PK UNIQUE on level: duplicate insertion cleanly rejected") service.close() # ------------------------------------------------------------------------- # 4. In-Memory SQLite Support & Lifecycle # ------------------------------------------------------------------------- def test_in_memory_sqlite_lifecycle(self) -> None: print("\n--- 4. In-Memory SQLite Lifecycle & Clean Validation ---") # 4.1 Clean initialization without prior database mem_service = GameDesignMatrixService(db_path=":memory:") self.record_pass("GameDesignMatrixService(':memory:') initialized cleanly") # 4.2 Canonical seed on fresh memory instance seed_stats = mem_service.seed_canonical_data(force=True) if seed_stats.get("progression_benchmarks") == 100: self.record_pass("In-memory seeded 100 progression benchmarks") else: self.record_fail("In-memory seeding", f"Expected 100, got {seed_stats}") # 4.3 Validation passes 100% cleanly report = mem_service.validate_game_design_integrity() if report.is_valid and len(report.violations) == 0: self.record_pass("In-memory validate_game_design_integrity() reports is_valid=True with 0 violations") else: self.record_fail("In-memory validation", f"Violations: {report.violations}") # 4.4 All 13 columns mapped accurately in ProgressionBenchmarkRow DTO row_lvl1 = mem_service.get_level_progression_benchmark(1) row_lvl99 = mem_service.get_level_progression_benchmark(99) row_lvl100 = mem_service.get_level_progression_benchmark(100) if row_lvl1 and row_lvl99 and row_lvl100: dto_checks = [ row_lvl1.level == 1 and row_lvl1.target_exp == 0 and row_lvl1.death_penalty_ratio == 0.0, row_lvl99.level == 99 and row_lvl99.death_penalty_ratio == 0.25 and row_lvl99.exp_to_next_level > 0, row_lvl100.level == 100 and row_lvl100.exp_to_next_level == 0 and row_lvl100.death_penalty_ratio == 0.0, ] if all(dto_checks): self.record_pass("ProgressionBenchmarkRow DTO mappings verified for boundary levels (1, 99, 100)") else: self.record_fail("ProgressionBenchmarkRow DTO check", "DTO fields did not match expected values") else: self.record_fail("ProgressionBenchmarkRow fetch", "Failed to retrieve boundary levels from memory DB") # 4.5 Clean close mem_service.close() self.record_pass("GameDesignMatrixService.close() terminated memory connection cleanly") # 4.6 Multi-instance memory isolation check svc1 = GameDesignMatrixService(db_path=":memory:") svc2 = GameDesignMatrixService(db_path=":memory:") svc1.seed_canonical_data(force=True) # svc2 should be unseeded with svc2._get_connection() as conn2: cur2 = conn2.cursor() cur2.execute("SELECT COUNT(*) FROM progression_benchmarks") count2 = cur2.fetchone()[0] if count2 == 0: self.record_pass("Multiple :memory: instances maintain strict memory isolation") else: self.record_fail("Memory isolation", f"svc2 saw {count2} rows from svc1") svc1.close() svc2.close() # ------------------------------------------------------------------------- # 5. Schema Auto-Migration Test # ------------------------------------------------------------------------- def test_schema_auto_migration_from_legacy(self) -> None: print("\n--- 5. Legacy Database Auto-Migration ---") with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: tmp_db_path = tmp.name try: # Create a mock legacy database with old 7-column schema conn = sqlite3.connect(tmp_db_path) conn.executescript(SQL_PRAGMA_SETUP) conn.execute(""" CREATE TABLE progression_benchmarks ( level INTEGER PRIMARY KEY CHECK (level BETWEEN 1 AND 100), target_exp INTEGER NOT NULL CHECK (target_exp >= 0), player_base_hp REAL NOT NULL CHECK (player_base_hp > 0), player_benchmark_dps REAL NOT NULL CHECK (player_benchmark_dps > 0), monster_base_hp REAL NOT NULL CHECK (monster_base_hp > 0), monster_base_dps REAL NOT NULL CHECK (monster_base_dps > 0), max_affix_tier_allowed INTEGER NOT NULL CHECK (max_affix_tier_allowed BETWEEN 1 AND 15) ) """) conn.commit() conn.close() # Initialize GameDesignMatrixService on the legacy database legacy_service = GameDesignMatrixService(db_path=tmp_db_path) with legacy_service._get_connection() as c: cur = c.cursor() cur.execute("PRAGMA table_info(progression_benchmarks)") cols = cur.fetchall() col_names = [col["name"] for col in cols] if len(col_names) == 13 and "exp_to_next_level" in col_names and "death_penalty_ratio" in col_names: self.record_pass("Legacy 7-column schema was automatically upgraded to 13 columns") else: self.record_fail("Legacy migration", f"Expected 13 columns, found {len(col_names)}: {col_names}") # Re-seed and validate upgraded schema legacy_service.seed_canonical_data(force=True) bm = legacy_service.get_level_progression_benchmark(99) if bm and bm.death_penalty_ratio == 0.25: self.record_pass("Upgraded legacy database seeded and queried successfully") else: self.record_fail("Upgraded DB query", f"Unexpected benchmark: {bm}") legacy_service.close() finally: if os.path.exists(tmp_db_path): os.remove(tmp_db_path) # ------------------------------------------------------------------------- # Main Execution # ------------------------------------------------------------------------- def run_all_challenges(self) -> bool: print("======================================================================") print(" ADVERSARIAL EMPIRICAL CHALLENGE HARNESS: PROGRESSION BENCHMARKS (M1)") print("======================================================================") self.test_boundary_inputs_query_rejection() self.test_seeder_idempotency_and_no_corruption() self.test_schema_constraints_and_column_checks() self.test_in_memory_sqlite_lifecycle() self.test_schema_auto_migration_from_legacy() print("\n======================================================================") print(f" CHALLENGE RESULTS SUMMARY: {self.passed} PASSED, {self.failed} FAILED") print("======================================================================") if self.failed > 0: print("FINDINGS / FAILURES:") for f in self.findings: print(f" - {f}") return False else: print("VERDICT: ALL ADVERSARIAL STRESS TESTS PASSED WITH 0 DEFECTS.") return True if __name__ == "__main__": runner = EmpiricalChallengeRunner() success = runner.run_all_challenges() sys.exit(0 if success else 1)