music-assistant-server

1"""Generated converter tests using fixture test mappings.
2
3This module provides automated converter testing for the Nicovideo provider.
4The test system is type-safe with automatic fixture updates and parameterized
5converter/type specification through common test functions.
6
7Type System:
8    - API Responses: Pydantic BaseModel (for JSON validation and fixture saving)
9    - Converter Results: mashumaro DataClassDictMixin (for snapshot serialization)
10
11Architecture Overview:
12    1. Fixture Collection (fixtures/scripts/api_fixture_collector.py):
13       - Collects API responses by calling Niconico APIs
14       - Saves responses as JSON fixtures in generated/fixtures/
15
16    2. Type Mapping (fixtures/fixture_type_mapping.py):
17       - Maps fixture paths to their Pydantic types
18       - Auto-generates generated/fixture_types.py
19
20    3. Converter Mapping (fixtures/api_response_converter_mapping.py):
21       - Defines which converter function to use for each API response type
22       - Registry provides O(1) type -> converter lookup
23
24    4. Test Execution (this file):
25       - Loads fixtures using FixtureLoader
26       - Applies converters via mapping registry
27       - Validates results against snapshots
28
29
30Adding New API Endpoints:
31    See: tests/providers/nicovideo/fixtures/scripts/api_fixture_collector.py
32    Add collection method and call from collect_all_fixtures()
33    Note: API response types must inherit from Pydantic BaseModel
34
35
36Adding New Converters:
37    1. Implement converter: music_assistant/providers/nicovideo/converters/
38       Note: Return types must inherit from mashumaro DataClassDictMixin
39    2. Register: music_assistant/providers/nicovideo/converters/manager.py
40    3. Add mapping: tests/providers/nicovideo/fixtures/api_response_converter_mapping.py
41
42"""
43
44from __future__ import annotations
45
46import warnings
47from pathlib import Path
48from typing import TYPE_CHECKING
49
50import pytest
51
52from tests.providers.nicovideo.helpers import (
53    to_dict_for_snapshot,
54)
55
56if TYPE_CHECKING:
57    from pydantic import BaseModel
58    from syrupy.assertion import SnapshotAssertion
59
60    from music_assistant.providers.nicovideo.converters.manager import NicovideoConverterManager
61    from tests.providers.nicovideo.fixtures.api_response_converter_mapping import (
62        APIResponseConverterMappingRegistry,
63        SnapshotableItem,
64    )
65    from tests.providers.nicovideo.fixtures.fixture_loader import FixtureLoader
66
67
68from .constants import GENERATED_FIXTURES_DIR
69
70
71class ConverterTestRunner:
72    """Helper class to run converter tests with fixture files."""
73
74    def __init__(
75        self,
76        mapping_registry: APIResponseConverterMappingRegistry,
77        converter_manager: NicovideoConverterManager,
78        fixture_loader: FixtureLoader,
79        snapshot: SnapshotAssertion,
80        fixtures_dir: Path,
81    ) -> None:
82        """Initialize the test runner."""
83        self.mapping_registry = mapping_registry
84        self.converter_manager = converter_manager
85        self.fixture_loader = fixture_loader
86        self.snapshot = snapshot
87        self.fixtures_dir = fixtures_dir
88        self.failed_tests: list[str] = []
89        self.skipped_tests: list[str] = []
90
91    def run_all_tests(self) -> None:
92        """Execute converter tests for all fixture files."""
93        # Recursively get all JSON files
94        json_files = list(self.fixtures_dir.rglob("*.json"))
95
96        if not json_files:
97            pytest.skip("No fixture files found")
98
99        for fixture_path in json_files:
100            self._process_fixture_file(fixture_path)
101
102        # Report results
103        self._report_test_results()
104
105    def _process_fixture_file(self, fixture_path: Path) -> None:
106        """Process a single fixture file."""
107        relative_path = fixture_path.relative_to(self.fixtures_dir)
108        fixture_name = str(relative_path)
109
110        try:
111            # Load fixture data
112            fixture_data = self.fixture_loader.load_fixture(relative_path)
113            if fixture_data is None:
114                self.failed_tests.append(f"{fixture_name}: Failed to load fixture")
115                return
116
117            fixture_list = fixture_data if isinstance(fixture_data, list) else [fixture_data]
118
119            for fixture_index, fixture in enumerate(fixture_list):
120                fixture_id = (
121                    f"{fixture_name}[{fixture_index}]" if len(fixture_list) > 1 else fixture_name
122                )
123                # fixture is BaseModel type from FixtureLoader.load_fixture
124                self._process_single_fixture(fixture_id, fixture)
125
126        except Exception as e:
127            self.failed_tests.append(f"{fixture_name}: {e}")
128
129    def _process_single_fixture(self, fixture_id: str, fixture: BaseModel) -> None:
130        """Process a single fixture within a fixture file."""
131        try:
132            # Get mapping directly by type
133            mapping = self.mapping_registry.get_by_type(type(fixture))
134            if mapping is None:
135                # Skip if no mapping found
136                self.skipped_tests.append(f"{fixture_id}: No mapping for {type(fixture).__name__}")
137                return
138
139            # Execute test
140            converted_result = mapping.convert_func(fixture, self.converter_manager)
141            if converted_result is None:
142                self.skipped_tests.append(f"{fixture_id}: No conversion result")
143                return
144
145            # Process all converted items (handles both single and list results)
146            self._process_all_converted_items(fixture_id, converted_result)
147
148        except Exception as e:
149            self.failed_tests.append(f"{fixture_id}: {e}")
150
151    def _process_all_converted_items(
152        self,
153        base_fixture_id: str,
154        converted_result: SnapshotableItem | list[SnapshotableItem],
155    ) -> None:
156        """Process all items in converted result (handles both single and list)."""
157        # Convert to list for uniform processing
158        items = converted_result if isinstance(converted_result, list) else [converted_result]
159
160        for idx, item in enumerate(items):
161            # Generate unique snapshot ID for each item
162            snapshot_id = f"{base_fixture_id}_{idx}" if len(items) > 1 else base_fixture_id
163            self._process_converted_result(snapshot_id, item)
164
165    def _process_converted_result(
166        self,
167        snapshot_id: str,
168        converted: SnapshotableItem,
169    ) -> None:
170        """Process a single converted result and compare with snapshot."""
171        stable_dict = to_dict_for_snapshot(converted)
172
173        # Compare with snapshot
174        converted_snapshot = self.snapshot(name=snapshot_id)
175        snapshot_matches = converted_snapshot == stable_dict
176
177        if not snapshot_matches:
178            # Get detailed diff information
179            diff_lines = converted_snapshot.get_assert_diff()
180            diff_summary = "\n".join(diff_lines[:10])  # Limit to first 10 lines
181            if len(diff_lines) > 10:
182                diff_summary += f"\n... ({len(diff_lines) - 10} more lines)"
183
184            self.failed_tests.append(
185                f"{snapshot_id}: Converted result doesn't match snapshot\nDiff:\n{diff_summary}"
186            )
187
188    def _report_test_results(self) -> None:
189        """Report the final test results."""
190        if self.failed_tests:
191            error_msg = f"Failed tests ({len(self.failed_tests)}):\n" + "\n".join(
192                f"  - {test}" for test in self.failed_tests
193            )
194            pytest.fail(error_msg)
195
196        if self.skipped_tests:
197            skip_msg = f"Skipped tests ({len(self.skipped_tests)}):\n" + "\n".join(
198                f"  - {test}" for test in self.skipped_tests
199            )
200            warnings.warn(skip_msg, stacklevel=2)
201
202
203def test_converter_with_fixture(
204    mapping_registry: APIResponseConverterMappingRegistry,
205    converter_manager: NicovideoConverterManager,
206    fixture_loader: FixtureLoader,
207    snapshot: SnapshotAssertion,
208) -> None:
209    """Execute converter tests for all fixture files."""
210    runner = ConverterTestRunner(
211        mapping_registry=mapping_registry,
212        converter_manager=converter_manager,
213        fixture_loader=fixture_loader,
214        snapshot=snapshot,
215        fixtures_dir=GENERATED_FIXTURES_DIR,
216    )
217
218    runner.run_all_tests()
219