Coverage for tests/test_source_feed.py: 98.96%

1"""Tests for the provider pricing-feed reader (prices.source_feed). 

2 

3The feed is a format other companies are being asked to implement, so the conversion grid and the 

4fail-closed behaviour are pinned hard: a silently mis-converted unit would publish a wrong price 

5under a provider's own name. 

6""" 

7 

8from __future__ import annotations (empty)

9 

10import json (empty)

11from decimal import Decimal (empty)

12from pathlib import Path (empty)

13from typing import Any (empty)

14 

15import pytest (empty)

16from pydantic import ValidationError (empty)

17 

18from prices.source_feed import ( (empty)

19 DATA_JSON, 

20 RATE_FIELDS, 

21 Feed, 

22 FeedModel, 

23 base_rates, 

24 compare, 

25 convert_rates, 

26 feed_check, 

27 load_feed, 

28 render_rate_grid, 

29 render_report, 

30) 

31 

32 

33def _rate(meter: str, unit: str, amount: str, **extra: Any) -> dict[str, Any]: (empty)

34 return {'meter': meter, 'unit': unit, 'amount': amount, **extra} (empty)

35 

36 

37# ---- conversion -------------------------------------------------------------- 

38 

39 

40def test_every_meter_unit_pair_converts_to_its_documented_field(): (empty)

41 for (meter, unit), mapping in RATE_FIELDS.items(): (empty)

42 model = FeedModel.model_validate({'id': 'm', 'rates': [_rate(meter, unit, '1')]}) (empty)

43 fields, problems = convert_rates(model) (empty)

44 assert problems == [] (empty)

45 assert list(fields) == [mapping.field], f'{meter}/{unit} landed in the wrong field' (empty)

46 

47 

48def test_per_minute_audio_converts_exactly(): (empty)

49 # 0.015/min is Telnyx's published STT rate and must land on 0.25 exactly, not 0.2499999. 

50 model = FeedModel.model_validate({'id': 'm', 'rates': [_rate('audio_input', 'minute', '0.015')]}) (empty)

51 fields, _ = convert_rates(model) (empty)

52 assert fields['input_audio_kseconds'] == Decimal('0.25') (empty)

53 

54 

55def test_per_character_and_per_token_scale_correctly(): (empty)

56 model = FeedModel.model_validate( (empty)

57 { 

58 'id': 'm', 

59 'rates': [_rate('text_input', 'character', '0.000003'), _rate('text_output', 'token', '0.000010')], 

60 } 

61 ) 

62 fields, _ = convert_rates(model) (empty)

63 assert fields['input_kchars'] == Decimal('0.003') # x 1,000 (empty)

64 assert fields['output_mtok'] == Decimal('10') # x 1,000,000 (empty)

65 

66 

67def test_unknown_meter_unit_pair_is_refused_not_guessed(): (empty)

68 # A per-minute text rate is meaningless. Guessing at it would publish a wrong number. 

69 model = FeedModel.model_validate({'id': 'm', 'rates': [_rate('text_input', 'minute', '5')]}) (empty)

70 fields, problems = convert_rates(model) (empty)

71 assert fields == {} (empty)

72 assert problems and 'no mapping' in problems[0] (empty)

73 

74 

75def test_two_rates_competing_for_one_field_is_refused(): (empty)

76 model = FeedModel.model_validate( (empty)

77 {'id': 'm', 'rates': [_rate('audio_input', 'second', '0.001'), _rate('audio_input', 'minute', '0.06')]} 

78 ) 

79 fields, problems = convert_rates(model) (empty)

80 assert problems and 'two base rates' in problems[0] (empty)

81 assert len(fields) == 1 # the first one wins, the collision is reported rather than merged (empty)

82 

83 

84def test_negative_amount_is_refused(): (empty)

85 model = FeedModel.model_validate({'id': 'm', 'rates': [_rate('text_input', 'character', '-0.1')]}) (empty)

86 fields, problems = convert_rates(model) (empty)

87 assert fields == {} (empty)

88 assert problems and 'negative' in problems[0] (empty)

89 

90 

91def test_base_rates_excludes_plan_volume_and_voice_class_rates(): (empty)

92 # Only the unqualified rate is comparable to a catalog headline price. 

93 model = FeedModel.model_validate( (empty)

94 { 

95 'id': 'm', 

96 'rates': [ 

97 _rate('text_input', 'character', '0.000030'), 

98 _rate('text_input', 'character', '0.000024', plan='enterprise'), 

99 _rate('text_input', 'character', '0.000020', from_quantity=1_000_000_000), 

100 _rate('text_input', 'character', '0.000045', voice_class='cloned'), 

101 ], 

102 } 

103 ) 

104 assert len(model.rates) == 4 (empty)

105 assert len(base_rates(model)) == 1 (empty)

106 fields, problems = convert_rates(model) (empty)

107 assert problems == [] (empty)

108 assert fields == {'input_kchars': Decimal('0.03')} (empty)

109 

110 

111# ---- validation -------------------------------------------------------------- 

112 

113 

114def test_a_typo_in_a_rate_key_is_rejected(): (empty)

115 # Rates are the money, so extra keys are forbidden and a misspelling surfaces immediately. 

116 with pytest.raises(ValidationError): (empty)

117 FeedModel.model_validate({'id': 'm', 'rates': [{'meter': 'text_input', 'unit': 'character', 'ammount': '1'}]}) (empty)

118 

119 

120def test_provider_metadata_on_a_model_is_allowed_through(): (empty)

121 # Extra keys at the model level are fine: a provider may carry their own fields. 

122 model = FeedModel.model_validate({'id': 'm', 'region': 'us-east', 'rates': [_rate('text_input', 'token', '1')]}) (empty)

123 assert model.id == 'm' (empty)

124 

125 

126def test_an_unknown_meter_fails_validation(): (empty)

127 with pytest.raises(ValidationError): (empty)

128 FeedModel.model_validate({'id': 'm', 'rates': [_rate('vibes', 'character', '1')]}) (empty)

129 

130 

131def test_load_feed_warns_when_an_amount_is_a_json_number(tmp_path: Path): (empty)

132 path = tmp_path / 'pricing.json' (empty)

133 path.write_text( (empty)

134 json.dumps( 

135 { 

136 'provider': 'acme', 

137 'models': [{'id': 'a', 'rates': [{'meter': 'text_input', 'unit': 'character', 'amount': 0.000003}]}], 

138 } 

139 ) 

140 ) 

141 feed, warnings = load_feed(str(path)) (empty)

142 assert feed.provider == 'acme' (empty)

143 assert warnings and 'strings' in warnings[0] (empty)

144 

145 

146def test_load_feed_is_silent_when_amounts_are_strings(tmp_path: Path): (empty)

147 path = tmp_path / 'pricing.json' (empty)

148 path.write_text( (empty)

149 json.dumps( 

150 {'provider': 'acme', 'models': [{'id': 'a', 'rates': [_rate('text_input', 'character', '0.000003')]}]} 

151 ) 

152 ) 

153 _, warnings = load_feed(str(path)) (empty)

154 assert warnings == [] (empty)

155 

156 

157# ---- comparison -------------------------------------------------------------- 

158 

159CATALOG: list[dict[str, Any]] = [ (empty)

160 { 

161 'id': 'acme', 

162 'name': 'Acme', 

163 'models': [ 

164 {'id': 'acme-tts-1', 'prices': {'input_kchars': 0.03}}, 

165 {'id': 'acme-tts-2', 'prices': {'input_kchars': 0.05}}, 

166 ], 

167 } 

168] 

169 

170 

171def _feed(*models: dict[str, Any]) -> Feed: (empty)

172 return Feed.model_validate({'provider': 'acme', 'models': list(models)}) (empty)

173 

174 

175def test_matching_rate_reports_match(): (empty)

176 feed = _feed({'id': 'acme-tts-1', 'rates': [_rate('text_input', 'character', '0.000030')]}) (empty)

177 kinds = {f.kind for f in compare(feed, CATALOG)} (empty)

178 assert 'MATCH' in kinds (empty)

179 assert 'DRIFT' not in kinds (empty)

180 

181 

182def test_changed_rate_reports_drift_with_the_delta(): (empty)

183 feed = _feed({'id': 'acme-tts-1', 'rates': [_rate('text_input', 'character', '0.000045')]}) (empty)

184 drift = [f for f in compare(feed, CATALOG) if f.kind == 'DRIFT'] (empty)

185 assert len(drift) == 1 (empty)

186 assert '+50.0%' in drift[0].detail (empty)

187 

188 

189def test_model_absent_from_the_catalog_reports_new(): (empty)

190 feed = _feed({'id': 'acme-tts-9', 'rates': [_rate('text_input', 'character', '0.00001')]}) (empty)

191 new = [f for f in compare(feed, CATALOG) if f.kind == 'NEW'] (empty)

192 assert [f.model_id for f in new] == ['acme-tts-9'] (empty)

193 

194 

195def test_model_absent_from_the_feed_reports_missing_in_the_same_modality(): (empty)

196 feed = _feed({'id': 'acme-tts-1', 'rates': [_rate('text_input', 'character', '0.000030')]}) (empty)

197 missing = [f for f in compare(feed, CATALOG) if f.kind == 'MISSING'] (empty)

198 assert [f.model_id for f in missing] == ['acme-tts-2'] (empty)

199 

200 

201def test_missing_is_scoped_to_modalities_the_feed_actually_covers(): (empty)

202 # A TTS-only feed must not report every LLM model the provider has as missing. 

203 catalog: list[dict[str, Any]] = [ (empty)

204 { 

205 'id': 'acme', 

206 'models': [ 

207 {'id': 'acme-tts-1', 'prices': {'input_kchars': 0.03}}, 

208 {'id': 'acme-llm-1', 'prices': {'input_mtok': 1.0, 'output_mtok': 2.0}}, 

209 ], 

210 } 

211 ] 

212 feed = _feed({'id': 'acme-tts-1', 'rates': [_rate('text_input', 'character', '0.000030')]}) (empty)

213 assert [f for f in compare(feed, catalog) if f.kind == 'MISSING'] == [] (empty)

214 

215 

216def test_rounding_difference_is_not_reported_as_drift(): (empty)

217 # A per-minute rate cannot round-trip exactly through $/1k seconds. 0.0043/min lands on 

218 # 0.07166666..., and a catalog storing 0.071667 is the same price, not a drift. 

219 catalog: list[dict[str, Any]] = [ (empty)

220 {'id': 'acme', 'models': [{'id': 'a', 'prices': {'input_audio_kseconds': 0.071667}}]} 

221 ] 

222 feed = _feed({'id': 'a', 'rates': [_rate('audio_input', 'minute', '0.0043')]}) (empty)

223 assert [f for f in compare(feed, catalog) if f.kind == 'DRIFT'] == [] (empty)

224 

225 

226def test_unknown_provider_reports_everything_as_new_rather_than_crashing(): (empty)

227 feed = Feed.model_validate( (empty)

228 {'provider': 'nobody', 'models': [{'id': 'x', 'rates': [_rate('text_input', 'character', '0.00001')]}]} 

229 ) 

230 assert [f.kind for f in compare(feed, CATALOG)] == ['NEW'] (empty)

231 

232 

233# ---- round trip against real catalog data ------------------------------------ 

234 

235 

236def test_a_feed_built_from_the_real_catalog_compares_clean(): (empty)

237 """The format has to be able to express prices this project already publishes. 

238 

239 Takes Telnyx's committed rates, inverts them back into feed units, and asserts the feed round 

240 trips to zero drift. If the conversion grid is wrong in either direction, this fails. 

241 """ 

242 data: list[dict[str, Any]] = json.loads(DATA_JSON.read_text()) (empty)

243 telnyx = next(p for p in data if p['id'] == 'telnyx') (empty)

244 

245 models: list[dict[str, Any]] = [] (empty)

246 for model in telnyx['models']: (empty)

247 prices: dict[str, Any] = model['prices'] (empty)

248 if 'input_kchars' in prices: (empty)

249 amount = Decimal(str(prices['input_kchars'])) / 1000 # $/1k chars -> $/char (empty)

250 models.append({'id': model['id'], 'rates': [_rate('text_input', 'character', str(amount))]}) (empty)

251 elif 'input_audio_kseconds' in prices: (empty)

252 amount = Decimal(str(prices['input_audio_kseconds'])) * 60 / 1000 # $/1k sec -> $/min (empty)

253 models.append({'id': model['id'], 'rates': [_rate('audio_input', 'minute', str(amount))]}) (empty)

254 

255 assert len(models) == 5 # Telnyx ships 4 TTS models and 1 STT model (empty)

256 findings = compare(Feed.model_validate({'provider': 'telnyx', 'models': models}), data) (empty)

257 assert [f for f in findings if f.kind != 'MATCH'] == [] (empty)

258 assert len(findings) == 5 (empty)

259 

260 

261# ---- report and CLI ---------------------------------------------------------- 

262 

263 

264def test_render_rate_grid_documents_every_supported_pair(): (empty)

265 grid = render_rate_grid() (empty)

266 for (meter, unit), mapping in RATE_FIELDS.items(): (empty)

267 assert f'`{meter}`' in grid (empty)

268 assert f'`{mapping.field}`' in grid (empty)

269 assert f'`{unit}`' in grid (empty)

270 assert grid.count('\n') == len(RATE_FIELDS) + 1 # header + separator + one row each (empty)

271 

272 

273def test_report_names_the_human_step_when_there_is_drift(): (empty)

274 feed = _feed({'id': 'acme-tts-1', 'rates': [_rate('text_input', 'character', '0.000045')]}) (empty)

275 report = render_report(feed, compare(feed, CATALOG), []) (empty)

276 assert 'DRIFT' in report (empty)

277 assert 'set prices_checked yourself' in report (empty)

278 

279 

280def test_report_says_so_when_nothing_drifted(): (empty)

281 feed = _feed({'id': 'acme-tts-1', 'rates': [_rate('text_input', 'character', '0.000030')]}) (empty)

282 report = render_report(feed, compare(feed, CATALOG), []) (empty)

283 assert 'no drift' in report (empty)

284 

285 

286def test_pricing_feed_url_must_be_https(): (empty)

287 # A polled rate fetched over plaintext can be rewritten in transit, which defeats the point of 

288 # having a trusted source at all. 

289 from prices.prices_types import Provider (empty)

290 

291 base: dict[str, Any] = {'id': 'p', 'name': 'P', 'api_pattern': '.*', 'models': []} (empty)

292 assert Provider.model_validate({**base, 'pricing_feed_url': 'https://api.acme.com/pricing.json'}) (empty)

293 with pytest.raises(ValidationError, match='must be https'): (empty)

294 Provider.model_validate({**base, 'pricing_feed_url': 'http://api.acme.com/pricing.json'}) (empty)

295 

296 

297def test_pricing_feed_url_may_not_embed_credentials(): (empty)

298 # Provider files are public and feed into a published schema, so a URL with a password in it 

299 # would leak it the moment it merged. 

300 from prices.prices_types import Provider (empty)

301 

302 base: dict[str, Any] = {'id': 'p', 'name': 'P', 'api_pattern': '.*', 'models': []} (empty)

303 with pytest.raises(ValidationError, match='must not embed credentials'): (empty)

304 Provider.model_validate({**base, 'pricing_feed_url': 'https://user:pw@api.acme.com/pricing.json'}) (empty)

305 

306 

307def test_pricing_feed_url_allows_a_query_string(): (empty)

308 # Deliberately not enforced: a version marker is a reasonable thing for a provider to serve, and 

309 # refusing it would block a valid registration over a style preference. 

310 from prices.prices_types import Provider (empty)

311 

312 base: dict[str, Any] = {'id': 'p', 'name': 'P', 'api_pattern': '.*', 'models': []} (empty)

313 assert Provider.model_validate({**base, 'pricing_feed_url': 'https://api.acme.com/pricing.json?v=1'}) (empty)

314 

315 

316def test_registered_feeds_reads_the_provider_yaml_registry(): (empty)

317 # The registry is the provider files themselves, so there is no second list to fall out of sync. 

318 from prices.source_feed import registered_feeds (empty)

319 

320 feeds = registered_feeds() (empty)

321 assert all(url.startswith('http') for _, url in feeds) (empty)

322 assert len({provider_id for provider_id, _ in feeds}) == len(feeds) # one endpoint per provider (empty)

323 

324 

325def test_feed_sync_is_a_no_op_until_a_provider_registers_an_endpoint(capsys: pytest.CaptureFixture[str]): (empty)

326 from prices.source_feed import feed_sync, registered_feeds (empty)

327 

328 if registered_feeds(): 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true(empty)

329 pytest.skip('a provider now publishes an endpoint; this test covers the empty registry only') 

330 assert feed_sync() == 0 # nothing registered is not a failure (empty)

331 assert 'pricing_feed_url' in capsys.readouterr().out (empty)

332 

333 

334def test_feed_check_without_a_feed_env_var_explains_itself( (empty)

335 monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] 

336): 

337 monkeypatch.delenv('FEED', raising=False) (empty)

338 assert feed_check() == 1 (empty)

339 assert 'FEED' in capsys.readouterr().out (empty)

340 

341 

342def test_feed_check_exits_non_zero_on_drift(monkeypatch: pytest.MonkeyPatch, tmp_path: Path): (empty)

343 # Telnyx's real STT rate is $0.015/min; publish a different one and the check must fail. 

344 path = tmp_path / 'pricing.json' (empty)

345 path.write_text( (empty)

346 json.dumps( 

347 {'provider': 'telnyx', 'models': [{'id': 'telnyx-stt', 'rates': [_rate('audio_input', 'minute', '0.030')]}]} 

348 ) 

349 ) 

350 monkeypatch.setenv('FEED', str(path)) (empty)

351 assert feed_check() == 1 (empty)

352 

353 

354def test_feed_check_exits_zero_when_the_feed_agrees(monkeypatch: pytest.MonkeyPatch, tmp_path: Path): (empty)

355 path = tmp_path / 'pricing.json' (empty)

356 path.write_text( (empty)

357 json.dumps( 

358 {'provider': 'telnyx', 'models': [{'id': 'telnyx-stt', 'rates': [_rate('audio_input', 'minute', '0.015')]}]} 

359 ) 

360 ) 

361 monkeypatch.setenv('FEED', str(path)) (empty)

362 assert feed_check() == 0 (empty)

363 

364 

365# ---- currency ---------------------------------------------------------------- 

366 

367 

368def test_currency_defaults_to_usd(): (empty)

369 assert Feed.model_validate({'provider': 'acme', 'models': []}).currency == 'USD' (empty)

370 

371 

372@pytest.mark.parametrize('value', ['USD', 'usd', 'Usd']) (empty)

373def test_usd_is_accepted_case_insensitively_and_normalised(value: str): (empty)

374 assert Feed.model_validate({'provider': 'acme', 'currency': value, 'models': []}).currency == 'USD' (empty)

375 

376 

377@pytest.mark.parametrize('value', ['CNY', 'INR', 'EUR', 'KRW', 'RUB', 'GBP']) (empty)

378def test_non_usd_feed_is_rejected(value: str): (empty)

379 """Accepting it would ingest the amounts as dollars and publish a wrong price under the 

380 vendor's own name. Rejecting is recoverable; a quietly mis-scaled rate is not.""" 

381 with pytest.raises(ValidationError) as exc: (empty)

382 Feed.model_validate({'provider': 'acme', 'currency': value, 'models': []}) (empty)

383 assert 'USD only' in str(exc.value) (empty)