#!/usr/bin/env python3 import importlib.util import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock PROJECT_ROOT = Path(__file__).resolve().parents[1] MODULE_PATH = PROJECT_ROOT / "wechat_health_watch.py" SPEC = importlib.util.spec_from_file_location("wechat_health_watch", MODULE_PATH) MODULE = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None sys.modules[SPEC.name] = MODULE SPEC.loader.exec_module(MODULE) class TransitionTests(unittest.TestCase): def test_incident_requires_two_consecutive_unhealthy_scans(self): state = {} self.assertIsNone( MODULE.update_incident( state, codes=["recorder_stale"], now=100.0, ) ) self.assertEqual(state["wechat_chain"]["consecutive_failures"], 1) self.assertEqual( MODULE.update_incident( state, codes=["recorder_stale"], now=160.0, ), "incident", ) self.assertTrue(state["wechat_chain"]["active"]) self.assertTrue(state["wechat_chain"]["incident_pending"]) def test_healthy_scan_resets_the_consecutive_failure_counter(self): state = {"wechat_chain": {"consecutive_failures": 1}} self.assertIsNone(MODULE.update_incident(state, codes=[], now=100.0)) self.assertEqual(state["wechat_chain"]["consecutive_failures"], 0) self.assertIsNone( MODULE.update_incident( state, codes=["wechat_window"], now=160.0, ) ) def test_recovery_is_immediate_after_a_sent_incident(self): state = { "wechat_chain": { "active": True, "incident_sent": True, "incident_last_notification_attempt_at": 100.0, "consecutive_failures": 2, } } self.assertEqual( MODULE.update_incident(state, codes=[], now=101.0), "recovery", ) def test_failed_incident_notification_retries_after_cooldown(self): state = { "wechat_chain": { "active": True, "incident_pending": True, "incident_last_notification_attempt_at": 100.0, } } self.assertIsNone( MODULE.update_incident( state, codes=["recorder_stale"], now=100.0 + MODULE.NOTIFICATION_COOLDOWN_SECONDS - 1, ) ) self.assertEqual( MODULE.update_incident( state, codes=["recorder_stale"], now=100.0 + MODULE.NOTIFICATION_COOLDOWN_SECONDS, ), "incident", ) class RecorderStateTests(unittest.TestCase): def test_missing_recorder_state_is_unhealthy(self): with tempfile.TemporaryDirectory() as directory, mock.patch.object( MODULE, "RECORDER_STATE", Path(directory) / "missing.json", ): self.assertEqual( MODULE.recorder_is_fresh(100.0), (False, "recorder_state"), ) def test_stale_recorder_state_is_unhealthy(self): with tempfile.TemporaryDirectory() as directory: state_path = Path(directory) / "recorder-state.json" state_path.write_text( '{"last_capture_at":"1970-01-01T00:00:00+00:00"}', encoding="utf-8", ) with mock.patch.object(MODULE, "RECORDER_STATE", state_path): self.assertEqual( MODULE.recorder_is_fresh( MODULE.RECORDER_FRESHNESS_SECONDS + 1 ), (False, "recorder_stale"), ) def test_fresh_recorder_state_is_healthy(self): with tempfile.TemporaryDirectory() as directory: state_path = Path(directory) / "recorder-state.json" state_path.write_text( '{"last_capture_at":"1970-01-01T00:00:00+00:00"}', encoding="utf-8", ) with mock.patch.object(MODULE, "RECORDER_STATE", state_path): self.assertEqual( MODULE.recorder_is_fresh( MODULE.RECORDER_FRESHNESS_SECONDS ), (True, ""), ) class HealthCodeTests(unittest.TestCase): def test_login_sized_window_is_not_accepted_as_ready(self): with mock.patch.object( MODULE, "is_active", return_value=True, ), mock.patch.object( MODULE, "is_user_active", return_value=True, ), mock.patch.object( MODULE, "wechat_process_present", return_value=True, ), mock.patch.object( MODULE, "weixin_window_geometry", return_value=(280, 380), ), mock.patch.object( MODULE, "recorder_is_fresh", return_value=(True, ""), ), mock.patch.object( MODULE, "has_unacknowledged_failed_delivery", return_value=False, ): self.assertEqual( MODULE.health_codes(100.0), ["wechat_login_required"], ) def test_missing_window_is_reported(self): with mock.patch.object( MODULE, "is_active", return_value=True, ), mock.patch.object( MODULE, "is_user_active", return_value=True, ), mock.patch.object( MODULE, "wechat_process_present", return_value=True, ), mock.patch.object( MODULE, "weixin_window_geometry", return_value=None, ), mock.patch.object( MODULE, "recorder_is_fresh", return_value=(True, ""), ), mock.patch.object( MODULE, "has_unacknowledged_failed_delivery", return_value=False, ): self.assertEqual(MODULE.health_codes(100.0), ["wechat_window"]) def test_failed_reply_is_reported_as_an_incident(self): with mock.patch.object( MODULE, "is_active", return_value=True, ), mock.patch.object( MODULE, "is_user_active", return_value=True, ), mock.patch.object( MODULE, "wechat_process_present", return_value=True, ), mock.patch.object( MODULE, "weixin_window_geometry", return_value=(1470, 694), ), mock.patch.object( MODULE, "recorder_is_fresh", return_value=(True, ""), ), mock.patch.object( MODULE, "has_unacknowledged_failed_delivery", return_value=True, ): self.assertEqual(MODULE.health_codes(100.0), ["responder_failed"]) class ConfigurationTests(unittest.TestCase): def test_notification_adapter_is_loaded_from_generated_config(self): with tempfile.TemporaryDirectory() as directory: config_file = Path(directory) / "notification.json" config_file.write_text( '{"enabled":true,"command":["/usr/bin/true"],' '"timeout_seconds":15}', encoding="utf-8", ) with mock.patch.object( MODULE, "NOTIFICATION_CONFIG", config_file, ): self.assertEqual( MODULE.notification_config(), { "enabled": True, "command": ["/usr/bin/true"], "timeout_seconds": 15, }, ) def test_notification_message_is_sent_on_stdin_without_shell(self): with mock.patch.object( MODULE, "notification_config", return_value={ "enabled": True, "command": ["/usr/local/bin/notifier", "--json"], "timeout_seconds": 20, }, ), mock.patch.object( MODULE, "run", return_value=subprocess.CompletedProcess([], 0, "", ""), ) as run: MODULE.send_official_notification("health message") run.assert_called_once_with( ["/usr/local/bin/notifier", "--json"], timeout=20, data="health message", ) class NotificationFailureTests(unittest.TestCase): def test_timeout_is_accepted_as_an_uncertain_incident(self): state = {"wechat_chain": {"incident_pending": True}} with mock.patch.object( MODULE, "send_official_notification", side_effect=MODULE.NotificationDeliveryUncertain("delivery_timeout"), ): self.assertTrue( MODULE.finish_notification( state, "incident", codes=["wechat_window"], start_result="not_needed", now=100.0, dry_run=False, ) ) self.assertEqual(state["wechat_chain"]["last_notification_result"], "uncertain") self.assertFalse(state["wechat_chain"]["incident_pending"]) self.assertTrue(state["wechat_chain"]["incident_sent"]) def test_remote_error_code_has_priority(self): result = subprocess.CompletedProcess( [], 75, "", "delivery_error=provider_context_rejected", ) self.assertEqual( MODULE.notification_failure_code(result), "provider_context_rejected", ) def test_prepare_failure_is_sanitized(self): result = subprocess.CompletedProcess( [], 1, "", "sendMessage ret=-2 errmsg=prepare failed", ) self.assertEqual( MODULE.notification_failure_code(result), "provider_prepare_failed", ) def test_context_failure_is_sanitized(self): result = subprocess.CompletedProcess( [], 1, "", "context_token expired", ) self.assertEqual( MODULE.notification_failure_code(result), "provider_context_rejected", ) if __name__ == "__main__": unittest.main()