Coverage for tests/test_update_prices.py: 96.64%

1import threading (empty)

2from decimal import Decimal (empty)

3 

4import httpx2 (empty)

5import pytest (empty)

6from inline_snapshot import snapshot (empty)

7 

8from voice_prices import ( (empty)

9 UpdatePrices, 

10 Usage, 

11 calc_price, 

12 data_snapshot, 

13 wait_prices_updated_async, 

14 wait_prices_updated_sync, 

15) 

16 

17pytestmark = pytest.mark.anyio (empty)

18 

19PROVIDER_ARRAY_PAYLOAD = ( (empty)

20 b'[{"id":"openai","name":"OpenAI","api_pattern":"https://api\\\\.openai\\\\.com",' 

21 b'"models":[{"id":"gpt-4o","match":{"equals":"gpt-4o"},' 

22 b'"prices":{"input_mtok":2.5,"output_mtok":10}}]}]' 

23) 

24 

25 

26def _mock_update_prices_get(monkeypatch: pytest.MonkeyPatch, content: bytes = PROVIDER_ARRAY_PAYLOAD) -> None: (empty)

27 class Response: (empty)

28 def __init__(self, content: bytes) -> None: (empty)

29 self.content = content (empty)

30 

31 def raise_for_status(self) -> None: (empty)

32 pass (empty)

33 

34 def fake_get(url: str, timeout: httpx2.Timeout) -> Response: (empty)

35 assert url in { (empty)

36 'https://example.test/prices.json', 

37 'https://raw.githubusercontent.com/mahimailabs/voice-prices/refs/heads/main/prices/data.json', 

38 } 

39 assert timeout is not None (empty)

40 return Response(content) (empty)

41 

42 monkeypatch.setattr(httpx2, 'get', fake_get) (empty)

43 

44 

45def test_update_prices_fetch_parses_provider_array(monkeypatch: pytest.MonkeyPatch): (empty)

46 content = ( (empty)

47 b'[{"id":"openai","name":"OpenAI","api_pattern":"https://api\\\\.openai\\\\.com",' 

48 b'"models":[{"id":"gpt-4o-mini","match":{"equals":"gpt-4o-mini"},' 

49 b'"prices":{"input_mtok":0.15,"output_mtok":0.6}}]}]' 

50 ) 

51 

52 _mock_update_prices_get(monkeypatch, content) (empty)

53 

54 snapshot = UpdatePrices(url='https://example.test/prices.json').fetch() (empty)

55 

56 assert snapshot is not None (empty)

57 assert snapshot.from_auto_update is True (empty)

58 provider, model = snapshot.find_provider_model('gpt-4o-mini', None, 'openai', None) (empty)

59 assert provider.id == 'openai' (empty)

60 assert model.id == 'gpt-4o-mini' (empty)

61 

62 

63def test_update_prices_wait_on_start(monkeypatch: pytest.MonkeyPatch): (empty)

64 _mock_update_prices_get(monkeypatch) (empty)

65 assert data_snapshot._custom_snapshot is None (empty)

66 with UpdatePrices() as update_prices: (empty)

67 update_prices.wait() (empty)

68 assert data_snapshot._custom_snapshot is not None (empty)

69 price = calc_price(Usage(input_tokens=1000, output_tokens=100), model_ref='gpt-4o', provider_id='openai') (empty)

70 assert price.input_price == snapshot(Decimal('0.0025')) (empty)

71 assert price.output_price == snapshot(Decimal('0.001')) (empty)

72 assert price.total_price == snapshot(Decimal('0.0035')) (empty)

73 assert price.provider.id == snapshot('openai') (empty)

74 assert price.auto_update_timestamp is not None (empty)

75 

76 

77def test_wait_prices_updated_sync(monkeypatch: pytest.MonkeyPatch): (empty)

78 _mock_update_prices_get(monkeypatch) (empty)

79 assert data_snapshot._custom_snapshot is None (empty)

80 with UpdatePrices(): (empty)

81 wait_prices_updated_sync() (empty)

82 assert data_snapshot._custom_snapshot is not None (empty)

83 price = calc_price(Usage(input_tokens=1000, output_tokens=100), model_ref='gpt-4o', provider_id='openai') (empty)

84 assert price.input_price == snapshot(Decimal('0.0025')) (empty)

85 assert price.output_price == snapshot(Decimal('0.001')) (empty)

86 assert price.total_price == snapshot(Decimal('0.0035')) (empty)

87 assert price.provider.id == snapshot('openai') (empty)

88 assert price.auto_update_timestamp is not None (empty)

89 

90 

91async def test_wait_prices_updated_async(monkeypatch: pytest.MonkeyPatch): (empty)

92 _mock_update_prices_get(monkeypatch) (empty)

93 assert data_snapshot._custom_snapshot is None (empty)

94 with UpdatePrices(): (empty)

95 await wait_prices_updated_async() (empty)

96 assert data_snapshot._custom_snapshot is not None (empty)

97 price = calc_price(Usage(input_tokens=1000, output_tokens=100), model_ref='gpt-4o', provider_id='openai') (empty)

98 assert price.input_price == snapshot(Decimal('0.0025')) (empty)

99 assert price.output_price == snapshot(Decimal('0.001')) (empty)

100 assert price.total_price == snapshot(Decimal('0.0035')) (empty)

101 assert price.provider.id == snapshot('openai') (empty)

102 assert price.auto_update_timestamp is not None (empty)

103 

104 

105def test_update_prices_stop_clears_snapshot_after_in_flight_fetch(monkeypatch: pytest.MonkeyPatch) -> None: (empty)

106 fetch_started = threading.Event() (empty)

107 allow_fetch_return = threading.Event() (empty)

108 stop_errors: list[BaseException] = [] (empty)

109 

110 class Response: (empty)

111 content = PROVIDER_ARRAY_PAYLOAD (empty)

112 

113 def raise_for_status(self) -> None: (empty)

114 pass (empty)

115 

116 def fake_get(url: str, timeout: httpx2.Timeout) -> Response: (empty)

117 assert url == 'https://example.test/prices.json' (empty)

118 assert timeout is not None (empty)

119 fetch_started.set() (empty)

120 assert allow_fetch_return.wait(timeout=5) (empty)

121 return Response() (empty)

122 

123 def stop_update_prices(update_prices: UpdatePrices) -> None: (empty)

124 try: (empty)

125 update_prices.stop() (empty)

126 except BaseException as exc: 

127 stop_errors.append(exc) 

128 

129 monkeypatch.setattr(httpx2, 'get', fake_get) (empty)

130 update_prices = UpdatePrices(url='https://example.test/prices.json', update_interval=3600) (empty)

131 update_prices.start() (empty)

132 try: (empty)

133 assert fetch_started.wait(timeout=5) (empty)

134 

135 stop_thread = threading.Thread(target=stop_update_prices, args=(update_prices,)) (empty)

136 stop_thread.start() (empty)

137 assert update_prices._stop_event.wait(timeout=5) (empty)

138 allow_fetch_return.set() (empty)

139 stop_thread.join(timeout=5) (empty)

140 

141 assert not stop_thread.is_alive() (empty)

142 if stop_errors: 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true(empty)

143 raise stop_errors[0] 

144 assert data_snapshot._custom_snapshot is None (empty)

145 finally: 

146 allow_fetch_return.set() (empty)

147 update_prices.stop() (empty)

148 data_snapshot.set_custom_snapshot(None) (empty)

149 

150 

151@pytest.mark.default_cassette('fail.yaml') (empty)

152@pytest.mark.vcr() (empty)

153def test_update_prices_failed(): (empty)

154 assert data_snapshot._custom_snapshot is None (empty)

155 with UpdatePrices(url='https://demo-endpoints.pydantic.workers.dev/bin?status=404') as update_prices: (empty)

156 with pytest.raises(httpx2.HTTPStatusError): (empty)

157 update_prices.wait() (empty)

158 assert data_snapshot._custom_snapshot is None (empty)

159 

160 

161@pytest.mark.default_cassette('fail.yaml') (empty)

162@pytest.mark.vcr() (empty)

163def test_update_prices_failed_stop(): (empty)

164 assert data_snapshot._custom_snapshot is None (empty)

165 update_prices = UpdatePrices(url='https://demo-endpoints.pydantic.workers.dev/bin?status=404') (empty)

166 update_prices.start() (empty)

167 with pytest.raises(httpx2.HTTPStatusError): (empty)

168 update_prices.stop() (empty)

169 assert data_snapshot._custom_snapshot is None (empty)

170 

171 

172def test_update_prices_multiple(monkeypatch: pytest.MonkeyPatch): (empty)

173 _mock_update_prices_get(monkeypatch) (empty)

174 with UpdatePrices(): (empty)

175 with pytest.raises( (empty)

176 RuntimeError, 

177 match='UpdatePrices global task already started, only one UpdatePrices can be active at a time', 

178 ): 

179 with UpdatePrices(): (empty)

180 pass