"""Tests for MappingAdapter — YAML parsing, payload translation, validation.""" import textwrap from pathlib import Path import pytest from ledgrab.core.game_integration.mapping_adapter import ( MappingAdapter, load_adapter_from_yaml, validate_adapter_yaml, ) # ── YAML validation tests ─────────────────────────────────────────────── class TestValidateAdapterYaml: def test_valid_minimal(self) -> None: data = { "name": "test_adapter", "game": "TestGame", "protocol": "webhook", "mappings": [ {"source_path": "player.health", "event": "health"}, ], } errors = validate_adapter_yaml(data) assert errors == [] def test_missing_name(self) -> None: data = { "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], } errors = validate_adapter_yaml(data) assert any("name" in e for e in errors) def test_missing_game(self) -> None: data = { "name": "test", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], } errors = validate_adapter_yaml(data) assert any("game" in e for e in errors) def test_invalid_protocol(self) -> None: data = { "name": "test", "game": "TestGame", "protocol": "invalid", "mappings": [{"source_path": "x", "event": "health"}], } errors = validate_adapter_yaml(data) assert any("protocol" in e for e in errors) def test_empty_mappings(self) -> None: data = { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [], } errors = validate_adapter_yaml(data) assert any("mappings" in e for e in errors) def test_unknown_event_type(self) -> None: data = { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "nonexistent_event"}], } errors = validate_adapter_yaml(data) assert any("unknown event type" in e for e in errors) def test_invalid_trigger_mode(self) -> None: data = { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [ {"source_path": "x", "event": "health", "trigger": "bad_mode"}, ], } errors = validate_adapter_yaml(data) assert any("trigger mode" in e for e in errors) def test_non_numeric_min_max(self) -> None: data = { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [ {"source_path": "x", "event": "health", "min": "not_a_number"}, ], } errors = validate_adapter_yaml(data) assert any("'min' must be numeric" in e for e in errors) # ── MappingAdapter payload parsing tests ───────────────────────────────── class TestMappingAdapterParsePayload: def _make_adapter(self, mappings: list[dict]) -> MappingAdapter: return MappingAdapter( { "name": "test_adapter", "game": "TestGame", "protocol": "webhook", "mappings": mappings, } ) def test_continuous_value_normalization(self) -> None: adapter = self._make_adapter( [ { "source_path": "player.health", "event": "health", "min": 0, "max": 100, "trigger": "on_value", }, ] ) payload = {"player": {"health": 75}} events, state = adapter.parse_payload(payload, {"adapter_id": "test"}, {}) assert len(events) == 1 assert events[0].event_type == "health" assert events[0].value == pytest.approx(0.75) assert events[0].adapter_id == "test" def test_value_clamped_to_0_1(self) -> None: adapter = self._make_adapter( [ { "source_path": "val", "event": "health", "min": 0, "max": 100, "trigger": "on_value", }, ] ) events_over, _ = adapter.parse_payload({"val": 150}, {"adapter_id": "t"}, {}) assert events_over[0].value == 1.0 events_under, _ = adapter.parse_payload({"val": -50}, {"adapter_id": "t"}, {}) assert events_under[0].value == 0.0 def test_on_change_trigger(self) -> None: adapter = self._make_adapter( [ { "source_path": "kills", "event": "kill", "min": 0, "max": 50, "trigger": "on_change", }, ] ) # First call: no prev_state, should emit events1, state1 = adapter.parse_payload({"kills": 5}, {"adapter_id": "t"}, {}) assert len(events1) == 1 # Same value: should NOT emit events2, state2 = adapter.parse_payload({"kills": 5}, {"adapter_id": "t"}, state1) assert len(events2) == 0 # Changed value: should emit events3, _ = adapter.parse_payload({"kills": 6}, {"adapter_id": "t"}, state2) assert len(events3) == 1 def test_on_increase_trigger(self) -> None: adapter = self._make_adapter( [ { "source_path": "kills", "event": "kill", "min": 0, "max": 50, "trigger": "on_increase", }, ] ) # First call: prev_state empty, should NOT emit (no baseline) events1, state1 = adapter.parse_payload({"kills": 5}, {"adapter_id": "t"}, {}) assert len(events1) == 0 # Decrease: should NOT emit events2, state2 = adapter.parse_payload({"kills": 3}, {"adapter_id": "t"}, state1) assert len(events2) == 0 # Increase: should emit events3, _ = adapter.parse_payload({"kills": 7}, {"adapter_id": "t"}, state2) assert len(events3) == 1 def test_on_decrease_trigger(self) -> None: adapter = self._make_adapter( [ { "source_path": "hp", "event": "damage_taken", "min": 0, "max": 100, "trigger": "on_decrease", }, ] ) _, state1 = adapter.parse_payload({"hp": 100}, {"adapter_id": "t"}, {}) events, _ = adapter.parse_payload({"hp": 80}, {"adapter_id": "t"}, state1) assert len(events) == 1 assert events[0].event_type == "damage_taken" def test_missing_path_skipped(self) -> None: adapter = self._make_adapter( [ {"source_path": "player.health", "event": "health", "trigger": "on_value"}, ] ) events, _ = adapter.parse_payload({"player": {}}, {"adapter_id": "t"}, {}) assert len(events) == 0 def test_non_numeric_value_emits_trigger(self) -> None: adapter = self._make_adapter( [ {"source_path": "status", "event": "buffed", "trigger": "on_value"}, ] ) events, _ = adapter.parse_payload({"status": "active"}, {"adapter_id": "t"}, {}) assert len(events) == 1 assert events[0].value == 1.0 def test_nested_json_path(self) -> None: adapter = self._make_adapter( [ { "source_path": "a.b.c.d", "event": "health", "min": 0, "max": 10, "trigger": "on_value", }, ] ) payload = {"a": {"b": {"c": {"d": 5}}}} events, _ = adapter.parse_payload(payload, {"adapter_id": "t"}, {}) assert len(events) == 1 assert events[0].value == pytest.approx(0.5) def test_multiple_mappings(self) -> None: adapter = self._make_adapter( [ { "source_path": "hp", "event": "health", "min": 0, "max": 100, "trigger": "on_value", }, {"source_path": "mp", "event": "mana", "min": 0, "max": 200, "trigger": "on_value"}, ] ) events, _ = adapter.parse_payload( {"hp": 50, "mp": 100}, {"adapter_id": "t"}, {}, ) assert len(events) == 2 types = {e.event_type for e in events} assert types == {"health", "mana"} # ── Auth validation tests ──────────────────────────────────────────────── class TestMappingAdapterAuth: def test_no_auth_accepts_all(self) -> None: adapter = MappingAdapter( { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], } ) assert adapter.validate_auth({}, {}, {}) is True def test_header_auth_valid(self) -> None: adapter = MappingAdapter( { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], "auth": {"type": "header", "header": "X-Auth-Token"}, } ) result = adapter.validate_auth( {"X-Auth-Token": "secret123"}, {}, {"auth_token": "secret123"}, ) assert result is True def test_header_auth_invalid(self) -> None: adapter = MappingAdapter( { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], "auth": {"type": "header", "header": "X-Auth-Token"}, } ) result = adapter.validate_auth( {"X-Auth-Token": "wrong"}, {}, {"auth_token": "secret123"}, ) assert result is False # ── YAML file loading tests ────────────────────────────────────────────── class TestLoadAdapterFromYaml: def test_load_valid_yaml(self, tmp_path: Path) -> None: yaml_content = textwrap.dedent( """\ name: cs2_gsi game: "Counter-Strike 2" protocol: webhook mappings: - source_path: player.state.health event: health min: 0 max: 100 - source_path: player.state.armor event: armor min: 0 max: 100 auth: type: header header: X-GSI-Auth """ ) yaml_file = tmp_path / "cs2.yaml" yaml_file.write_text(yaml_content) adapter = load_adapter_from_yaml(yaml_file) assert adapter.name == "cs2_gsi" assert adapter.game == "Counter-Strike 2" assert adapter.protocol == "webhook" assert "health" in adapter.supported_events assert "armor" in adapter.supported_events def test_load_nonexistent_file_raises(self) -> None: with pytest.raises(FileNotFoundError): load_adapter_from_yaml("/nonexistent/path.yaml") def test_load_invalid_yaml_raises(self, tmp_path: Path) -> None: yaml_file = tmp_path / "bad.yaml" yaml_file.write_text("name: test\n") # Missing required fields with pytest.raises(ValueError, match="Invalid adapter YAML"): load_adapter_from_yaml(yaml_file) def test_load_non_dict_yaml_raises(self, tmp_path: Path) -> None: yaml_file = tmp_path / "list.yaml" yaml_file.write_text("- item1\n- item2\n") with pytest.raises(ValueError, match="must be a dict"): load_adapter_from_yaml(yaml_file) def test_loaded_adapter_parses_payload(self, tmp_path: Path) -> None: yaml_content = textwrap.dedent( """\ name: test_game game: TestGame protocol: webhook mappings: - source_path: hp event: health min: 0 max: 100 trigger: on_value """ ) yaml_file = tmp_path / "test.yaml" yaml_file.write_text(yaml_content) adapter = load_adapter_from_yaml(yaml_file) events, _ = adapter.parse_payload( {"hp": 60}, {"adapter_id": "loaded_test"}, {}, ) assert len(events) == 1 assert events[0].value == pytest.approx(0.6) # ── Properties and metadata tests ──────────────────────────────────────── class TestMappingAdapterMetadata: def test_config_schema_with_auth(self) -> None: adapter = MappingAdapter( { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], "auth": {"type": "header", "header": "X-Token"}, } ) schema = adapter.get_config_schema() assert "auth_token" in schema["properties"] def test_config_schema_without_auth(self) -> None: adapter = MappingAdapter( { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], } ) schema = adapter.get_config_schema() assert schema["properties"] == {} def test_setup_instructions_from_yaml(self) -> None: adapter = MappingAdapter( { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], "setup_instructions": "# Step 1\nDo this.", } ) assert "Step 1" in adapter.get_setup_instructions() def test_setup_instructions_default(self) -> None: adapter = MappingAdapter( { "name": "test", "game": "TestGame", "protocol": "webhook", "mappings": [{"source_path": "x", "event": "health"}], } ) instructions = adapter.get_setup_instructions() assert "TestGame" in instructions