"""Tests for GameIntegrationStore — CRUD, validation, uniqueness.""" import pytest from ledgrab.storage.base_store import EntityNotFoundError from ledgrab.storage.game_integration import EventMapping, GameIntegrationConfig from ledgrab.storage.game_integration_store import GameIntegrationStore from ledgrab.utils import secret_box @pytest.fixture def store(tmp_db) -> GameIntegrationStore: return GameIntegrationStore(tmp_db) # --------------------------------------------------------------------------- # Dataclass model tests # --------------------------------------------------------------------------- class TestEventMapping: def test_round_trip(self): m = EventMapping( event_type="health", effect="pulse", color=[0, 255, 0], duration_ms=1000, intensity=0.8, priority=5, ) data = m.to_dict() restored = EventMapping.from_dict(data) assert restored.event_type == "health" assert restored.effect == "pulse" assert restored.color == [0, 255, 0] assert restored.duration_ms == 1000 assert restored.intensity == 0.8 assert restored.priority == 5 def test_defaults(self): m = EventMapping(event_type="kill") assert m.effect == "flash" assert m.color == [255, 0, 0] assert m.duration_ms == 500 assert m.intensity == 1.0 assert m.priority == 0 def test_from_dict_defaults(self): m = EventMapping.from_dict({"event_type": "death"}) assert m.effect == "flash" assert m.color == [255, 0, 0] class TestGameIntegrationConfig: def test_round_trip(self): config = GameIntegrationConfig.create_from_kwargs( name="CS2 Integration", adapter_type="cs2_gsi", enabled=True, adapter_config={"token": "secret123"}, event_mappings=[ EventMapping(event_type="health", effect="gradient", color=[255, 0, 0]), EventMapping(event_type="kill", effect="flash", color=[0, 255, 0]), ], description="Counter-Strike 2 game state integration", tags=["fps", "cs2"], ) data = config.to_dict() restored = GameIntegrationConfig.from_dict(data) assert restored.id == config.id assert restored.name == "CS2 Integration" assert restored.adapter_type == "cs2_gsi" assert restored.enabled is True assert restored.adapter_config == {"token": "secret123"} assert len(restored.event_mappings) == 2 assert restored.event_mappings[0].event_type == "health" assert restored.event_mappings[1].event_type == "kill" assert restored.description == "Counter-Strike 2 game state integration" assert restored.tags == ["fps", "cs2"] def test_create_from_kwargs_generates_id(self): config = GameIntegrationConfig.create_from_kwargs(name="Test", adapter_type="test") assert config.id.startswith("gi_") assert len(config.id) == 11 # gi_ + 8 hex chars def test_apply_update_immutable(self): original = GameIntegrationConfig.create_from_kwargs( name="Original", adapter_type="test", enabled=True ) updated = original.apply_update(name="Updated", enabled=False) # Original unchanged assert original.name == "Original" assert original.enabled is True # Updated has new values assert updated.name == "Updated" assert updated.enabled is False assert updated.id == original.id assert updated.created_at == original.created_at assert updated.updated_at >= original.updated_at def test_apply_update_partial(self): original = GameIntegrationConfig.create_from_kwargs( name="Test", adapter_type="test", description="original desc" ) updated = original.apply_update(description="new desc") assert updated.name == "Test" assert updated.adapter_type == "test" assert updated.description == "new desc" # --------------------------------------------------------------------------- # Store CRUD tests # --------------------------------------------------------------------------- class TestGameIntegrationStoreCRUD: def test_create_and_get(self, store): config = store.create_integration( name="My Integration", adapter_type="webhook", description="Test integration", ) assert config.id.startswith("gi_") assert config.name == "My Integration" assert config.adapter_type == "webhook" assert config.enabled is True fetched = store.get_integration(config.id) assert fetched.name == "My Integration" def test_list_all(self, store): store.create_integration(name="Int 1", adapter_type="webhook") store.create_integration(name="Int 2", adapter_type="cs2_gsi") all_configs = store.get_all_integrations() assert len(all_configs) == 2 names = {c.name for c in all_configs} assert names == {"Int 1", "Int 2"} def test_update(self, store): config = store.create_integration( name="Old Name", adapter_type="webhook", ) updated = store.update_integration( config.id, name="New Name", enabled=False, description="Updated description", ) assert updated.name == "New Name" assert updated.enabled is False assert updated.description == "Updated description" assert updated.updated_at > config.updated_at def test_update_event_mappings(self, store): config = store.create_integration( name="Test", adapter_type="webhook", event_mappings=[EventMapping(event_type="health")], ) new_mappings = [ EventMapping(event_type="kill", effect="flash", color=[0, 255, 0]), EventMapping(event_type="death", effect="pulse", color=[255, 0, 0]), ] updated = store.update_integration(config.id, event_mappings=new_mappings) assert len(updated.event_mappings) == 2 assert updated.event_mappings[0].event_type == "kill" assert updated.event_mappings[1].event_type == "death" def test_delete(self, store): config = store.create_integration(name="ToDelete", adapter_type="webhook") store.delete_integration(config.id) with pytest.raises(EntityNotFoundError): store.get_integration(config.id) def test_delete_nonexistent(self, store): with pytest.raises(EntityNotFoundError): store.delete_integration("gi_nonexist") def test_get_nonexistent(self, store): with pytest.raises(EntityNotFoundError): store.get_integration("gi_nonexist") def test_create_with_adapter_config(self, store): config = store.create_integration( name="CS2", adapter_type="cs2_gsi", adapter_config={"auth_token": "secret", "port": 3000}, ) fetched = store.get_integration(config.id) assert fetched.adapter_config == {"auth_token": "secret", "port": 3000} # --------------------------------------------------------------------------- # Name uniqueness tests # --------------------------------------------------------------------------- class TestNameUniqueness: def test_duplicate_name_rejected(self, store): store.create_integration(name="Unique Name", adapter_type="webhook") with pytest.raises(ValueError, match="already exists"): store.create_integration(name="Unique Name", adapter_type="webhook") def test_empty_name_rejected(self, store): with pytest.raises(ValueError, match="required"): store.create_integration(name="", adapter_type="webhook") def test_whitespace_name_rejected(self, store): with pytest.raises(ValueError, match="required"): store.create_integration(name=" ", adapter_type="webhook") def test_update_same_name_allowed(self, store): config = store.create_integration(name="Same", adapter_type="webhook") updated = store.update_integration(config.id, name="Same") assert updated.name == "Same" def test_update_to_existing_name_rejected(self, store): store.create_integration(name="Name A", adapter_type="webhook") config_b = store.create_integration(name="Name B", adapter_type="webhook") with pytest.raises(ValueError, match="already exists"): store.update_integration(config_b.id, name="Name A") # --------------------------------------------------------------------------- # Persistence tests # --------------------------------------------------------------------------- class TestPersistence: def test_survives_reload(self, tmp_db): store1 = GameIntegrationStore(tmp_db) config = store1.create_integration( name="Persistent", adapter_type="webhook", event_mappings=[EventMapping(event_type="health", effect="gradient")], tags=["test"], ) # Create a new store instance (simulates restart) store2 = GameIntegrationStore(tmp_db) fetched = store2.get_integration(config.id) assert fetched.name == "Persistent" assert fetched.adapter_type == "webhook" assert len(fetched.event_mappings) == 1 assert fetched.event_mappings[0].event_type == "health" assert fetched.tags == ["test"] # --------------------------------------------------------------------------- # get_references tests # --------------------------------------------------------------------------- class TestGetReferences: def test_returns_empty(self, store): config = store.create_integration(name="Test", adapter_type="webhook") refs = store.get_references(config.id) assert refs == [] # --------------------------------------------------------------------------- # Secret encryption (adapter_config.auth_token) tests # --------------------------------------------------------------------------- class TestAdapterConfigSecretEncryption: def test_auth_token_round_trips(self): config = GameIntegrationConfig.create_from_kwargs( name="Webhook", adapter_type="generic_webhook", adapter_config={"auth_token": "super-secret", "auth_header": "X-Token"}, ) restored = GameIntegrationConfig.from_dict(config.to_dict()) # Decrypted at runtime — plaintext is recovered. assert restored.adapter_config["auth_token"] == "super-secret" # Non-secret fields are untouched. assert restored.adapter_config["auth_header"] == "X-Token" def test_stored_token_is_not_plaintext(self): config = GameIntegrationConfig.create_from_kwargs( name="Webhook", adapter_type="generic_webhook", adapter_config={"auth_token": "plaintext-secret"}, ) stored = config.to_dict() raw_token = stored["adapter_config"]["auth_token"] assert raw_token != "plaintext-secret" assert secret_box.is_encrypted(raw_token) def test_legacy_plaintext_row_still_decrypts(self): # Simulate a row written before encryption was added: auth_token is # bare plaintext (not an ENC: envelope). It must still load. legacy = { "id": "gi_legacy01", "name": "Legacy Webhook", "adapter_type": "generic_webhook", "enabled": True, "adapter_config": {"auth_token": "legacy-plaintext"}, "event_mappings": [], "created_at": "2024-01-01T00:00:00+00:00", "updated_at": "2024-01-01T00:00:00+00:00", } restored = GameIntegrationConfig.from_dict(legacy) assert restored.adapter_config["auth_token"] == "legacy-plaintext" def test_no_auth_token_key_is_safe(self): config = GameIntegrationConfig.create_from_kwargs( name="NoSecret", adapter_type="cs2", adapter_config={"max_gold": 99999}, ) restored = GameIntegrationConfig.from_dict(config.to_dict()) assert restored.adapter_config == {"max_gold": 99999} def test_db_stores_token_encrypted(self, tmp_db): store = GameIntegrationStore(tmp_db) store.create_integration( name="Webhook", adapter_type="generic_webhook", adapter_config={"auth_token": "db-secret"}, ) rows = tmp_db.load_all("game_integrations") assert len(rows) == 1 raw_token = rows[0]["adapter_config"]["auth_token"] assert raw_token != "db-secret" assert secret_box.is_encrypted(raw_token) def test_db_round_trip_after_reload(self, tmp_db): store1 = GameIntegrationStore(tmp_db) created = store1.create_integration( name="Webhook", adapter_type="generic_webhook", adapter_config={"auth_token": "reload-secret"}, ) # New store instance reads from disk and must decrypt. store2 = GameIntegrationStore(tmp_db) fetched = store2.get_integration(created.id) assert fetched.adapter_config["auth_token"] == "reload-secret"