import contextlib import importlib.util import io import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import threading import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from unittest.mock import patch SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "guiji_app.py" spec = importlib.util.spec_from_file_location("guiji_app", SCRIPT) api = importlib.util.module_from_spec(spec) spec.loader.exec_module(api) class ClientTest(unittest.TestCase): def setUp(self): self.calls = [] self.status = 200 self.raw = '{"camera":null,"filters":{}}' self.redirect = None owner = self class Handler(BaseHTTPRequestHandler): def do_GET(self): self.respond() def do_POST(self): self.respond() def respond(self): data = self.rfile.read(int(self.headers.get("Content-Length", 0))) owner.calls.append((self.command, self.path, dict(self.headers), data)) self.send_response(owner.status) self.send_header("Content-Type", "application/json") if owner.redirect: self.send_header("Location", owner.redirect) self.end_headers() self.wfile.write(owner.raw.encode("utf-8")) def log_message(self, *args): pass self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.url = "http://127.0.0.1:%s" % self.server.server_port self.env = {"GUIJI_APP_URL": self.url, "GUIJI_APP_TOKEN": "mock-token"} self.addCleanup(self.close_server) def close_server(self): self.server.shutdown() self.server.server_close() self.thread.join() def run_cli(self, args, env=None): stdout, stderr = io.StringIO(), io.StringIO() with patch.dict(os.environ, self.env if env is None else env, clear=True): with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): code = api.main(args) return code, stdout.getvalue(), stderr.getvalue() def test_state_and_authorization_header(self): code, stdout, stderr = self.run_cli(["state"]) self.assertEqual(code, 0) self.assertEqual(json.loads(stdout), json.loads(self.raw)) self.assertEqual(stderr, "") method, path, headers, body = self.calls[0] self.assertEqual((method, path, body), ("GET", "/api/v1/state", b"")) self.assertEqual(headers["Authorization"], "Bearer mock-token") self.assertNotIn("mock-token", stdout + stderr + path) def test_arbitrary_command_json_and_headers(self): self.raw = '{"ok":true,"result":{"未知字段":1},"settled":false,"state":{}}' params = {"patchIds": ["林块"], "lng": 117.2, "lat": 40.2} code, stdout, _ = self.run_cli(["command", "intel.patches", "--params", json.dumps(params)]) self.assertEqual(code, 0) self.assertFalse(json.loads(stdout)["settled"]) self.assertEqual(len(self.calls), 1) # No automatic state read or retry. method, path, headers, body = self.calls[0] self.assertEqual((method, path), ("POST", "/api/v1/command")) self.assertEqual(headers["Authorization"], "Bearer mock-token") self.assertEqual(headers["Content-Type"], "application/json") self.assertEqual(json.loads(body), {"action": "intel.patches", "params": params}) self.assertNotIn(b"mock-token", body) def test_default_empty_params(self): code, _, _ = self.run_cli(["command", "account.status"]) self.assertEqual(code, 0) self.assertEqual(json.loads(self.calls[0][3]), {"action": "account.status", "params": {}}) def test_polish_actions_and_optional_favorites_source_are_preserved(self): responses = [ ("map.closeDetail", {"closed": True}, {}), ("species.list", [{"id": "boletus_edulis", "name": "Porcini", "aliases": ["牛肝菌"]}], {}), ("forestTypes.list", [{"code": 70, "name": "Evergreen needleleaf"}], {}), *[("favorites.list", [], {"offline": source}) for source in (True, False, None)], ] for action, result, extra in responses: with self.subTest(action=action, extra=extra): body = {"ok": True, "action": action, "result": result, "settled": True, "state": {}, **extra} self.raw = json.dumps(body, ensure_ascii=False) code, stdout, stderr = self.run_cli(["command", action]) self.assertEqual((code, stderr), (0, "")) self.assertEqual(json.loads(stdout), body) self.assertEqual(json.loads(self.calls[-1][3]), {"action": action, "params": {}}) self.assertEqual(len(self.calls), len(responses)) def test_error_passthrough_with_code_reason_and_extra_fields(self): self.status = 403 self.raw = '{"ok":false, "error":"forbidden", "message":"会员不足", "code":"plan_required", "reason":"limit", "requiredPlan":"pro", "currentLimit":3, "requested":5}' code, stdout, stderr = self.run_cli(["command", "intel.topPatches"]) self.assertEqual((code, stdout), (1, "")) self.assertEqual(stderr, "HTTP 403\n" + self.raw + "\n") self.assertEqual(len(self.calls), 1) def test_not_ready_and_business_error_at_http_200(self): for status in (409, 200): with self.subTest(status=status): self.status = status self.raw = '{"ok":false,"error":"not_ready","message":"Open map"}' code, stdout, stderr = self.run_cli(["state"]) self.assertEqual((code, stdout), (1, "")) self.assertEqual(stderr, f"HTTP {status}\n{self.raw}\n") self.assertEqual(len(self.calls), 2) def test_non_json_http_error_preserved(self): self.status, self.raw = 502, "original upstream error" code, _, stderr = self.run_cli(["state"]) self.assertEqual(code, 1) self.assertEqual(stderr, "HTTP 502\noriginal upstream error\n") def test_redirects_refused_for_state_and_command(self): # A second listener would record any credential-bearing redirected request. sink = ThreadingHTTPServer(("127.0.0.1", 0), self.server.RequestHandlerClass) thread = threading.Thread(target=sink.serve_forever, daemon=True) thread.start() try: for status in (301, 302, 303, 307, 308): for operation in (["state"], ["command", "account.status"]): with self.subTest(status=status, operation=operation): self.status = status self.redirect = f"http://127.0.0.1:{sink.server_port}/stolen" self.raw = '{"message":"redirect denied"}' before = len(self.calls) code, stdout, stderr = self.run_cli(operation) self.assertEqual((code, stdout), (1, "")) self.assertEqual(stderr, f"HTTP {status}\n{self.raw}\n") self.assertEqual(len(self.calls), before + 1) self.assertNotEqual(self.calls[-1][1], "/stolen") finally: sink.shutdown() sink.server_close() thread.join() def test_environment_proxies_disabled(self): env = dict(self.env, http_proxy="http://127.0.0.1:1", HTTP_PROXY="http://127.0.0.1:1", all_proxy="http://127.0.0.1:1", no_proxy="", NO_PROXY="") code, _, stderr = self.run_cli(["state"], env) self.assertEqual((code, stderr), (0, "")) self.assertEqual(len(self.calls), 1) def test_invalid_params_do_not_send(self): for params in ("[]", "null", "not-json", '{"value":NaN}'): with self.subTest(params=params): code, _, _ = self.run_cli(["command", "example.action", "--params", params]) self.assertEqual(code, 1) self.assertEqual(self.calls, []) def test_invalid_url_or_missing_token_do_not_send(self): urls = (self.url + "/other", self.url + "?token=secret", self.url + "#secret", "http://user:secret@localhost", "https://localhost", "http://localhost:99999") for url in urls: code, _, stderr = self.run_cli(["state"], dict(self.env, GUIJI_APP_URL=url)) self.assertEqual(code, 1) self.assertNotIn("secret", stderr) code, _, stderr = self.run_cli(["state"], {"GUIJI_APP_URL": self.url}) self.assertEqual(code, 1) self.assertIn("GUIJI_APP_TOKEN", stderr) self.assertEqual(self.calls, []) def test_copied_script_works_without_repository(self): with tempfile.TemporaryDirectory() as directory: copied = shutil.copy(SCRIPT, directory) env = dict(os.environ, **self.env) result = subprocess.run([sys.executable, copied, "state"], cwd=directory, env=env, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout), json.loads(self.raw)) class DiscoveryTest(unittest.TestCase): def test_missing_native_tools_asks_for_manual_address(self): with patch.object(api.shutil, "which", return_value=None): with self.assertRaisesRegex(RuntimeError, "address in APP settings"): api.discover() def test_dns_sd_is_bounded_and_resolves_exact_name(self): with patch.object(api.shutil, "which", return_value="/usr/bin/dns-sd"): with patch.object(api.subprocess, "run", side_effect=subprocess.TimeoutExpired("dns-sd", 5, output=b"phone.local.:8765\n")) as run: with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): api.discover("Guiji Agent (2)") run.assert_called_once_with(["dns-sd", "-L", "Guiji Agent (2)", "_guiji-agent._tcp"], capture_output=True, timeout=5) def test_avahi_fallback(self): with patch.object(api.shutil, "which", side_effect=lambda name: "/usr/bin/avahi-browse" if name == "avahi-browse" else None): with patch.object(api.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, b"phone address\n", b"")) as run: with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): api.discover() run.assert_called_once_with(["avahi-browse", "-rt", "_guiji-agent._tcp"], capture_output=True, timeout=5) if __name__ == "__main__": unittest.main()