|
1 | 1 | """Tests for _post_webhook, the _exec_on_change error branches, the |
2 | | -_check_stale webhook fan-out, and diff_against_baseline. |
| 2 | +_check_stale webhook fan-out, diff_against_baseline, and a handful of small |
| 3 | +pure helpers (_write_heartbeat's OSError branch, _redact_webhook_url's |
| 4 | +exception branch, load_key, and the no-arg branch of |
| 5 | +_seconds_to_next_midnight_utc) that were still gaps after the first pass. |
3 | 6 |
|
4 | 7 | These paths are exercised only on the happy path (or not at all) elsewhere: |
5 | 8 | test_security.py drives _exec_on_change's rc=0 success path to lock in the |
6 | 9 | env-var transport contract, but never its failure branches; test_watchdog.py |
7 | | -drives _check_stale with webhook_urls=None, never the alert fan-out; and |
8 | | -nothing calls _post_webhook or diff_against_baseline directly. Kept pure/ |
| 10 | +drives _check_stale with webhook_urls=None, never the alert fan-out, and |
| 11 | +covers _write_heartbeat/_read_heartbeat/_format_wedge_payload's normal |
| 12 | +paths but not _write_heartbeat's OSError branch; test_outage_backoff.py |
| 13 | +covers _seconds_to_next_midnight_utc but always passes an explicit `now`, |
| 14 | +never exercising the `now is None -> time.time()` branch; and nothing calls |
| 15 | +_post_webhook, diff_against_baseline, or load_key directly. Kept pure/ |
9 | 16 | offline — urllib.request.urlopen and subprocess.run are mocked, never called |
10 | 17 | for real. |
11 | 18 |
|
|
26 | 33 | _check_stale, |
27 | 34 | _exec_on_change, |
28 | 35 | _post_webhook, |
| 36 | + _read_heartbeat, |
| 37 | + _redact_webhook_url, |
| 38 | + _seconds_to_next_midnight_utc, |
| 39 | + _write_heartbeat, |
29 | 40 | diff_against_baseline, |
| 41 | + load_key, |
30 | 42 | ) |
31 | 43 |
|
32 | 44 |
|
@@ -264,5 +276,73 @@ def test_unchanged_result_produces_no_diff(self) -> None: |
264 | 276 | self.assertEqual(diffs, []) |
265 | 277 |
|
266 | 278 |
|
| 279 | +class TestWriteHeartbeatOSError(unittest.TestCase): |
| 280 | + def test_write_failure_is_caught_and_logged(self) -> None: |
| 281 | + # path.with_suffix()/write_text()/replace() all live on the real |
| 282 | + # Path object; patch Path.write_text so the write itself raises, |
| 283 | + # proving the OSError branch swallows the error instead of |
| 284 | + # propagating it and killing the watch loop. |
| 285 | + with tempfile.TemporaryDirectory() as d: |
| 286 | + hb = Path(d) / "hb.json" |
| 287 | + with mock.patch("pathlib.Path.write_text", |
| 288 | + side_effect=OSError("disk full")): |
| 289 | + with self.assertLogs(level="WARNING") as cm: |
| 290 | + _write_heartbeat(hb, "HEALTHY", 42, "ok") |
| 291 | + self.assertIn("heartbeat write failed", " ".join(cm.output)) |
| 292 | + # No partial/garbage file left behind by the failed write. |
| 293 | + self.assertFalse(hb.exists()) |
| 294 | + |
| 295 | + |
| 296 | +class TestRedactWebhookUrlExceptionBranch(unittest.TestCase): |
| 297 | + def test_urlparse_raising_falls_back_to_unparseable(self) -> None: |
| 298 | + # The existing edge-case tests (empty string, "not a url", etc.) |
| 299 | + # never actually hit the except branch because urllib.parse is |
| 300 | + # lenient about garbage strings. Force the branch directly. |
| 301 | + with mock.patch("wdgwars_api_tester.urllib.parse.urlparse", |
| 302 | + side_effect=ValueError("boom")): |
| 303 | + out = _redact_webhook_url("https://example.com/hook") |
| 304 | + self.assertEqual(out, "<unparseable-url>") |
| 305 | + |
| 306 | + |
| 307 | +class TestLoadKey(unittest.TestCase): |
| 308 | + def test_cli_key_takes_priority(self) -> None: |
| 309 | + self.assertEqual(load_key(" cli-key "), "cli-key") |
| 310 | + |
| 311 | + def test_env_var_used_when_no_cli_key(self) -> None: |
| 312 | + with mock.patch.dict("os.environ", |
| 313 | + {"WDGWARS_API_KEY": " env-key "}, clear=False): |
| 314 | + self.assertEqual(load_key(None), "env-key") |
| 315 | + |
| 316 | + def test_config_file_used_when_no_cli_or_env(self) -> None: |
| 317 | + with tempfile.TemporaryDirectory() as d: |
| 318 | + fake_home = Path(d) |
| 319 | + cfg_dir = fake_home / ".config" / "wigle-to-wdgwars" |
| 320 | + cfg_dir.mkdir(parents=True) |
| 321 | + (cfg_dir / "wdgwars.key").write_text(" file-key \n", encoding="utf-8") |
| 322 | + with mock.patch.dict("os.environ", {}, clear=True): |
| 323 | + with mock.patch("wdgwars_api_tester.Path.home", |
| 324 | + return_value=fake_home): |
| 325 | + self.assertEqual(load_key(None), "file-key") |
| 326 | + |
| 327 | + def test_returns_none_when_nothing_configured(self) -> None: |
| 328 | + with tempfile.TemporaryDirectory() as d: |
| 329 | + fake_home = Path(d) # no .config/wigle-to-wdgwars/wdgwars.key |
| 330 | + with mock.patch.dict("os.environ", {}, clear=True): |
| 331 | + with mock.patch("wdgwars_api_tester.Path.home", |
| 332 | + return_value=fake_home): |
| 333 | + self.assertIsNone(load_key(None)) |
| 334 | + |
| 335 | + |
| 336 | +class TestSecondsToNextMidnightDefaultNow(unittest.TestCase): |
| 337 | + def test_no_arg_uses_current_time(self) -> None: |
| 338 | + # test_outage_backoff.py always passes an explicit `now`; this |
| 339 | + # covers the `now is None -> time.time()` branch specifically. |
| 340 | + fixed = 1_800_000_000.0 # arbitrary fixed epoch, not near midnight |
| 341 | + with mock.patch("wdgwars_api_tester.time.time", return_value=fixed): |
| 342 | + via_default = _seconds_to_next_midnight_utc() |
| 343 | + via_explicit = _seconds_to_next_midnight_utc(fixed) |
| 344 | + self.assertEqual(via_default, via_explicit) |
| 345 | + |
| 346 | + |
267 | 347 | if __name__ == "__main__": |
268 | 348 | unittest.main() |
0 commit comments