import json import tempfile import unittest from pathlib import Path from dbussmsforward.config import load_config_dict from dbussmsforward.sms import NotificationMessage, SmsMessage from dbussmsforward.web import create_web_app class RecordingDispatcher: def __init__(self): self.messages = [] def dispatch(self, sms: SmsMessage): self.messages.append(sms) return [ NotificationMessage( channel="console", title=f"短信转发 {sms.phone_number}", body=f"{sms.content}", ) ] class WebProductTests(unittest.TestCase): def test_sms_send_response_shape_is_consistent_across_delivery_modes(self): class RealSmsSender: def send(self, phone_number: str, content: str): from types import SimpleNamespace return SimpleNamespace( success=True, phone_number=phone_number, content=content, modem_path="/org/freedesktop/ModemManager1/Modem/0", message_path="/org/freedesktop/ModemManager1/SMS/1", error=None, error_category=None, ) app = create_web_app(load_config_dict({})) app.runtime.dispatcher = RecordingDispatcher() app.runtime.sms_sender = RealSmsSender() dispatch_response = app.handle_request( "POST", "/api/sms/send", body=json.dumps({"phone_number": "10086", "content": "dispatch ok", "delivery_mode": "dispatch"}).encode("utf-8"), headers={"Content-Type": "application/json"}, ) real_sms_response = app.handle_request( "POST", "/api/sms/send", body=json.dumps({"phone_number": "10010", "content": "real sms ok", "delivery_mode": "real_sms"}).encode("utf-8"), headers={"Content-Type": "application/json"}, ) dispatch_payload = json.loads(dispatch_response["body"]) real_sms_payload = json.loads(real_sms_response["body"]) self.assertEqual(dispatch_response["status"], 200) self.assertEqual(real_sms_response["status"], 200) for payload in (dispatch_payload, real_sms_payload): self.assertEqual(payload["ok"], True) self.assertIn(payload["delivery_mode"], {"dispatch", "real_sms"}) self.assertIn("status", payload) self.assertIn("phone_number", payload) self.assertIn("received_at", payload) self.assertIn("notification_count", payload) self.assertIn("message_path", payload) self.assertIn("modem_path", payload) self.assertIn("error_category", payload) self.assertIn("detail", payload) self.assertEqual(dispatch_payload["notification_count"], 1) self.assertEqual(dispatch_payload["message_path"], None) self.assertEqual(dispatch_payload["modem_path"], None) self.assertEqual(real_sms_payload["notification_count"], 1) self.assertEqual(real_sms_payload["message_path"], "/org/freedesktop/ModemManager1/SMS/1") app = create_web_app( load_config_dict( { "forwarding": {"device_name": "router-finish"}, "notifications": {"console": {"enabled": True}}, } ) ) app.runtime.dispatcher = RecordingDispatcher() send_response = app.handle_request( "POST", "/api/sms/send", body=json.dumps({"phone_number": "10010", "content": "hello dashboard"}).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(send_response["status"], 200) dashboard = app.handle_request("GET", "/api/dashboard") dashboard_payload = json.loads(dashboard["body"]) self.assertEqual(dashboard["status"], 200) self.assertEqual(dashboard_payload["summary"]["manual_send_count"], 1) self.assertEqual(dashboard_payload["summary"]["event_count"], 1) self.assertEqual(dashboard_payload["summary"]["last_sms_phone_number"], "10010") self.assertTrue(dashboard_payload["latest_events"]) events = app.handle_request("GET", "/api/events") events_payload = json.loads(events["body"]) self.assertEqual(events["status"], 200) self.assertEqual(len(events_payload["events"]), 1) self.assertEqual(events_payload["events"][0]["phone_number"], "10010") self.assertEqual(events_payload["events"][0]["notification_count"], 1) self.assertEqual(events_payload["events"][0]["result"], "success") self.assertEqual(events_payload["events"][0]["delivery_mode"], "dispatch") self.assertEqual(events_payload["events"][0]["error_category"], None) def test_dashboard_reports_send_failures(self): class FailingDispatcher: def dispatch(self, sms: SmsMessage): raise RuntimeError("dispatch exploded") app = create_web_app(load_config_dict({})) app.runtime.dispatcher = FailingDispatcher() response = app.handle_request( "POST", "/api/sms/send", body=json.dumps({"phone_number": "10000", "content": "boom"}).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(response["status"], 500) payload = json.loads(response["body"]) self.assertEqual(payload["error"], "dispatch_failed") self.assertEqual(payload["ok"], False) self.assertEqual(payload["status"], "failed") self.assertEqual(payload["phone_number"], "10000") dashboard = json.loads(app.handle_request("GET", "/api/dashboard")["body"]) self.assertEqual(dashboard["summary"]["failure_count"], 1) self.assertEqual(dashboard["summary"]["event_count"], 1) self.assertEqual(dashboard["latest_events"][0]["result"], "failed") self.assertEqual(dashboard["latest_events"][0]["delivery_mode"], "dispatch") self.assertEqual(dashboard["latest_events"][0]["error_category"], "dispatch_failed") def test_static_index_contains_dashboard_sections(self): app = create_web_app(load_config_dict({})) index_response = app.handle_request("GET", "/") body = index_response["body"] self.assertEqual(index_response["status"], 200) self.assertIn("运行总览", body) self.assertIn("最近事件", body) self.assertIn("配置中心", body) self.assertIn("短信测试台", body) def test_asset_bundle_contains_dashboard_rendering_logic(self): app = create_web_app(load_config_dict({})) js_response = app.handle_request("GET", "/app.js") self.assertEqual(js_response["status"], 200) self.assertIn("loadDashboard", js_response["body"]) self.assertIn("renderEvents", js_response["body"]) def test_web_console_exposes_delivery_mode_selector_for_product_use(self): app = create_web_app(load_config_dict({})) index_response = app.handle_request("GET", "/") js_response = app.handle_request("GET", "/app.js") css_response = app.handle_request("GET", "/app.css") self.assertEqual(index_response["status"], 200) self.assertEqual(js_response["status"], 200) self.assertEqual(css_response["status"], 200) self.assertIn("delivery_mode", index_response["body"]) self.assertIn("真实短信", index_response["body"]) self.assertIn("mode-hint", index_response["body"]) self.assertIn("dispatch", js_response["body"]) self.assertIn("real_sms", js_response["body"]) self.assertIn("formData.get('delivery_mode')", js_response["body"]) self.assertIn("需要已配置 ModemManager 发送能力", js_response["body"]) self.assertIn("renderResult", js_response["body"]) self.assertIn("error_category", js_response["body"]) self.assertIn("mode-hint-warning", css_response["body"]) def test_recent_events_support_operator_filtering_and_failure_visuals(self): app = create_web_app(load_config_dict({})) index_response = app.handle_request("GET", "/") js_response = app.handle_request("GET", "/app.js") css_response = app.handle_request("GET", "/app.css") self.assertEqual(index_response["status"], 200) self.assertEqual(js_response["status"], 200) self.assertEqual(css_response["status"], 200) self.assertIn("事件搜索", index_response["body"]) self.assertIn("只看失败", index_response["body"]) self.assertIn("event-search", index_response["body"]) self.assertIn("getFilteredEvents", js_response["body"]) self.assertIn("没有匹配当前筛选条件的事件", js_response["body"]) self.assertIn("eventResultFilter", js_response["body"]) self.assertIn("event-mode-badge", css_response["body"]) self.assertIn("event-failure-meta", css_response["body"]) self.assertIn("event-toolbar", css_response["body"]) self.assertIn("escapeHtml", js_response["body"]) self.assertIn("当前事件数", js_response["body"]) self.assertIn("状态、方式", index_response["body"]) def test_settings_update_reflects_in_dashboard(self): with tempfile.TemporaryDirectory() as tmpdir: config_path = Path(tmpdir) / "appsettings.json" app = create_web_app(load_config_dict({"forwarding": {"device_name": "before"}}), config_path=config_path) response = app.handle_request( "POST", "/api/settings", body=json.dumps({"forwarding": {"device_name": "after"}}).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(response["status"], 200) dashboard = json.loads(app.handle_request("GET", "/api/dashboard")["body"]) self.assertEqual(dashboard["runtime"]["forwarding"]["device_name"], "after") persisted = json.loads(config_path.read_text(encoding="utf-8")) self.assertEqual(persisted["forwarding"]["device_name"], "after") if __name__ == "__main__": unittest.main()