/
/
/
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