import json import tempfile import unittest from pathlib import Path from dbussmsforward.config import load_config_dict from dbussmsforward.sms import SmsMessage, SmsSendResult from dbussmsforward.web import build_handler_class, create_web_app class RecordingDispatcher: def __init__(self): self.messages = [] def dispatch(self, sms: SmsMessage): self.messages.append(sms) return [] class RecordingSmsSender: def __init__(self, *, result: SmsSendResult | None = None): self.calls = [] self._result = result def send(self, phone_number: str, content: str): self.calls.append({"phone_number": phone_number, "content": content}) if self._result is not None: return SmsSendResult( success=self._result.success, phone_number=phone_number, content=content, modem_path=self._result.modem_path, message_path=self._result.message_path, error=self._result.error, error_category=self._result.error_category, ) return SmsSendResult( success=True, phone_number=phone_number, content=content, modem_path="/org/freedesktop/ModemManager1/Modem/0", message_path="/org/freedesktop/ModemManager1/SMS/100", ) class WebAppTests(unittest.TestCase): def test_health_status_and_settings_round_trip(self): with tempfile.TemporaryDirectory() as tmpdir: config_path = Path(tmpdir) / "appsettings.json" app = create_web_app( load_config_dict( { "daemon": {"host": "127.0.0.1", "port": 9090}, "forwarding": {"device_name": "router-web"}, "notifications": {"console": {"enabled": True}}, } ), config_path=config_path, ) health = app.handle_request("GET", "/api/health") self.assertEqual(health["status"], 200) self.assertEqual(json.loads(health["body"])["status"], "ok") system_status = app.handle_request("GET", "/api/system/status") status_payload = json.loads(system_status["body"]) self.assertEqual(system_status["status"], 200) self.assertEqual(status_payload["daemon"]["host"], "127.0.0.1") self.assertEqual(status_payload["forwarding"]["device_name"], "router-web") self.assertEqual(status_payload["notifications"]["enabled_channels"], ["console"]) settings = app.handle_request("GET", "/api/settings") self.assertEqual(settings["status"], 200) settings_payload = json.loads(settings["body"]) self.assertEqual(settings_payload["forwarding"]["device_name"], "router-web") update = app.handle_request( "POST", "/api/settings", body=json.dumps( { "forwarding": {"device_name": "router-updated"}, "notifications": {"console": {"enabled": False}}, } ).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(update["status"], 200) updated_payload = json.loads(update["body"]) self.assertEqual(updated_payload["forwarding"]["device_name"], "router-updated") self.assertFalse(updated_payload["notifications"]["console"]["enabled"]) persisted = json.loads(config_path.read_text(encoding="utf-8")) self.assertEqual(persisted["forwarding"]["device_name"], "router-updated") def test_send_api_dispatches_message(self): app = create_web_app(load_config_dict({"forwarding": {"device_name": "router-web"}})) dispatcher = RecordingDispatcher() app.runtime.dispatcher = dispatcher response = app.handle_request( "POST", "/api/sms/send", body=json.dumps({"phone_number": "10086", "content": "hello from web"}).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(response["status"], 200) payload = json.loads(response["body"]) self.assertEqual(payload["status"], "queued") self.assertEqual(len(dispatcher.messages), 1) self.assertEqual(dispatcher.messages[0].phone_number, "10086") self.assertEqual(dispatcher.messages[0].content, "hello from web") self.assertEqual(dispatcher.messages[0].received_at, "manual") def test_send_api_supports_real_sms_send_mode(self): app = create_web_app(load_config_dict({})) sender = RecordingSmsSender() app.runtime.sms_sender = sender response = app.handle_request( "POST", "/api/sms/send", body=json.dumps( {"phone_number": "10086", "content": "web real send", "delivery_mode": "real_sms"} ).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(response["status"], 200) payload = json.loads(response["body"]) self.assertEqual(payload["status"], "sent") self.assertEqual(payload["delivery_mode"], "real_sms") self.assertEqual(payload["modem_path"], "/org/freedesktop/ModemManager1/Modem/0") self.assertEqual(payload["message_path"], "/org/freedesktop/ModemManager1/SMS/100") self.assertEqual(sender.calls, [{"phone_number": "10086", "content": "web real send"}]) def test_send_api_reports_real_sms_send_failures(self): app = create_web_app(load_config_dict({})) sender = RecordingSmsSender( result=SmsSendResult( success=False, phone_number="", content="", error="sms send is not configured", error_category="disabled", ) ) app.runtime.sms_sender = sender response = app.handle_request( "POST", "/api/sms/send", body=json.dumps( {"phone_number": "10010", "content": "will fail", "delivery_mode": "real_sms"} ).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(response["status"], 503) payload = json.loads(response["body"]) self.assertEqual(payload["error"], "sms_send_failed") self.assertEqual(payload["error_category"], "disabled") self.assertIn("sms send is not configured", payload["detail"]) def test_send_api_rejects_missing_fields(self): app = create_web_app(load_config_dict({})) response = app.handle_request( "POST", "/api/sms/send", body=json.dumps({"content": "missing phone"}).encode("utf-8"), headers={"Content-Type": "application/json"}, ) self.assertEqual(response["status"], 400) payload = json.loads(response["body"]) self.assertEqual(payload["error"], "validation_error") self.assertIn("phone_number", payload["detail"]) def test_static_assets_are_served(self): app = create_web_app(load_config_dict({})) index_response = app.handle_request("GET", "/") self.assertEqual(index_response["status"], 200) self.assertIn("DbusSmsForward", index_response["body"]) self.assertEqual(index_response["content_type"], "text/html; charset=utf-8") js_response = app.handle_request("GET", "/app.js") self.assertEqual(js_response["status"], 200) self.assertIn("fetch", js_response["body"]) self.assertEqual(js_response["content_type"], "application/javascript; charset=utf-8") def test_handler_class_serves_http_requests(self): app = create_web_app(load_config_dict({"forwarding": {"device_name": "router-http"}})) handler_class = build_handler_class(app) with tempfile.TemporaryDirectory() as tmpdir: request_path = Path(tmpdir) / "request.txt" request_path.write_text( "GET /api/health HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", encoding="utf-8", ) import socketserver import threading from http.server import HTTPServer class TestServer(socketserver.TCPServer): allow_reuse_address = True server = TestServer(("127.0.0.1", 0), handler_class) thread = threading.Thread(target=server.handle_request, daemon=True) thread.start() import urllib.request try: with urllib.request.urlopen(f"http://127.0.0.1:{server.server_address[1]}/api/health", timeout=5) as response: body = response.read().decode("utf-8") self.assertEqual(response.status, 200) self.assertEqual(json.loads(body)["status"], "ok") finally: server.server_close() thread.join(timeout=5) if __name__ == "__main__": unittest.main()