diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..28f52cf --- /dev/null +++ b/.coveragerc @@ -0,0 +1,21 @@ +[run] +source = src/commander_ai_lab +omit = + */tests/* + */__pycache__/* + */web/static/* + +[report] +exclude_lines = + pragma: no cover + def __repr__ + raise NotImplementedError + if TYPE_CHECKING: + if __name__ == .__main__.: + \.\.\. # ellipsis + +show_missing = true +precision = 1 + +[html] +directory = htmlcov diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..e34aff2 --- /dev/null +++ b/Makefile @@ -0,0 +1,65 @@ +# Commander AI Lab — Developer Makefile +# ====================================== +# Targets: +# make test — run all tests +# make test-unit — run unit tests only (fast) +# make test-integration— run integration tests +# make test-benchmark — run benchmark suite with -s (print timings) +# make coverage — run tests with coverage report +# make serve — start the FastAPI dev server +# make lint — ruff + mypy +# make fmt — ruff format +# make clean — remove build artefacts + +PYTHON ?= python +UVICORN ?= uvicorn +PORT ?= 8000 +SRC = src/commander_ai_lab + +.PHONY: test test-unit test-integration test-benchmark coverage serve lint fmt clean + +test: + $(PYTHON) -m pytest tests/ -q + +test-unit: + $(PYTHON) -m pytest tests/ -q -m "not integration and not benchmark" + +test-integration: + $(PYTHON) -m pytest tests/test_integration.py -v + +test-benchmark: + $(PYTHON) -m pytest tests/test_benchmark.py -v -s + +test-phase3: + $(PYTHON) -m pytest tests/test_phase3.py -v + +test-phase4: + $(PYTHON) -m pytest tests/test_phase4.py -v + +test-phase5: + $(PYTHON) -m pytest tests/test_phase5.py -v + +test-phase6: + $(PYTHON) -m pytest tests/test_phase6.py -v + +test-phase7: + $(PYTHON) -m pytest tests/test_integration.py tests/test_benchmark.py -v + +coverage: + $(PYTHON) -m pytest tests/ --cov=$(SRC) --cov-report=term-missing --cov-report=html + echo "Coverage report: htmlcov/index.html" + +serve: + $(UVICORN) commander_ai_lab.web.app:app --reload --port $(PORT) + +lint: + $(PYTHON) -m ruff check $(SRC) tests/ + $(PYTHON) -m mypy $(SRC) --ignore-missing-imports + +fmt: + $(PYTHON) -m ruff format $(SRC) tests/ + +clean: + find . -type d -name __pycache__ -exec rm -rf {} + 2>/dev/null || true + find . -type f -name '*.pyc' -delete + rm -rf .coverage htmlcov .mypy_cache .ruff_cache dist build diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..fe4b8c5 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,24 @@ +[pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* Bench* +python_functions = test_* + +# Asyncio mode for pytest-asyncio +asyncio_mode = auto + +# Markers +markers = + slow: marks tests as slow (deselect with -m "not slow") + integration: marks integration tests + benchmark: marks benchmark tests + unit: marks fast unit tests + +# Output +addopts = + --tb=short + -q + +# Coverage (used when running: pytest --cov) +# pip install pytest-cov +# pytest --cov=src/commander_ai_lab --cov-report=term-missing diff --git a/tests/conftest.py b/tests/conftest.py index 8427d4b..84a250b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,75 +1,175 @@ -"""Shared pytest fixtures for Commander AI Lab tests.""" +""" +Commander AI Lab — Shared Test Fixtures (Phase 7) +================================================== +Provides pytest fixtures reused across all phase test modules. +""" + +from __future__ import annotations + +import asyncio +import os +from dataclasses import dataclass, field +from typing import Optional + import pytest -from commander_ai_lab.sim.models import Card, Player, PlayerStats, SimState + +# Headless Pygame for GUI tests +os.environ.setdefault("SDL_VIDEODRIVER", "dummy") +os.environ.setdefault("SDL_AUDIODRIVER", "dummy") + + +# --------------------------------------------------------------------------- +# Stub domain objects (shared across all phases) +# --------------------------------------------------------------------------- + +@dataclass +class StubCard: + name: str + type: str = "creature" + pt: Optional[str] = "2/2" + cmc: int = 3 + tapped: bool = False + is_commander: bool = False + oracle: str = "" + + def to_dict(self) -> dict: + return { + "name": self.name, + "type": self.type, + "pt": self.pt, + "cmc": self.cmc, + "tapped": self.tapped, + "is_commander": self.is_commander, + "oracle": self.oracle, + } + + +@dataclass +class StubPlayer: + name: str + seat: int + life: int = 40 + eliminated: bool = False + hand: list = field(default_factory=list) + battlefield: list = field(default_factory=list) + graveyard: list = field(default_factory=list) + exile: list = field(default_factory=list) + command_zone: list = field(default_factory=list) + library: list = field(default_factory=list) + commander_damage: dict = field(default_factory=dict) + mana_pool: dict = field(default_factory=dict) + + def to_dict(self, private: bool = False) -> dict: + return { + "name": self.name, + "seat": self.seat, + "life": self.life, + "eliminated": self.eliminated, + "hand_count": len(self.hand), + "hand": [c.to_dict() if hasattr(c, 'to_dict') else c for c in self.hand] + if private else [], + "battlefield": [c.to_dict() if hasattr(c, 'to_dict') else c for c in self.battlefield], + "graveyard": [c.to_dict() if hasattr(c, 'to_dict') else c for c in self.graveyard], + "exile": [c.to_dict() if hasattr(c, 'to_dict') else c for c in self.exile], + "command_zone":[c.to_dict() if hasattr(c, 'to_dict') else c for c in self.command_zone], + } + + +class StubGameState: + def __init__( + self, + num_players: int = 4, + names: Optional[list[str]] = None, + ) -> None: + names = names or [f"P{i}" for i in range(num_players)] + self.players = [StubPlayer(name=n, seat=i) for i, n in enumerate(names)] + self.turn = 1 + self.current_phase = "main1" + self.active_player_seat = 0 + self.stack = [] + self.game_over = False + self.winner = None + self.commander_damage: dict[tuple, int] = {} + + def get_legal_moves(self, seat: int) -> list[dict]: + return [ + {"id": 1, "category": "pass_priority", "description": "Pass priority"}, + {"id": 2, "category": "play_land", "description": "Play a land"}, + {"id": 3, "category": "cast_spell", "description": "Cast a spell"}, + ] + + def apply_move(self, seat: int, move_id: int) -> None: + pass + + def advance_turn(self) -> None: + self.active_player_seat = (self.active_player_seat + 1) % len(self.players) + if self.active_player_seat == 0: + self.turn += 1 + + +# --------------------------------------------------------------------------- +# Pytest fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def game_state(): + return StubGameState() @pytest.fixture -def basic_creature(): - return Card(name="Grizzly Bears", type_line="Creature — Bear", cmc=2, pt="2/2", power="2", toughness="2") +def game_state_2p(): + return StubGameState(num_players=2, names=["Alice", "Bob"]) @pytest.fixture -def flying_creature(): - return Card( - name="Serra Angel", - type_line="Creature — Angel", - cmc=5, - pt="4/4", - power="4", - toughness="4", - oracle_text="Flying, vigilance", - keywords=["Flying", "Vigilance"], +def targeting_memory(): + from commander_ai_lab.sim.politics.memory import TargetingMemory + return TargetingMemory(num_players=4) + + +@pytest.fixture +def comms_channel(): + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + return PoliticsCommsChannel() + + +@pytest.fixture +def politics_engine(targeting_memory, comms_channel): + from commander_ai_lab.sim.politics.engine import PoliticsEngine + return PoliticsEngine( + num_players=4, + memory=targeting_memory, + comms=comms_channel, + personalities={0: "timmy", 1: "spike", 2: "johnny", 3: "aggressive"}, + threat_fn=lambda s: {i: 0.3 for i in range(4) if i != s}, ) @pytest.fixture -def basic_land(): - return Card(name="Forest", type_line="Basic Land — Forest", cmc=0) +def threat_assessor(): + from commander_ai_lab.sim.threat_assessor import ThreatAssessor + gs = StubGameState() + return ThreatAssessor(game_state=gs, num_players=4) @pytest.fixture -def sol_ring(): - return Card(name="Sol Ring", type_line="Artifact", cmc=1, oracle_text="{T}: Add {C}{C}", is_ramp=True) +def fresh_app(): + """FastAPI test app with a clean session store.""" + from commander_ai_lab.web.session_store import SessionStore + SessionStore._instance = None + from commander_ai_lab.web.app import create_app + return create_app() @pytest.fixture -def removal_spell(): - return Card(name="Murder", type_line="Instant", cmc=3, oracle_text="Destroy target creature.", is_removal=True) +def http_client(fresh_app): + from fastapi.testclient import TestClient + return TestClient(fresh_app) @pytest.fixture -def board_wipe(): - return Card(name="Wrath of God", type_line="Sorcery", cmc=4, oracle_text="Destroy all creatures.", is_board_wipe=True) - - -@pytest.fixture -def commander_card(): - return Card( - name="Korvold, Fae-Cursed King", - type_line="Legendary Creature — Dragon Noble", - cmc=5, - pt="4/4", - power="4", - toughness="4", - is_commander=True, - oracle_text="Flying, haste\nWhenever Korvold, Fae-Cursed King enters the battlefield or attacks, sacrifice another permanent.", - keywords=["Flying", "Haste"], - ) - - -@pytest.fixture -def sample_player(basic_creature, basic_land, sol_ring): - p = Player(name="Alice", life=40, owner_id=0) - p.library = [basic_creature.clone(), basic_land.clone(), sol_ring.clone()] * 20 - return p - - -@pytest.fixture -def two_player_state(sample_player): - import copy - p2 = copy.deepcopy(sample_player) - p2.name = "Bob" - p2.owner_id = 1 - state = SimState(players=[sample_player, p2]) - state.init_battlefields(2) - return state +def event_loop(): + """Provide a fresh event loop for async tests.""" + loop = asyncio.new_event_loop() + yield loop + loop.close() diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py new file mode 100644 index 0000000..2d59a0a --- /dev/null +++ b/tests/test_benchmark.py @@ -0,0 +1,301 @@ +""" +Commander AI Lab — AI Benchmarking Suite (Phase 7) +=================================================== +Measures AI decision throughput and quality metrics: + + BenchThreatAssessor — threat assessment calls per second + BenchPoliticsEngine — deal proposal + evaluation throughput + BenchTargetingMemory — record + query throughput at scale + BenchIntegrated — end-to-end simulated turns / second + +All tests use wall-clock timing via time.perf_counter(). +Thresholds are conservative to pass on CI hardware. + +Run with: pytest tests/test_benchmark.py -v -s +""" + +from __future__ import annotations + +import asyncio +import time +from dataclasses import dataclass, field +from typing import Optional + +import pytest + + +# --------------------------------------------------------------------------- +# Shared stubs +# --------------------------------------------------------------------------- + +@dataclass +class _Player: + name: str + seat: int + life: int = 40 + eliminated: bool = False + hand: list = field(default_factory=list) + battlefield: list = field(default_factory=list) + graveyard: list = field(default_factory=list) + exile: list = field(default_factory=list) + command_zone: list = field(default_factory=list) + library: list = field(default_factory=list) + + +class _GS: + def __init__(self, n=4): + self.players = [_Player(f"P{i}", i) for i in range(n)] + self.turn = 1 + self.current_phase = "main1" + self.active_player_seat = 0 + self.stack = [] + self.game_over = False + self.winner = None + + def get_legal_moves(self, seat): + return [{"id": 1, "category": "pass_priority", "description": "Pass"}] + + def apply_move(self, seat, move_id): + pass + + +def _run(coro): + return asyncio.get_event_loop().run_until_complete(coro) + + +# --------------------------------------------------------------------------- +# 1. ThreatAssessor throughput +# --------------------------------------------------------------------------- + +class BenchThreatAssessor: + ITERATIONS = 2000 + MIN_OPS_PER_SEC = 500 + + def test_assess_threats_throughput(self): + from commander_ai_lab.sim.threat_assessor import ThreatAssessor + gs = _GS() + ta = ThreatAssessor(game_state=gs, num_players=4) + + start = time.perf_counter() + for _ in range(self.ITERATIONS): + for seat in range(4): + ta.assess_threats(viewer_seat=seat) + elapsed = time.perf_counter() - start + + ops = self.ITERATIONS * 4 + ops_per_sec = ops / elapsed + print(f"\nThreatAssessor: {ops_per_sec:.0f} assess_threats/s ({elapsed*1000:.1f} ms total)") + assert ops_per_sec >= self.MIN_OPS_PER_SEC, ( + f"assess_threats too slow: {ops_per_sec:.0f} ops/s < {self.MIN_OPS_PER_SEC}" + ) + + def test_top_threat_throughput(self): + from commander_ai_lab.sim.threat_assessor import ThreatAssessor + gs = _GS() + ta = ThreatAssessor(game_state=gs, num_players=4) + + start = time.perf_counter() + for _ in range(self.ITERATIONS): + ta.top_threat(viewer_seat=0) + elapsed = time.perf_counter() - start + + ops_per_sec = self.ITERATIONS / elapsed + print(f"\nThreatAssessor.top_threat: {ops_per_sec:.0f} ops/s") + assert ops_per_sec >= self.MIN_OPS_PER_SEC + + +# --------------------------------------------------------------------------- +# 2. TargetingMemory throughput +# --------------------------------------------------------------------------- + +class BenchTargetingMemory: + RECORD_COUNT = 10_000 + MIN_REC_PER_S = 50_000 + MIN_QRY_PER_S = 5_000 + + def test_record_throughput(self): + from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType + mem = TargetingMemory(num_players=4) + + start = time.perf_counter() + for i in range(self.RECORD_COUNT): + actor = i % 4 + target = (i + 1) % 4 + mem.record(turn=i // 4, actor=actor, target=target, + action_type=ActionType.ATTACKED) + elapsed = time.perf_counter() - start + + rps = self.RECORD_COUNT / elapsed + print(f"\nTargetingMemory.record: {rps:.0f} records/s") + assert rps >= self.MIN_REC_PER_S + + def test_aggression_query_throughput(self): + from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType + mem = TargetingMemory(num_players=4) + for i in range(1000): + mem.record(i // 4, i % 4, (i+1) % 4, ActionType.ATTACKED) + + start = time.perf_counter() + for turn in range(self.RECORD_COUNT // 10): + mem.aggression_score(0, 1, current_turn=turn) + elapsed = time.perf_counter() - start + + qps = (self.RECORD_COUNT // 10) / elapsed + print(f"\nTargetingMemory.aggression_score: {qps:.0f} queries/s") + assert qps >= self.MIN_QRY_PER_S + + def test_summary_matrix_correctness(self): + from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType + mem = TargetingMemory(num_players=4) + mem.record(1, 0, 1, ActionType.ATTACKED) + mem.record(1, 2, 3, ActionType.REMOVED_PERMANENT) + summary = mem.summary(current_turn=2) + assert (0, 1) in summary + assert (2, 3) in summary + assert summary[(0, 1)] > 0 + assert summary[(2, 3)] > 0 + + +# --------------------------------------------------------------------------- +# 3. PoliticsEngine throughput +# --------------------------------------------------------------------------- + +class BenchPoliticsEngine: + PROPOSALS = 200 + MIN_PROPOSALS_PER_S = 100 + + def test_deal_proposal_throughput(self): + from commander_ai_lab.sim.politics.memory import TargetingMemory + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + from commander_ai_lab.sim.politics.engine import PoliticsEngine + from commander_ai_lab.sim.politics.deals import DealType + + mem = TargetingMemory(4) + ch = PoliticsCommsChannel() + pe = PoliticsEngine( + 4, mem, ch, + {0:"timmy",1:"spike",2:"johnny",3:"aggressive"}, + lambda s: {i:0.3 for i in range(4) if i!=s}, + ) + + async def run_proposals(): + start = time.perf_counter() + for i in range(self.PROPOSALS): + await pe.propose_deal( + proposer=i % 4, + target=(i + 1) % 4, + deal_type=DealType.NON_AGGRESSION, + current_turn=i, + ) + return time.perf_counter() - start + + elapsed = _run(run_proposals()) + pps = self.PROPOSALS / elapsed + print(f"\nPoliticsEngine.propose_deal: {pps:.0f} proposals/s") + assert pps >= self.MIN_PROPOSALS_PER_S + + def test_check_violation_throughput(self): + from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + from commander_ai_lab.sim.politics.engine import PoliticsEngine + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + + mem = TargetingMemory(4) + ch = PoliticsCommsChannel() + pe = PoliticsEngine(4, mem, ch, + {0:"timmy",1:"spike",2:"johnny",3:"aggressive"}, + lambda s: {i:0.3 for i in range(4) if i!=s}) + + # Pre-populate 50 active deals + async def setup(): + for i in range(50): + d = await pe.propose_deal(i%4, (i+1)%4, DealType.NON_AGGRESSION, current_turn=1) + d.response = DealResponse.ACCEPTED + _run(setup()) + + start = time.perf_counter() + for _ in range(1000): + pe.check_deal_violation(1, 0, ActionType.ATTACKED, current_turn=2) + elapsed = time.perf_counter() - start + + cps = 1000 / elapsed + print(f"\nPoliticsEngine.check_deal_violation: {cps:.0f} checks/s") + assert cps >= 500 + + +# --------------------------------------------------------------------------- +# 4. Integrated simulated-turn throughput +# --------------------------------------------------------------------------- + +class BenchIntegrated: + SIMULATED_TURNS = 50 + MIN_TURNS_PER_S = 10 + + def test_simulated_turn_throughput(self): + """ + Simulate N full turns: threat assessment + politics check + memory record. + Measures combined per-turn overhead of the AI subsystems. + """ + from commander_ai_lab.sim.threat_assessor import ThreatAssessor + from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + from commander_ai_lab.sim.politics.engine import PoliticsEngine + from commander_ai_lab.sim.politics.deals import DealType + + gs = _GS() + ta = ThreatAssessor(game_state=gs, num_players=4) + mem = TargetingMemory(4) + ch = PoliticsCommsChannel() + pe = PoliticsEngine( + 4, mem, ch, + {0:"timmy",1:"spike",2:"johnny",3:"aggressive"}, + lambda s: ta.assess_threats(viewer_seat=s), + ) + + async def simulate(): + start = time.perf_counter() + for turn in range(self.SIMULATED_TURNS): + active = turn % 4 + # Assess threats for all seats + for seat in range(4): + ta.assess_threats(viewer_seat=seat) + # Record a simulated attack + attacker = active + defender = (active + 1) % 4 + mem.record_attack(turn, attacker, defender) + mem.record_damage(turn, attacker, defender, amount=5) + # Politics: on_turn_start + await pe.on_turn_start( + active_seat=active, + current_turn=turn, + game_state=gs, + ) + return time.perf_counter() - start + + elapsed = _run(simulate()) + tps = self.SIMULATED_TURNS / elapsed + print(f"\nIntegrated simulated turns: {tps:.1f} turns/s ({elapsed*1000:.0f} ms total)") + assert tps >= self.MIN_TURNS_PER_S + + +# --------------------------------------------------------------------------- +# 5. Memory scalability (large event log) +# --------------------------------------------------------------------------- + +class BenchMemoryScale: + def test_aggression_score_scales_linearly(self): + """ + Verify O(n) query time doesn't blow up with 100k events. + """ + from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType + mem = TargetingMemory(4) + N = 100_000 + for i in range(N): + mem.record(i, i % 4, (i+1) % 4, ActionType.ATTACKED) + + start = time.perf_counter() + mem.aggression_score(0, 1, current_turn=N) + elapsed = time.perf_counter() - start + + print(f"\naggression_score with {N} events: {elapsed*1000:.2f} ms") + assert elapsed < 1.0, f"Too slow with {N} events: {elapsed:.3f}s" diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..2153887 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,408 @@ +""" +Commander AI Lab — Integration Tests (Phase 7) +=============================================== +End-to-end tests that wire multiple subsystems together: + - ThreatAssessor + PoliticsEngine + - PoliticsEngine + TargetingMemory + CommsChannel + - REST API + SessionStore + WebSocket + - Turn lifecycle: phase transitions, AI decision flow, elimination + +Run with: pytest tests/test_integration.py -v +""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import dataclass, field +from typing import Optional + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def run(coro): + return asyncio.get_event_loop().run_until_complete(coro) + + +@dataclass +class _Player: + name: str + seat: int + life: int = 40 + eliminated: bool = False + hand: list = field(default_factory=list) + battlefield: list = field(default_factory=list) + graveyard: list = field(default_factory=list) + exile: list = field(default_factory=list) + command_zone: list = field(default_factory=list) + library: list = field(default_factory=list) + + +class _GS: + def __init__(self, n=4): + self.players = [_Player(f"P{i}", i) for i in range(n)] + self.turn = 1 + self.current_phase = "main1" + self.active_player_seat = 0 + self.stack = [] + self.game_over = False + self.winner = None + + def get_legal_moves(self, seat): + return [{"id": 1, "category": "pass_priority", "description": "Pass"}] + + def apply_move(self, seat, move_id): + pass + + +# --------------------------------------------------------------------------- +# 1. ThreatAssessor + PoliticsEngine integration +# --------------------------------------------------------------------------- + +class TestThreatPoliticsIntegration: + """ + Verify that ThreatAssessor scores flow into PoliticsEngine correctly + and that the adjusted_threat_score incorporates spite and memory. + """ + + def _build(self): + from commander_ai_lab.sim.threat_assessor import ThreatAssessor + from commander_ai_lab.sim.politics.memory import TargetingMemory + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + from commander_ai_lab.sim.politics.engine import PoliticsEngine + + gs = _GS() + ta = ThreatAssessor(game_state=gs, num_players=4) + mem = TargetingMemory(num_players=4) + ch = PoliticsCommsChannel() + pe = PoliticsEngine( + num_players=4, + memory=mem, + comms=ch, + personalities={0:"timmy",1:"spike",2:"johnny",3:"aggressive"}, + threat_fn=lambda s: ta.assess_threats(viewer_seat=s), + ) + return gs, ta, mem, ch, pe + + def test_adjusted_threat_no_spite_equals_base(self): + gs, ta, mem, ch, pe = self._build() + base = ta.assess_threats(viewer_seat=0).get(1, 0.0) + adjusted = pe.adjusted_threat_score(viewer=0, target=1, current_turn=1) + # Without spite or memory events, adjusted ≈ base (may differ by mem term) + assert abs(adjusted - base) < 0.35 + + def test_spite_raises_adjusted_score(self): + gs, ta, mem, ch, pe = self._build() + before = pe.adjusted_threat_score(0, 1, 1) + pe._add_spite(wronged_seat=0, target_seat=1, current_turn=1, intensity=0.5) + after = pe.adjusted_threat_score(0, 1, 1) + assert after > before + + def test_memory_aggression_raises_adjusted_score(self): + from commander_ai_lab.sim.politics.memory import ActionType + gs, ta, mem, ch, pe = self._build() + before = pe.adjusted_threat_score(0, 2, 1) + for _ in range(5): + mem.record(1, 2, 0, ActionType.ATTACKED) + after = pe.adjusted_threat_score(0, 2, 1) + assert after > before + + def test_top_threat_matches_assessor(self): + gs, ta, mem, ch, pe = self._build() + # Give seat 3 a huge life lead to inflate threat + gs.players[3].life = 5 # low life — looks like they need a kill + threats = ta.assess_threats(viewer_seat=0) + # Just confirm assess_threats returns 3 keys for a 4-player game + assert len(threats) == 3 + + +# --------------------------------------------------------------------------- +# 2. Politics full cycle: propose → accept → violate → spite +# --------------------------------------------------------------------------- + +class TestPoliticsFullCycle: + def test_deal_cycle_end_to_end(self): + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + from commander_ai_lab.sim.politics.engine import PoliticsEngine + + mem = TargetingMemory(4) + ch = PoliticsCommsChannel() + pe = PoliticsEngine( + 4, mem, ch, + personalities={0:"combo",1:"control",2:"timmy",3:"spike"}, + threat_fn=lambda s: {i:0.4 for i in range(4) if i!=s}, + ) + + # Propose + deal = run(pe.propose_deal(0, 1, DealType.NON_AGGRESSION, current_turn=2)) + assert deal.deal_id in pe._deals + + # Force accept for determinism + deal.response = DealResponse.ACCEPTED + + # Violate: seat 1 attacks seat 0 + broken = pe.check_deal_violation( + actor=1, action_target=0, + action_type=ActionType.ATTACKED, + current_turn=3, + ) + assert len(broken) == 1 + assert broken[0].response == DealResponse.BROKEN + + # Spite applied + assert pe.get_spite_bias(0, 1, 3) > 0 + + # Memory recorded breach + score = mem.aggression_score(1, 0, current_turn=3) + assert score > 0 + + def test_comms_log_has_proposal_and_response(self): + from commander_ai_lab.sim.politics.deals import DealType + from commander_ai_lab.sim.politics.memory import TargetingMemory + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + from commander_ai_lab.sim.politics.engine import PoliticsEngine + + mem = TargetingMemory(4) + ch = PoliticsCommsChannel() + pe = PoliticsEngine(4, mem, ch, + {0:"timmy",1:"spike",2:"johnny",3:"aggressive"}, + lambda s: {i:0.3 for i in range(4) if i!=s}) + run(pe.propose_deal(0, 2, DealType.SPARE_THIS_TURN, current_turn=4)) + assert len(ch._log) >= 2 + speakers = {b.speaker for b in ch._log} + assert 0 in speakers # proposer spoke + assert 2 in speakers # responder spoke + + +# --------------------------------------------------------------------------- +# 3. REST API → SessionStore → State snapshot round-trip +# --------------------------------------------------------------------------- + +class TestAPISessionIntegration: + @pytest.fixture + def client(self): + from commander_ai_lab.web.session_store import SessionStore + SessionStore._instance = None + from commander_ai_lab.web.app import create_app + from fastapi.testclient import TestClient + return TestClient(create_app()) + + def _new_game(self, client, names=None): + names = names or ["Alice", "Timmy", "Spike", "Johnny"] + r = client.post("/api/v1/games", json={ + "player_names": names, + "human_seat": 0, + "ai_personality": ["aggressive", "control", "combo"], + }) + assert r.status_code == 201 + return r.json()["game_id"] + + def test_create_then_get_state(self, client): + gid = self._new_game(client) + snap = client.get(f"/api/v1/games/{gid}").json() + assert snap["game_id"] == gid + assert len(snap["players"]) == 4 + + def test_player_names_preserved(self, client): + gid = self._new_game(client, ["Alice","Bob","Carol","Dave"]) + snap = client.get(f"/api/v1/games/{gid}").json() + names = [p["name"] for p in snap["players"]] + assert names == ["Alice", "Bob", "Carol", "Dave"] + + def test_starting_life_40(self, client): + gid = self._new_game(client) + snap = client.get(f"/api/v1/games/{gid}").json() + for p in snap["players"]: + assert p["life"] == 40 + + def test_move_accepted(self, client): + gid = self._new_game(client) + moves = client.get(f"/api/v1/games/{gid}/moves?seat=0").json()["moves"] + r = client.post(f"/api/v1/games/{gid}/move", + json={"seat": 0, "move_id": moves[0]["id"]}) + assert r.json()["accepted"] is True + + def test_concede_marks_eliminated(self, client): + gid = self._new_game(client) + client.post(f"/api/v1/games/{gid}/concede", json={"seat": 3}) + snap = client.get(f"/api/v1/games/{gid}").json() + assert snap["players"][3]["eliminated"] is True + + def test_delete_then_404(self, client): + gid = self._new_game(client) + client.delete(f"/api/v1/games/{gid}") + r = client.get(f"/api/v1/games/{gid}") + assert r.status_code == 404 + + def test_multiple_games_isolated(self, client): + gid1 = self._new_game(client, ["A","B","C","D"]) + gid2 = self._new_game(client, ["W","X","Y","Z"]) + assert gid1 != gid2 + snap1 = client.get(f"/api/v1/games/{gid1}").json() + snap2 = client.get(f"/api/v1/games/{gid2}").json() + assert snap1["players"][0]["name"] == "A" + assert snap2["players"][0]["name"] == "W" + + +# --------------------------------------------------------------------------- +# 4. Turn-manager phase transition integration +# --------------------------------------------------------------------------- + +class TestTurnManagerIntegration: + def test_run_game_completes(self): + from commander_ai_lab.sim.turn_manager import CommanderTurnManager, TurnManagerConfig + + gs = _GS() + # Make all players except seat 0 eliminated so game ends immediately + for p in gs.players[1:]: + p.eliminated = True + + events = [] + + async def on_event(ev): + events.append(ev) + + async def on_thinking(seat, flag): + pass + + tm = CommanderTurnManager( + game_state=gs, + ai_opponents=[], + human_seats=set(), + on_event=on_event, + on_thinking=on_thinking, + config=TurnManagerConfig(ai_decision_delay=0.0, max_turns=2), + ) + asyncio.get_event_loop().run_until_complete(tm.run_game()) + # Game should have fired at least one event + assert len(events) >= 1 + + def test_phase_sequence_order(self): + from commander_ai_lab.sim.turn_manager import CommanderTurnManager, TurnManagerConfig + + gs = _GS() + for p in gs.players[1:]: + p.eliminated = True + + phases_seen = [] + + async def on_event(ev): + if hasattr(ev, 'phase') and ev.phase: + phases_seen.append(ev.phase) + + async def on_thinking(seat, flag): + pass + + tm = CommanderTurnManager( + game_state=gs, + ai_opponents=[], + human_seats=set(), + on_event=on_event, + on_thinking=on_thinking, + config=TurnManagerConfig(ai_decision_delay=0.0, max_turns=1), + ) + asyncio.get_event_loop().run_until_complete(tm.run_game()) + # If phases were emitted, they should start with untap + if phases_seen: + assert phases_seen[0] in ("untap", "upkeep", "draw", "main1") + + def test_elimination_ends_game(self): + from commander_ai_lab.sim.turn_manager import CommanderTurnManager, TurnManagerConfig + + gs = _GS() + for p in gs.players[1:]: + p.eliminated = True + p.life = 0 + + game_over_fired = [False] + + async def on_event(ev): + if getattr(ev, 'event_type', '') == 'game_over': + game_over_fired[0] = True + + async def on_thinking(seat, flag): + pass + + tm = CommanderTurnManager( + game_state=gs, + ai_opponents=[], + human_seats=set(), + on_event=on_event, + on_thinking=on_thinking, + config=TurnManagerConfig(ai_decision_delay=0.0, max_turns=3), + ) + asyncio.get_event_loop().run_until_complete(tm.run_game()) + assert game_over_fired[0] is True + + +# --------------------------------------------------------------------------- +# 5. WebSocket protocol integration (via TestClient) +# --------------------------------------------------------------------------- + +class TestWebSocketIntegration: + @pytest.fixture + def ws_client(self): + from commander_ai_lab.web.session_store import SessionStore + SessionStore._instance = None + from commander_ai_lab.web.app import create_app + from fastapi.testclient import TestClient + return TestClient(create_app()) + + def _create_game(self, client) -> str: + r = client.post("/api/v1/games", json={ + "player_names": ["H","AI1","AI2","AI3"], + "human_seat": 0, + "ai_personality": ["aggressive","control","combo"], + }) + return r.json()["game_id"] + + def test_ws_connect_receives_state(self, ws_client): + gid = self._create_game(ws_client) + with ws_client.websocket_connect(f"/ws/game/{gid}?seat=0") as ws: + data = json.loads(ws.receive_text()) + assert data["type"] == "state" + assert "players" in data["data"] + + def test_ws_ping_pong(self, ws_client): + gid = self._create_game(ws_client) + with ws_client.websocket_connect(f"/ws/game/{gid}?seat=0") as ws: + ws.receive_text() # consume initial state + ws.send_text(json.dumps({"type": "ping"})) + resp = json.loads(ws.receive_text()) + assert resp["type"] == "pong" + + def test_ws_invalid_json_returns_error(self, ws_client): + gid = self._create_game(ws_client) + with ws_client.websocket_connect(f"/ws/game/{gid}?seat=0") as ws: + ws.receive_text() + ws.send_text("not json{{{") + resp = json.loads(ws.receive_text()) + assert resp["type"] == "error" + + def test_ws_unknown_message_type_returns_error(self, ws_client): + gid = self._create_game(ws_client) + with ws_client.websocket_connect(f"/ws/game/{gid}?seat=0") as ws: + ws.receive_text() + ws.send_text(json.dumps({"type": "unknown_cmd"})) + resp = json.loads(ws.receive_text()) + assert resp["type"] == "error" + + def test_ws_move_accepted(self, ws_client): + gid = self._create_game(ws_client) + with ws_client.websocket_connect(f"/ws/game/{gid}?seat=0") as ws: + ws.receive_text() # initial state + ws.send_text(json.dumps({"type": "move", "seat": 0, "move_id": 1})) + resp = json.loads(ws.receive_text()) + # Server should broadcast updated state + assert resp["type"] in ("state", "error") + + def test_ws_invalid_game_id_closes(self, ws_client): + with pytest.raises(Exception): + with ws_client.websocket_connect("/ws/game/doesnotexist?seat=0") as ws: + ws.receive_text()