224 lines
10 KiB
Python
224 lines
10 KiB
Python
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()
|