Coverage for tests/test_package_schema.py: 95.27%
108 statements
« prev ^ index » next coverage.py v7.9.2, created at 2026-09-27 04:08 +0000
« prev ^ index » next coverage.py v7.9.2, created at 2026-09-27 04:08 +0000
1from decimal import Decimal
2from pathlib import Path
3from typing import Any, cast
5import pytest
6import ruamel.yaml
7from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue
8from pydantic_core import core_schema, from_json
10from prices.prices_types import Provider as _PydanticProvider
11from prices.utils import package_dir as prices_package_dir, simplify_json_schema
12from voice_prices.data import providers_schema
13from voice_prices.types import Usage as _RuntimeUsage
15_PROVIDERS_DIR = prices_package_dir / 'providers'
17_yaml = ruamel.yaml.YAML(typ='safe')
18_yaml.constructor.add_constructor( # pyright: ignore[reportUnknownMemberType]
19 'tag:yaml.org,2002:float',
20 lambda loader, node: Decimal(loader.construct_scalar(node)), # pyright: ignore[reportUnknownLambdaType, reportUnknownMemberType, reportUnknownArgumentType]
21)
24def _yaml_files() -> list[Path]:
25 return sorted(p for p in _PROVIDERS_DIR.iterdir() if p.suffix in ('.yml', '.yaml'))
28# Models in these providers' files that ship as TTS entries (input_kchars-priced).
29# Used by the provenance test to enforce per-entry pricing_source_url + prices_checked.
30# Empty in the initial schema-only commit; populated as catalog YAMLs land.
31_TTS_PROVIDER_IDS: set[str] = set()
34class CustomGenerateJsonSchema(GenerateJsonSchema):
35 def decimal_schema(self, schema: core_schema.DecimalSchema) -> JsonSchemaValue:
36 return self.float_schema(core_schema.float_schema())
39def remove_ignored_fields(json_schema: Any):
40 if isinstance(json_schema, dict):
41 json_schema = cast(dict[str, Any], json_schema)
43 for f in 'description', 'maxLength', 'minLength', 'pattern', 'additionalProperties':
44 json_schema.pop(f, None)
46 for value in json_schema.values():
47 remove_ignored_fields(value)
48 elif isinstance(json_schema, list):
49 for item in cast(list[Any], json_schema):
50 remove_ignored_fields(item)
53@pytest.mark.requires_latest_pydantic
54def test_package_schema():
55 package_schema = simplify_json_schema(providers_schema.json_schema(schema_generator=CustomGenerateJsonSchema))
56 remove_ignored_fields(package_schema)
58 # prices is not required in the model info package schema for simplicity
59 package_schema['$defs']['ModelInfo']['required'].append('prices')
61 # models is not required in the provider package schema for simplicity
62 package_schema['$defs']['Provider']['required'].append('models')
63 package_schema['$defs']['Provider']['properties']['pricing_urls']['items']['format'] = 'uri'
64 # ModelInfo.pricing_source_url is HttpUrl on the Pydantic side, str on the runtime side
65 package_schema['$defs']['ModelInfo']['properties']['pricing_source_url']['format'] = 'uri'
67 # work around for hack on ConditionalPrice
68 package_schema['$defs']['ConditionalPrice']['required'] = ['prices']
70 package_schema['$defs']['ClauseRegex']['properties']['regex']['format'] = 'regex'
72 prices_schema_path = prices_package_dir / 'data.schema.json'
73 prices_schema = from_json(prices_schema_path.read_bytes())
75 remove_ignored_fields(prices_schema)
77 assert prices_schema == package_schema
80@pytest.mark.parametrize('yaml_file', _yaml_files(), ids=lambda p: p.stem)
81def test_yaml_roundtrip(yaml_file: Path):
82 """Every catalog YAML parses with the Pydantic schema and exercises calc_price.
84 Parametrized across `glob('prices/providers/*.yml')` so new provider PRs surface
85 as named test rows in CI. The synthetic Usage is shaped per the model's priced
86 fields, just enough to exercise the engine without asserting specific numbers.
87 """
88 import pydantic_core
90 with yaml_file.open('rb') as f:
91 data = _yaml.load(f) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
93 # Validate against the Pydantic schema; raises on bad data.
94 provider = _PydanticProvider.model_validate_json(pydantic_core.to_json(data), strict=True)
95 assert provider.id, f'{yaml_file.name} has no provider id'
97 # Round-trip through the runtime provider schema as well, to confirm the JSON the
98 # build emits parses with the consumer-side dataclasses. We serialize the build-side
99 # Pydantic model to JSON, then validate that JSON with the runtime schema.
100 runtime_payload = b'[' + provider.model_dump_json(by_alias=True, exclude_none=True).encode() + b']'
101 runtime_providers = providers_schema.validate_json(runtime_payload)
102 assert len(runtime_providers) == 1
103 runtime_provider = runtime_providers[0]
105 # Exercise calc_price on at least one model per provider with a shape matching
106 # the model's priced fields. We don't assert numbers; we just confirm calc_price
107 # runs without raising for valid data shapes.
108 if runtime_provider.models: 108 ↛ exitline 108 didn't return from function 'test_yaml_roundtrip' because the condition on line 108 was always true
109 first_model = runtime_provider.models[0]
110 first_price_block = (
111 first_model.prices if not isinstance(first_model.prices, list) else first_model.prices[0].prices
112 )
114 synthetic = _RuntimeUsage()
115 if first_price_block.input_mtok is not None or first_price_block.input_audio_mtok is not None:
116 synthetic.input_tokens = 1000
117 if first_price_block.input_audio_mtok is not None:
118 # input_tokens carries the parent-bucket total; input_audio_tokens is the
119 # disjoint priced bucket the audio-mtok rate actually multiplies. Set it
120 # so the audio-mtok code path runs through calc_mtok_price instead of
121 # short-circuiting at zero tokens.
122 synthetic.input_audio_tokens = 1000
123 if first_price_block.output_mtok is not None or first_price_block.output_audio_mtok is not None:
124 synthetic.output_tokens = 100
125 if first_price_block.output_audio_mtok is not None:
126 synthetic.output_audio_tokens = 100
127 if first_price_block.input_kchars is not None:
128 synthetic.characters = 1000
129 if first_price_block.output_audio_kseconds is not None: 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 synthetic.audio_output_seconds = Decimal(60)
131 if first_price_block.input_audio_kseconds is not None:
132 synthetic.audio_input_seconds = Decimal(60)
133 # If no priced field is recognized, fall back to a token-shaped Usage so calc still runs.
134 if all(
135 getattr(synthetic, f) is None
136 for f in ('input_tokens', 'output_tokens', 'characters', 'audio_output_seconds', 'audio_input_seconds')
137 ):
138 synthetic.input_tokens = 1
140 result = first_model.calc_price(synthetic, runtime_provider)
141 assert result.total_price >= Decimal(0)
144def test_audio_priced_entries_have_provenance():
145 """Every model using a TTS or STT priced field (input_kchars, output_audio_kseconds,
146 input_audio_kseconds) MUST carry pricing_source_url + prices_checked.
148 Renamed from test_tts_entries_have_provenance in v0.0.7 to cover STT entries too.
149 Enforced across the catalog so any audio-pricing PR cannot land without provenance,
150 regardless of which provider file it modifies.
151 """
152 import pydantic_core
154 from prices.prices_types import ModelPrice as _PydanticModelPrice
156 missing: list[str] = []
157 for yaml_file in _yaml_files():
158 with yaml_file.open('rb') as f:
159 data = _yaml.load(f) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
160 provider = _PydanticProvider.model_validate_json(pydantic_core.to_json(data), strict=True)
162 for model in provider.models:
163 if isinstance(model.prices, list):
164 price_blocks: list[_PydanticModelPrice] = [cp.prices for cp in model.prices]
165 else:
166 price_blocks = [model.prices]
168 uses_audio_priced_field = any(
169 block.input_kchars is not None
170 or block.output_audio_kseconds is not None
171 or block.input_audio_kseconds is not None
172 for block in price_blocks
173 )
174 if not uses_audio_priced_field:
175 continue
177 if model.pricing_source_url is None: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true
178 missing.append(f'{provider.id}/{model.id}: missing pricing_source_url')
179 if model.prices_checked is None: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 missing.append(f'{provider.id}/{model.id}: missing prices_checked')
182 assert not missing, '\n'.join(missing)
185def _template_files() -> list[Path]:
186 return sorted((Path(__file__).parent.parent / 'docs' / 'templates').glob('*.yml'))
189@pytest.mark.parametrize('template_path', _template_files(), ids=lambda p: p.stem)
190def test_provider_template_validates(template_path: Path):
191 """Every contributor template in docs/templates/ must parse against the same
192 Pydantic schema as real providers, so contributors who copy-paste-and-edit
193 start from a working baseline.
194 """
195 import pydantic_core
197 with template_path.open('rb') as f:
198 data = _yaml.load(f) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
199 # Will raise pydantic.ValidationError if the template is broken.
200 _PydanticProvider.model_validate_json(pydantic_core.to_json(data), strict=True)
203def test_expected_templates_present():
204 """Guard against the empty-glob pitfall: a renamed or deleted template would
205 make the parametrized test above silently cover nothing. Pin the known set.
206 """
207 found = {p.name for p in _template_files()}
208 # All known templates must be present (guards the empty-glob edge case above).
209 assert {'provider-tts.yml', 'provider-stt.yml', 'provider-llm.yml'} <= found