import json import subprocess import sys import tempfile import textwrap import unittest from pathlib import Path PROJECT_ROOT = Path("/Users/chick/.Hermes/workspace/DbusSmsForward") CLI_ENV = {**(__import__("os").environ), "PYTHONPATH": str(PROJECT_ROOT / "src")} class CliTests(unittest.TestCase): def test_validate_config_command_rejects_enabled_autman_push_without_required_fields(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" config_path.write_text( json.dumps( { "notifications": { "autman_push": { "enabled": True, "api_url": "", "access_token": "", } } }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "validate-config", "--config", str(config_path), ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("notifications.autman_push.api_url is required", result.stderr) def test_validate_config_command_rejects_invalid_autman_push_method(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" config_path.write_text( json.dumps( { "notifications": { "autman_push": { "enabled": True, "api_url": "https://autman.example/push", "access_token": "***", "method": "PUT", } } }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "validate-config", "--config", str(config_path), ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("notifications.autman_push.method must be GET or POST", result.stderr) def test_validate_config_command_rejects_invalid_modemmanager_ready_values(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" config_path.write_text( json.dumps( { "sms_source": { "kind": "modemmanager", "modem_object_path": "", "state_ready_values": [], } }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "validate-config", "--config", str(config_path), ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("sms_source.state_ready_values must not be empty", result.stderr) self.assertIn("sms_source.modem_object_path is required", result.stderr) def test_validate_config_command_rejects_negative_poll_interval(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" config_path.write_text( json.dumps( { "sms_source": { "kind": "file", "path": "/tmp/messages.json", "poll_interval_seconds": -1, } }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "validate-config", "--config", str(config_path), ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("sms_source.poll_interval_seconds must be >= 0", result.stderr) def test_run_command_processes_file_source_messages(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" messages_path = root / "messages.json" messages_path.write_text( json.dumps( [ { "phone_number": "1068", "content": "您的验证码是 654321", "received_at": "2026-04-24 12:00:00", } ], ensure_ascii=False, ), encoding="utf-8", ) config_path.write_text( json.dumps( { "forwarding": {"device_name": "router-b"}, "notifications": {"console": {"enabled": True}}, "sms_source": {"kind": "file", "path": str(messages_path)}, }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "run", "--config", str(config_path), ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, timeout=5, ) self.assertEqual(result.returncode, 0, msg=result.stderr) self.assertIn("短信转发 1068", result.stdout) self.assertIn("router-b", result.stdout) self.assertIn("654321", result.stdout) def test_send_command_outputs_manual_message(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" config_path.write_text( json.dumps( { "forwarding": {"device_name": "router-c"}, "notifications": {"console": {"enabled": True}}, }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "send", "--config", str(config_path), "--to", "10086", "--text", "manual body", ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertEqual(result.returncode, 0, msg=result.stderr) self.assertIn("manual body", result.stdout) self.assertIn("10086", result.stdout) def test_send_command_real_sms_requires_modemmanager_source(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" config_path.write_text( json.dumps( { "notifications": {"console": {"enabled": True}}, }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "send", "--config", str(config_path), "--to", "10086", "--text", "manual body", "--real-sms", ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("sms send is not configured", result.stderr) def test_run_command_supports_multi_poll_deduplication(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" messages_path = root / "messages.json" messages_path.write_text( json.dumps( [ { "phone_number": "1068", "content": "首条短信", "received_at": "2026-04-24 12:00:00", } ], ensure_ascii=False, ), encoding="utf-8", ) config_path.write_text( json.dumps( { "forwarding": {"device_name": "router-loop"}, "notifications": {"console": {"enabled": True}}, "sms_source": { "kind": "file", "path": str(messages_path), "poll_interval_seconds": 0, }, }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "run", "--config", str(config_path), "--iterations", "2", ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertEqual(result.returncode, 0, msg=result.stderr) self.assertEqual(result.stdout.count("短信转发 1068"), 1) self.assertIn("polls=2", result.stderr) self.assertIn("duplicates_skipped=1", result.stderr) def test_send_command_dispatches_autman_push_with_mock_server(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" capture_path = root / "capture.json" mock_server = root / "mock_autman_server.py" mock_server.write_text( """ import json import sys from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path capture_path = Path(sys.argv[1]) class Handler(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get('Content-Length', '0')) payload = self.rfile.read(length).decode('utf-8') capture_path.write_text(json.dumps({ 'path': self.path, 'headers': dict(self.headers), 'body': json.loads(payload), }, ensure_ascii=False), encoding='utf-8') self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"ok": true}') def log_message(self, format, *args): return server = HTTPServer(('127.0.0.1', 0), Handler) print(server.server_address[1], flush=True) server.handle_request() """.strip(), encoding="utf-8", ) with subprocess.Popen( [sys.executable, str(mock_server), str(capture_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(root), ) as server: result = None port_line = server.stdout.readline().strip() if not port_line: _, stderr_text = server.communicate(timeout=5) self.fail(stderr_text) config_path.write_text( json.dumps( { "forwarding": {"device_name": "router-d"}, "notifications": { "autman_push": { "enabled": True, "api_url": f"http://127.0.0.1:{port_line}/api/push", "access_token": "cli-token", "method": "POST", } }, }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "send", "--config", str(config_path), "--to", "10010", "--text", "autman cli body", ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, timeout=15, ) if server.poll() is None: server.communicate(timeout=5) self.assertEqual(result.returncode, 0, msg=result.stderr) captured = json.loads(capture_path.read_text(encoding="utf-8")) self.assertEqual(captured["path"], "/api/push") self.assertEqual(captured["body"]["title"], "短信转发 10010") self.assertEqual(captured["body"]["from"], "router-d") self.assertIn("autman cli body", captured["body"]["content"]) self.assertEqual(captured["headers"]["Authorization"], "Bearer cli-token") def test_run_command_reports_sender_level_failures_and_keeps_partial_success_output(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" messages_path = root / "messages.json" messages_path.write_text( json.dumps( [ { "phone_number": "1068", "content": "sender partial failure", "received_at": "2026-04-24 12:00:00", } ], ensure_ascii=False, ), encoding="utf-8", ) server_script = root / "failing_server.py" server_script.write_text( textwrap.dedent( """ import sys from http.server import BaseHTTPRequestHandler, HTTPServer class Handler(BaseHTTPRequestHandler): def do_POST(self): self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error": "boom"}') def log_message(self, format, *args): return server = HTTPServer(('127.0.0.1', 0), Handler) print(server.server_address[1], flush=True) server.handle_request() """ ).strip(), encoding="utf-8", ) with subprocess.Popen( [sys.executable, str(server_script)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(root), ) as server: port_line = server.stdout.readline().strip() if not port_line: _, stderr_text = server.communicate(timeout=5) self.fail(stderr_text) config_path.write_text( json.dumps( { "forwarding": {"device_name": "router-observe"}, "notifications": { "console": {"enabled": True}, "autman_push": { "enabled": True, "api_url": f"http://127.0.0.1:{port_line}/api/push", "access_token": "***", "method": "POST", }, }, "sms_source": {"kind": "file", "path": str(messages_path)}, }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "run", "--config", str(config_path), ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, timeout=15, ) if server.poll() is None: server.communicate(timeout=5) self.assertEqual(result.returncode, 0, msg=result.stderr) self.assertIn("[console] 短信转发 1068", result.stdout) self.assertIn("sender partial failure", result.stdout) self.assertIn("dispatched=1", result.stderr) self.assertIn("dispatch_failures=1", result.stderr) self.assertIn("dispatch_failure sender=autman_push", result.stderr) self.assertIn("error_category=http", result.stderr) self.assertIn("status=500", result.stderr) self.assertIn("response_excerpt={\"error\": \"boom\"}", result.stderr) def test_run_command_reports_file_source_fetch_issues(self): with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) config_path = root / "config.json" messages_path = root / "messages.json" messages_path.write_text('{"broken": ', encoding="utf-8") config_path.write_text( json.dumps( { "forwarding": {"device_name": "router-fetch"}, "notifications": {"console": {"enabled": True}}, "sms_source": {"kind": "file", "path": str(messages_path)}, }, ensure_ascii=False, ), encoding="utf-8", ) result = subprocess.run( [ sys.executable, "-m", "dbussmsforward.cli", "run", "--config", str(config_path), ], cwd=str(PROJECT_ROOT), env=CLI_ENV, capture_output=True, text=True, check=False, ) self.assertEqual(result.returncode, 0, msg=result.stderr) self.assertIn("fetch_issues=1", result.stderr) self.assertIn("fetch_issue kind=json_decode_error", result.stderr) self.assertEqual(result.stdout.strip(), "") if __name__ == "__main__": unittest.main()