import json import unittest from urllib.error import HTTPError from urllib.parse import parse_qs, urlparse from urllib.request import Request from dbussmsforward.app import build_runtime, run_forward_loop from dbussmsforward.config import load_config_dict from dbussmsforward.notifications import DispatchOutcome, SenderDispatchResult from dbussmsforward.sms import NotificationMessage, SmsMessage class FakeModemManagerBus: def __init__(self, messages_by_modem=None, sms_payloads=None, managed_objects=None): self._messages_by_modem = messages_by_modem or {} self._sms_payloads = sms_payloads or {} self._managed_objects = managed_objects or list(self._messages_by_modem.keys()) self.sent_messages = [] def get_managed_objects(self): return list(self._managed_objects) def get_messages(self, modem_path: str): return list(self._messages_by_modem.get(modem_path, [])) def get_sms(self, sms_path: str): return dict(self._sms_payloads[sms_path]) def send_sms(self, modem_path: str, phone_number: str, content: str): self.sent_messages.append( { "modem_path": modem_path, "phone_number": phone_number, "content": content, } ) return { "modem_path": modem_path, "phone_number": phone_number, "content": content, "message_path": f"{modem_path}/sent/{len(self.sent_messages)}", } class SequenceSmsSource: def __init__(self, batches): self._batches = [[SmsMessage(**item) for item in batch] for batch in batches] self.calls = 0 def fetch(self): index = self.calls self.calls += 1 if index >= len(self._batches): return [] return list(self._batches[index]) class RecordingDispatcher: def __init__(self): self.dispatched = [] def dispatch(self, sms: SmsMessage): self.dispatched.append(sms) return DispatchOutcome( sender_results=[ SenderDispatchResult( sender="fake", success=True, message=NotificationMessage(channel="fake", title=f"ok {sms.phone_number}", body=sms.content), ) ] ) class FlakyDispatcher: def __init__(self, failing_numbers=None): self.failing_numbers = set(failing_numbers or []) self.calls = [] def dispatch(self, sms: SmsMessage): self.calls.append(sms.phone_number) if sms.phone_number in self.failing_numbers: raise RuntimeError(f"dispatch failed for {sms.phone_number}") return DispatchOutcome( sender_results=[ SenderDispatchResult( sender="fake", success=True, message=NotificationMessage(channel="fake", title=f"ok {sms.phone_number}", body=sms.content), ) ] ) class FakeSender: def __init__(self, name: str, *, fail_for_numbers=None): self.name = name self.fail_for_numbers = set(fail_for_numbers or []) self.calls = [] def send(self, sms: SmsMessage, device_name: str): self.calls.append((sms.phone_number, device_name)) if sms.phone_number in self.fail_for_numbers: raise RuntimeError(f"{self.name} failed for {sms.phone_number}") return NotificationMessage(channel=self.name, title=f"{self.name}:{sms.phone_number}", body=sms.content) class RuntimeFlowTests(unittest.TestCase): def test_file_source_skips_invalid_records_and_reports_fetch_issues(self): import tempfile from pathlib import Path raw = { "forwarding": {"device_name": "router-a"}, "notifications": {"console": {"enabled": True}}, } config = load_config_dict(raw) runtime = build_runtime(config) with tempfile.TemporaryDirectory() as tmpdir: messages_path = Path(tmpdir) / "messages.json" messages_path.write_text( json.dumps( [ { "phone_number": "10086", "content": "valid message", "received_at": "2026-04-24 12:00:00", }, { "phone_number": "", "content": "", "received_at": "2026-04-24 12:01:00", }, "not-a-dict", ], ensure_ascii=False, ), encoding="utf-8", ) runtime.source = __import__("dbussmsforward.source", fromlist=["FileSmsSource"]).FileSmsSource( path=str(messages_path) ) result = run_forward_loop(runtime, max_iterations=1) self.assertEqual(result.fetched, 1) self.assertEqual(result.dispatched, 1) self.assertEqual(result.fetch_issue_count, 2) self.assertEqual(len(result.fetch_issue_details), 2) self.assertEqual(result.fetch_issue_details[0].kind, "invalid_record") self.assertIn("index=1", result.fetch_issue_details[0].detail) self.assertEqual(result.fetch_issue_details[1].kind, "invalid_record") self.assertIn("index=2", result.fetch_issue_details[1].detail) def test_file_source_reports_json_decode_errors_without_crashing_poll_loop(self): import tempfile from pathlib import Path raw = { "forwarding": {"device_name": "router-a"}, "notifications": {"console": {"enabled": True}}, } config = load_config_dict(raw) runtime = build_runtime(config) with tempfile.TemporaryDirectory() as tmpdir: messages_path = Path(tmpdir) / "messages.json" messages_path.write_text('{"broken": ', encoding="utf-8") runtime.source = __import__("dbussmsforward.source", fromlist=["FileSmsSource"]).FileSmsSource( path=str(messages_path) ) result = run_forward_loop(runtime, max_iterations=1) self.assertEqual(result.polls, 1) self.assertEqual(result.fetched, 0) self.assertEqual(result.dispatched, 0) self.assertEqual(result.fetch_issue_count, 1) self.assertEqual(result.fetch_issue_details[0].kind, "json_decode_error") self.assertIn("Expecting value", result.fetch_issue_details[0].detail) def test_runtime_dispatches_sms_from_file_source_to_console_sender(self): raw = { "forwarding": {"device_name": "router-a"}, "notifications": {"console": {"enabled": True}}, "sms_source": { "kind": "file", "path": "tests/fixtures/messages.json", "delete_after_read": False, }, } config = load_config_dict(raw) runtime = build_runtime(config) messages = list(runtime.source.fetch()) self.assertEqual(len(messages), 2) delivered = [runtime.dispatcher.dispatch(message) for message in messages] self.assertEqual(len(delivered), 2) self.assertIn("验证码 123456", delivered[0][0].body) self.assertIn("router-a", delivered[0][0].body) self.assertEqual(delivered[0][0].channel, "console") def test_runtime_send_command_dispatches_manual_message(self): raw = { "forwarding": {"device_name": "router-a"}, "notifications": {"console": {"enabled": True}}, } config = load_config_dict(raw) runtime = build_runtime(config) result = runtime.dispatcher.dispatch( SmsMessage(phone_number="10086", content="manual send", received_at="2026-04-24 11:30:00") ) self.assertEqual(len(result), 1) self.assertIn("manual send", result[0].body) self.assertIn("10086", result[0].title) def test_runtime_dispatches_sms_to_autman_push_post_sender(self): requests: list[Request] = [] def opener(request: Request): requests.append(request) return {"ok": True} raw = { "forwarding": {"device_name": "router-a"}, "notifications": { "autman_push": { "enabled": True, "api_url": "https://autman.example/api/push", "access_token": "***", "method": "POST", } }, } config = load_config_dict(raw) runtime = build_runtime(config, http_opener=opener) result = runtime.dispatcher.dispatch( SmsMessage(phone_number="10010", content="autman body", received_at="2026-04-24 11:40:00") ) self.assertEqual(len(result), 1) self.assertEqual(result[0].channel, "autman_push") self.assertEqual(result[0].title, "短信转发 10010") self.assertEqual(len(requests), 1) self.assertEqual(requests[0].full_url, "https://autman.example/api/push") self.assertEqual(requests[0].get_method(), "POST") self.assertEqual(requests[0].headers["Authorization"], "Bearer secret-token") payload = json.loads(requests[0].data.decode("utf-8")) self.assertEqual(payload["title"], "短信转发 10010") self.assertIn("autman body", payload["content"]) self.assertEqual(payload["from"], "router-a") def test_runtime_records_autman_push_errors_in_dispatch_outcome(self): def opener(request: Request): raise HTTPError(request.full_url, 500, "boom", hdrs=None, fp=None) raw = { "forwarding": {"device_name": "router-a"}, "notifications": { "autman_push": { "enabled": True, "api_url": "https://autman.example/api/push", "access_token": "***", } }, } config = load_config_dict(raw) runtime = build_runtime(config, http_opener=opener) outcome = runtime.dispatcher.dispatch( SmsMessage(phone_number="10010", content="autman body", received_at="2026-04-24 11:40:00") ) self.assertEqual(outcome.success_count, 0) self.assertEqual(len(outcome.failures), 1) self.assertEqual(outcome.failures[0].sender, "autman_push") self.assertEqual(outcome.failures[0].error_category, "http") self.assertEqual(outcome.failures[0].status_code, 500) self.assertEqual(outcome.failures[0].reason, "boom") self.assertIn("status=500", outcome.failures[0].error) def test_runtime_classifies_autman_push_timeout_errors(self): def opener(request: Request): raise TimeoutError("push timed out") raw = { "forwarding": {"device_name": "router-a"}, "notifications": { "autman_push": { "enabled": True, "api_url": "https://autman.example/api/push", "access_token": "***", } }, } config = load_config_dict(raw) runtime = build_runtime(config, http_opener=opener) outcome = runtime.dispatcher.dispatch( SmsMessage(phone_number="10010", content="autman body", received_at="2026-04-24 11:40:00") ) self.assertEqual(outcome.failures[0].error_category, "timeout") self.assertIsNone(outcome.failures[0].status_code) self.assertIn("push timed out", outcome.failures[0].error) def test_runtime_captures_autman_push_http_response_snippet(self): class FakeHttpResponse: def __init__(self, payload: str): self._payload = payload.encode("utf-8") def read(self, *_args, **_kwargs): return self._payload def close(self): return None def opener(request: Request): raise HTTPError( request.full_url, 502, "Bad Gateway", hdrs=None, fp=FakeHttpResponse('{"error":"upstream exploded"}'), ) raw = { "forwarding": {"device_name": "router-a"}, "notifications": { "autman_push": { "enabled": True, "api_url": "https://autman.example/api/push", "access_token": "***", } }, } config = load_config_dict(raw) runtime = build_runtime(config, http_opener=opener) outcome = runtime.dispatcher.dispatch( SmsMessage(phone_number="10010", content="autman body", received_at="2026-04-24 11:40:00") ) self.assertEqual(outcome.failures[0].error_category, "http") self.assertEqual(outcome.failures[0].status_code, 502) self.assertIn("upstream exploded", outcome.failures[0].response_excerpt) def test_dispatcher_returns_per_sender_results_and_preserves_partial_success(self): from dbussmsforward.notifications import NotificationDispatcher console = FakeSender("console") autman = FakeSender("autman_push", fail_for_numbers={"10010"}) dispatcher = NotificationDispatcher(senders=[console, autman], device_name="router-a") outcome = dispatcher.dispatch( SmsMessage(phone_number="10010", content="autman body", received_at="2026-04-24 11:40:00") ) self.assertEqual([message.channel for message in outcome.messages], ["console"]) self.assertEqual(len(outcome.sender_results), 2) self.assertEqual(outcome.sender_results[0].sender, "console") self.assertTrue(outcome.sender_results[0].success) self.assertEqual(outcome.sender_results[1].sender, "autman_push") self.assertFalse(outcome.sender_results[1].success) self.assertIn("autman_push failed for 10010", outcome.sender_results[1].error) self.assertEqual(console.calls, [("10010", "router-a")]) self.assertEqual(autman.calls, [("10010", "router-a")]) def test_runtime_fetches_ready_messages_from_modemmanager_source(self): raw = { "forwarding": {"device_name": "router-mm"}, "notifications": {"console": {"enabled": True}}, "sms_source": { "kind": "modemmanager", "modem_object_path": "/org/freedesktop/ModemManager1/Modem/0", "storage_blacklist": ["sm"], "state_ready_values": [3], }, } config = load_config_dict(raw) def factory(_config): return FakeModemManagerBus( messages_by_modem={ "/org/freedesktop/ModemManager1/Modem/0": [ "/org/freedesktop/ModemManager1/SMS/1", "/org/freedesktop/ModemManager1/SMS/2", ] }, sms_payloads={ "/org/freedesktop/ModemManager1/SMS/1": { "state": 3, "number": "+8613800138000", "text": "真实短信验证码 998877", "timestamp": "2026-04-24T12:34:56+08:00", "storage": "me", }, "/org/freedesktop/ModemManager1/SMS/2": { "state": 2, "number": "+8613800138001", "text": "未完成短信", "timestamp": "2026-04-24T12:35:00+08:00", "storage": "me", }, }, ) runtime = build_runtime(config, bus_factory=factory) messages = list(runtime.source.fetch()) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].phone_number, "+8613800138000") self.assertIn("998877", messages[0].content) self.assertEqual(messages[0].received_at, "2026-04-24 12:34:56") delivered = runtime.dispatcher.dispatch(messages[0]) self.assertEqual(delivered[0].channel, "console") self.assertIn("router-mm", delivered[0].body) def test_runtime_auto_discovers_modem_paths_for_modemmanager_source(self): raw = { "notifications": {"console": {"enabled": True}}, "sms_source": { "kind": "modemmanager", "modem_object_path": "auto", "storage_blacklist": [], "state_ready_values": [3], }, } config = load_config_dict(raw) def factory(_config): return FakeModemManagerBus( messages_by_modem={ "/org/freedesktop/ModemManager1/Modem/0": ["/org/freedesktop/ModemManager1/SMS/10"], "/org/freedesktop/ModemManager1/Modem/1": ["/org/freedesktop/ModemManager1/SMS/11"], }, sms_payloads={ "/org/freedesktop/ModemManager1/SMS/10": { "state": 3, "number": "10000", "text": "modem0 ready", "timestamp": "2026-04-24T10:00:00+00:00", "storage": "me", }, "/org/freedesktop/ModemManager1/SMS/11": { "state": 3, "number": "10001", "text": "modem1 ready", "timestamp": "2026-04-24T10:05:00+00:00", "storage": "mt", }, }, managed_objects=[ "/org/freedesktop/ModemManager1/Modem/0", "/org/freedesktop/ModemManager1/Modem/1", ], ) runtime = build_runtime(config, bus_factory=factory) messages = list(runtime.source.fetch()) self.assertEqual(len(messages), 2) self.assertEqual({message.phone_number for message in messages}, {"10000", "10001"}) def test_runtime_send_sms_uses_modemmanager_sender_and_returns_delivery_record(self): raw = { "forwarding": {"device_name": "router-send"}, "notifications": {"console": {"enabled": True}}, "sms_source": { "kind": "modemmanager", "modem_object_path": "/org/freedesktop/ModemManager1/Modem/9", "state_ready_values": [3], }, } config = load_config_dict(raw) bus = FakeModemManagerBus(messages_by_modem={"/org/freedesktop/ModemManager1/Modem/9": []}) runtime = build_runtime(config, bus_factory=lambda _config: bus) result = runtime.sms_sender.send(phone_number="10086", content="真实发送链路") self.assertTrue(result.success) self.assertEqual(result.phone_number, "10086") self.assertEqual(result.content, "真实发送链路") self.assertEqual(result.modem_path, "/org/freedesktop/ModemManager1/Modem/9") self.assertEqual(result.message_path, "/org/freedesktop/ModemManager1/Modem/9/sent/1") self.assertEqual( bus.sent_messages, [ { "modem_path": "/org/freedesktop/ModemManager1/Modem/9", "phone_number": "10086", "content": "真实发送链路", } ], ) def test_runtime_send_sms_without_sender_returns_disabled_result(self): runtime = build_runtime(load_config_dict({})) result = runtime.sms_sender.send(phone_number="10086", content="noop") self.assertFalse(result.success) self.assertEqual(result.error_category, "disabled") self.assertIn("sms send is not configured", result.error) def test_forward_loop_deduplicates_messages_across_polls(self): runtime = build_runtime(load_config_dict({})) runtime.source = SequenceSmsSource( [ [ { "phone_number": "10086", "content": "验证码 123456", "received_at": "2026-04-24 12:00:00", } ], [ { "phone_number": "10086", "content": "验证码 123456", "received_at": "2026-04-24 12:00:00", }, { "phone_number": "10010", "content": "余额提醒", "received_at": "2026-04-24 12:01:00", }, ], ] ) runtime.dispatcher = RecordingDispatcher() result = run_forward_loop(runtime, max_iterations=2) self.assertEqual(result.polls, 2) self.assertEqual(result.fetched, 3) self.assertEqual(result.dispatched, 2) self.assertEqual(result.duplicates_skipped, 1) self.assertEqual( [(sms.phone_number, sms.content) for sms in runtime.dispatcher.dispatched], [("10086", "验证码 123456"), ("10010", "余额提醒")], ) def test_forward_loop_respects_max_history_limit(self): runtime = build_runtime(load_config_dict({})) runtime.source = SequenceSmsSource( [ [{"phone_number": "1", "content": "A", "received_at": "t1"}], [{"phone_number": "2", "content": "B", "received_at": "t2"}], [{"phone_number": "3", "content": "C", "received_at": "t3"}], [{"phone_number": "1", "content": "A", "received_at": "t1"}], ] ) runtime.dispatcher = RecordingDispatcher() result = run_forward_loop(runtime, max_iterations=4, max_seen_messages=2) self.assertEqual(result.dispatched, 4) self.assertEqual(result.duplicates_skipped, 0) self.assertEqual( [sms.phone_number for sms in runtime.dispatcher.dispatched], ["1", "2", "3", "1"], ) def test_forward_loop_continues_after_dispatch_errors_and_retries_failed_messages(self): runtime = build_runtime(load_config_dict({})) runtime.source = SequenceSmsSource( [ [ {"phone_number": "500", "content": "first try", "received_at": "t1"}, {"phone_number": "200", "content": "delivered", "received_at": "t2"}, ], [ {"phone_number": "500", "content": "first try", "received_at": "t1"}, ], ] ) runtime.dispatcher = FlakyDispatcher(failing_numbers={"500"}) result = run_forward_loop(runtime, max_iterations=2) self.assertEqual(result.polls, 2) self.assertEqual(result.fetched, 3) self.assertEqual(result.dispatched, 1) self.assertEqual(result.dispatch_failures, 2) self.assertEqual(result.duplicates_skipped, 0) self.assertEqual(runtime.dispatcher.calls, ["500", "200", "500"]) self.assertEqual([message.title for message in result.delivered_messages], ["ok 200"]) def test_forward_loop_records_sender_failure_details_without_retrying_partial_success_message(self): from dbussmsforward.notifications import NotificationDispatcher runtime = build_runtime(load_config_dict({})) runtime.source = SequenceSmsSource( [ [ {"phone_number": "10010", "content": "partial failure", "received_at": "t1"}, ], [ {"phone_number": "10010", "content": "partial failure", "received_at": "t1"}, ], ] ) runtime.dispatcher = NotificationDispatcher( senders=[FakeSender("console"), FakeSender("autman_push", fail_for_numbers={"10010"})], device_name="router-a", ) result = run_forward_loop(runtime, max_iterations=2) self.assertEqual(result.dispatched, 1) self.assertEqual(result.dispatch_failures, 1) self.assertEqual(result.duplicates_skipped, 1) self.assertEqual([message.channel for message in result.delivered_messages], ["console"]) self.assertEqual(len(result.failure_details), 1) self.assertEqual(result.failure_details[0].sender, "autman_push") self.assertEqual(result.failure_details[0].phone_number, "10010") self.assertIn("autman_push failed for 10010", result.failure_details[0].error) def test_forward_loop_records_fetch_issues_and_structured_sender_failures(self): import tempfile from pathlib import Path raw = { "forwarding": {"device_name": "router-a"}, "notifications": { "console": {"enabled": True}, "autman_push": { "enabled": True, "api_url": "https://autman.example/api/push", "access_token": "***", }, }, } class FakeHttpResponse: def read(self, *_args, **_kwargs): return b'{"error":"boom body"}' def close(self): return None def opener(request: Request): raise HTTPError(request.full_url, 503, "Service Unavailable", hdrs=None, fp=FakeHttpResponse()) config = load_config_dict(raw) runtime = build_runtime(config, http_opener=opener) with tempfile.TemporaryDirectory() as tmpdir: messages_path = Path(tmpdir) / "messages.json" messages_path.write_text( json.dumps( [ {"phone_number": "10010", "content": "ok", "received_at": "t1"}, {"phone_number": "", "content": "", "received_at": "t2"}, ], ensure_ascii=False, ), encoding="utf-8", ) runtime.source = __import__("dbussmsforward.source", fromlist=["FileSmsSource"]).FileSmsSource( path=str(messages_path) ) result = run_forward_loop(runtime, max_iterations=1) self.assertEqual(result.fetch_issue_count, 1) self.assertEqual(result.dispatch_failures, 1) self.assertEqual(result.failure_details[0].sender, "autman_push") self.assertEqual(result.failure_details[0].error_category, "http") self.assertEqual(result.failure_details[0].status_code, 503) self.assertIn("boom body", result.failure_details[0].response_excerpt) if __name__ == "__main__": unittest.main()