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

1from decimal import Decimal 

2from pathlib import Path 

3from typing import Any, cast 

4 

5import pytest 

6import ruamel.yaml 

7from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue 

8from pydantic_core import core_schema, from_json 

9 

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 

14 

15_PROVIDERS_DIR = prices_package_dir / 'providers' 

16 

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) 

22 

23 

24def _yaml_files() -> list[Path]: 

25 return sorted(p for p in _PROVIDERS_DIR.iterdir() if p.suffix in ('.yml', '.yaml')) 

26 

27 

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() 

32 

33 

34class CustomGenerateJsonSchema(GenerateJsonSchema): 

35 def decimal_schema(self, schema: core_schema.DecimalSchema) -> JsonSchemaValue: 

36 return self.float_schema(core_schema.float_schema()) 

37 

38 

39def remove_ignored_fields(json_schema: Any): 

40 if isinstance(json_schema, dict): 

41 json_schema = cast(dict[str, Any], json_schema) 

42 

43 for f in 'description', 'maxLength', 'minLength', 'pattern', 'additionalProperties': 

44 json_schema.pop(f, None) 

45 

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) 

51 

52 

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) 

57 

58 # prices is not required in the model info package schema for simplicity 

59 package_schema['$defs']['ModelInfo']['required'].append('prices') 

60 

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' 

66 

67 # work around for hack on ConditionalPrice 

68 package_schema['$defs']['ConditionalPrice']['required'] = ['prices'] 

69 

70 package_schema['$defs']['ClauseRegex']['properties']['regex']['format'] = 'regex' 

71 

72 prices_schema_path = prices_package_dir / 'data.schema.json' 

73 prices_schema = from_json(prices_schema_path.read_bytes()) 

74 

75 remove_ignored_fields(prices_schema) 

76 

77 assert prices_schema == package_schema 

78 

79 

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. 

83 

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 

89 

90 with yaml_file.open('rb') as f: 

91 data = _yaml.load(f) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] 

92 

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' 

96 

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] 

104 

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 ) 

113 

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 

139 

140 result = first_model.calc_price(synthetic, runtime_provider) 

141 assert result.total_price >= Decimal(0) 

142 

143 

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. 

147 

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 

153 

154 from prices.prices_types import ModelPrice as _PydanticModelPrice 

155 

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) 

161 

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] 

167 

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 

176 

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') 

181 

182 assert not missing, '\n'.join(missing) 

183 

184 

185def _template_files() -> list[Path]: 

186 return sorted((Path(__file__).parent.parent / 'docs' / 'templates').glob('*.yml')) 

187 

188 

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 

196 

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) 

201 

202 

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