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