1"""
2Tests for fetcher.py and PPE charge safety.
3
4fetcher.py makes real network calls via httpx. All tests mock the HTTP
5client so no network is needed.
6"""
7
8import pytest
9from unittest.mock import MagicMock, patch
10
11import httpx
12
13from fetcher import (
14 _build_payload,
15 fetch_statement_html,
16 MOPS_BASE,
17 REDIRECT_ENDPOINT,
18)
19from parser import gregorian_to_roc
20
21
22
23
24
25
26def _make_mock_client(redirect_url: str, html_body: str) -> MagicMock:
27 """Return a mock httpx.Client that simulates the 2-step MOPS flow."""
28 client = MagicMock()
29
30
31 r1 = MagicMock()
32 r1.json.return_value = {"url": redirect_url}
33 r1.raise_for_status = MagicMock()
34
35
36 r2 = MagicMock()
37 r2.text = html_body
38 r2.raise_for_status = MagicMock()
39
40 client.post.return_value = r1
41 client.get.return_value = r2
42 return client
43
44
45
46
47
48
49class TestBuildPayload:
50 def test_income_statement_payload(self):
51 payload = _build_payload("income", 2024, 4, "listed")
52 assert payload["apiName"] == "ajax_t163sb04"
53 assert payload["parameters"]["TYPEK"] == "sii"
54 assert payload["parameters"]["year"] == str(gregorian_to_roc(2024))
55 assert payload["parameters"]["season"] == "04"
56
57 def test_balance_sheet_payload(self):
58 payload = _build_payload("balance", 2024, 1, "otc")
59 assert payload["apiName"] == "ajax_t163sb05"
60 assert payload["parameters"]["TYPEK"] == "otc"
61 assert payload["parameters"]["season"] == "01"
62
63 def test_cashflow_payload(self):
64 payload = _build_payload("cashflow", 2023, 3, "listed")
65 assert payload["apiName"] == "ajax_t163sb20"
66 assert payload["parameters"]["year"] == str(gregorian_to_roc(2023))
67
68 def test_season_zero_padded(self):
69 payload = _build_payload("income", 2024, 1, "listed")
70 assert payload["parameters"]["season"] == "01"
71
72 def test_season_two_digits_preserved(self):
73 payload = _build_payload("income", 2024, 4, "listed")
74 assert payload["parameters"]["season"] == "04"
75
76 def test_roc_year_conversion(self):
77 payload = _build_payload("income", 2024, 4, "listed")
78 assert payload["parameters"]["year"] == "113"
79
80 def test_payload_has_required_keys(self):
81 payload = _build_payload("income", 2024, 4, "listed")
82 params = payload["parameters"]
83 for key in ("encodeURIComponent", "firstin", "off", "step", "isQuery", "TYPEK", "year", "season"):
84 assert key in params, f"Missing parameter key: {key}"
85
86
87
88
89
90
91class TestFetchStatementHtml:
92 def test_returns_html_from_step2(self):
93 mock_client = _make_mock_client(
94 redirect_url="https://mops.twse.com.tw/mops/web/ajax_t163sb04?signed=abc",
95 html_body="<html><body><table>...</table></body></html>",
96 )
97 result = fetch_statement_html("income", 2024, 4, "listed", client=mock_client)
98 assert "<table>" in result
99
100 def test_posts_to_redirect_endpoint(self):
101 mock_client = _make_mock_client(
102 redirect_url="https://mops.twse.com.tw/data",
103 html_body="<html></html>",
104 )
105 fetch_statement_html("income", 2024, 4, "listed", client=mock_client)
106 mock_client.post.assert_called_once()
107 call_url = mock_client.post.call_args[0][0]
108 assert call_url == REDIRECT_ENDPOINT
109
110 def test_gets_redirect_url(self):
111 redirect_url = "https://mops.twse.com.tw/mops/web/ajax_t163sb04?signed=xyz"
112 mock_client = _make_mock_client(redirect_url=redirect_url, html_body="<html></html>")
113 fetch_statement_html("income", 2024, 4, "listed", client=mock_client)
114 mock_client.get.assert_called_once_with(redirect_url)
115
116 def test_relative_redirect_url_gets_base_prepended(self):
117 mock_client = _make_mock_client(
118 redirect_url="/mops/web/ajax_t163sb04?signed=xyz",
119 html_body="<html></html>",
120 )
121 fetch_statement_html("income", 2024, 4, "listed", client=mock_client)
122 called_url = mock_client.get.call_args[0][0]
123 assert called_url.startswith(MOPS_BASE)
124
125 def test_raises_on_missing_redirect_url(self):
126 client = MagicMock()
127 r1 = MagicMock()
128 r1.json.return_value = {"status": "ok"}
129 r1.raise_for_status = MagicMock()
130 client.post.return_value = r1
131 with pytest.raises(ValueError, match="redirect URL"):
132 fetch_statement_html("income", 2024, 4, "listed", client=client)
133
134 def test_raises_on_http_error_step1(self):
135 client = MagicMock()
136 r1 = MagicMock()
137 r1.raise_for_status.side_effect = httpx.HTTPStatusError(
138 "503", request=MagicMock(), response=MagicMock()
139 )
140 client.post.return_value = r1
141 with pytest.raises(httpx.HTTPStatusError):
142 fetch_statement_html("income", 2024, 4, "listed", client=client)
143
144 def test_raises_on_http_error_step2(self):
145 client = MagicMock()
146 r1 = MagicMock()
147 r1.json.return_value = {"url": "https://mops.twse.com.tw/data"}
148 r1.raise_for_status = MagicMock()
149 r2 = MagicMock()
150 r2.raise_for_status.side_effect = httpx.HTTPStatusError(
151 "404", request=MagicMock(), response=MagicMock()
152 )
153 client.post.return_value = r1
154 client.get.return_value = r2
155 with pytest.raises(httpx.HTTPStatusError):
156 fetch_statement_html("income", 2024, 4, "listed", client=client)
157
158 def test_accepts_redirecturl_key_variant(self):
159 """MOPS might return 'redirectUrl' instead of 'url'."""
160 client = MagicMock()
161 r1 = MagicMock()
162 r1.json.return_value = {"redirectUrl": "https://mops.twse.com.tw/data"}
163 r1.raise_for_status = MagicMock()
164 r2 = MagicMock()
165 r2.text = "<html>data</html>"
166 r2.raise_for_status = MagicMock()
167 client.post.return_value = r1
168 client.get.return_value = r2
169 result = fetch_statement_html("income", 2024, 4, "listed", client=client)
170 assert "data" in result
171
172
173
174
175
176
177class TestPPEChargeSafety:
178 def _load_main_source(self) -> str:
179 with open("src/main.py") as f:
180 return f.read()
181
182 def test_exactly_one_ppe_charge_call(self):
183 source = self._load_main_source()
184 charge_calls = [
185 line.strip()
186 for line in source.splitlines()
187 if "push_data" in line and "data-fetched" in line
188 ]
189 assert len(charge_calls) == 1, (
190 f"Expected exactly 1 PPE charge call, found {len(charge_calls)}: {charge_calls}"
191 )
192
193 def test_charge_fires_only_when_successful(self):
194 source = self._load_main_source()
195 assert "successful_types" in source, (
196 "main.py must guard PPE charge with a successful count/list"
197 )
198
199 def test_free_push_exists_for_all_errors_case(self):
200 source = self._load_main_source()
201 lines = source.splitlines()
202 free_push_lines = [
203 l.strip() for l in lines
204 if "push_data" in l and "data-fetched" not in l
205 ]
206 assert len(free_push_lines) >= 1, (
207 "main.py must have a push_data(output) call without PPE charge"
208 )
209
210 def test_event_name_is_data_fetched(self):
211 source = self._load_main_source()
212 assert '"data-fetched"' in source, (
213 "PPE event name must be 'data-fetched'"
214 )
215
216 def test_two_push_data_calls_total(self):
217 source = self._load_main_source()
218 push_calls = [l.strip() for l in source.splitlines() if "push_data" in l]
219 assert len(push_calls) == 2, (
220 f"Expected exactly 2 push_data calls, found {len(push_calls)}: {push_calls}"
221 )
222
223 def test_actor_fail_called_on_invalid_input(self):
224 source = self._load_main_source()
225 assert "Actor.fail" in source, "main.py must call Actor.fail() for invalid inputs"
226
227 def test_multiple_actor_fail_guards(self):
228 source = self._load_main_source()
229 fail_calls = [l.strip() for l in source.splitlines() if "Actor.fail" in l]
230 assert len(fail_calls) >= 3, (
231 f"Expected at least 3 Actor.fail() calls (one per validation), "
232 f"found {len(fail_calls)}: {fail_calls}"
233 )
234
235 def test_exchange_validated(self):
236 source = self._load_main_source()
237 assert "TYPEK_MAP" in source, "main.py must validate exchange against TYPEK_MAP"
238
239 def test_season_validated(self):
240 source = self._load_main_source()
241 assert "season not in (1, 2, 3, 4)" in source or "1, 2, 3, 4" in source, (
242 "main.py must validate season is 1–4"
243 )
244
245 def test_year_validated(self):
246 source = self._load_main_source()
247 assert "1912" in source, "main.py must validate year >= 1912"
248
249 def test_statement_types_validated_against_valid_list(self):
250 source = self._load_main_source()
251 assert "VALID_STATEMENT_TYPES" in source, (
252 "main.py must validate statement_types against VALID_STATEMENT_TYPES"
253 )
254
255 def test_no_charge_in_else_branch(self):
256 source = self._load_main_source()
257 lines = source.splitlines()
258 in_else = False
259 for line in lines:
260 stripped = line.strip()
261 if stripped == "else:":
262 in_else = True
263 continue
264 if in_else:
265 if stripped and not line.startswith(" "):
266 break
267 assert "data-fetched" not in stripped, (
268 f"'data-fetched' found inside else branch: {stripped}"
269 )
