From 09e8bbbbe875ba5178c67c0bedf9c8d69ab79336 Mon Sep 17 00:00:00 2001 From: bboonnee Date: Fri, 31 Jul 2026 11:26:37 +0800 Subject: [PATCH] U-007: local backend skeleton + c_pick_folder macOS implementation --- src/backend_server.py | 65 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/test_backend_server.py | 157 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 222 insertions(+), 0 deletions(-) create mode 100644 src/backend_server.py create mode 100644 src/test_backend_server.py diff --git a/src/backend_server.py b/src/backend_server.py new file mode 100644 index 0000000..fc2cd62 --- /dev/null +++ b/src/backend_server.py @@ -0,0 +1,65 @@ +"""Local backend process skeleton (U-007): pure stdlib `http.server`, no +third-party deps (05 §5.7) -- this process is also where the Graph Engine +scheduler will live later (05 §5.4.1), but that's a different Unit. + +MVP scope here: process start/stop, and `c_pick_folder` (03-C §3.-1) as the +one real route -- macOS only for now (`osascript`); Windows/Linux native +dialogs are PH3 (U-022). +""" + +import json +import platform +import subprocess +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + +class DialogUnavailableError(Exception): + """DIALOG_UNAVAILABLE -- 03-C §3.-1: no native-dialog capability here.""" + + +def pick_folder(): + """macOS implementation of c_pick_folder. Returns {"selected_path": str} + on success, {"selected_path": None} on user cancel (03-C: not an error). + Raises DialogUnavailableError off macOS or if osascript itself fails for + a reason other than the user cancelling.""" + if platform.system() != "Darwin": + raise DialogUnavailableError("native folder picker only implemented for macOS so far (PH3 adds Windows/Linux)") + + script = 'POSIX path of (choose folder with prompt "Select a graphlemon project folder")' + result = subprocess.run(["osascript", "-e", script], capture_output=True, text=True) + if result.returncode != 0: + if "User canceled" in result.stderr or "(-128)" in result.stderr: + return {"selected_path": None} + raise DialogUnavailableError(result.stderr.strip()) + return {"selected_path": result.stdout.strip()} + + +class _Handler(BaseHTTPRequestHandler): + def do_POST(self): + if self.path == "/api/system/pick-folder": + self._handle_pick_folder() + else: + self._send_json(404, {"error": "NOT_FOUND"}) + + def _handle_pick_folder(self): + try: + self._send_json(200, pick_folder()) + except DialogUnavailableError as e: + self._send_json(422, {"error": "DIALOG_UNAVAILABLE", "detail": str(e)}) + + def _send_json(self, status, obj): + body = json.dumps(obj).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format, *args): + pass # quiets test output; real request logging is a later Unit's concern + + +def create_server(host="127.0.0.1", port=0): + """port=0 lets the OS assign a free port -- avoids collisions in tests + and lets multiple graphlemon instances run side by side if ever needed.""" + return ThreadingHTTPServer((host, port), _Handler) diff --git a/src/test_backend_server.py b/src/test_backend_server.py new file mode 100644 index 0000000..a53c63c --- /dev/null +++ b/src/test_backend_server.py @@ -0,0 +1,157 @@ +"""U-007 tests: backend process start/stop, c_pick_folder route (osascript +mocked at the subprocess boundary), and the zero-third-party-dependency +constraint (05 §5.7) checked statically via AST, not just trusted. + +Run: python3 -m unittest src.test_backend_server -v +""" + +import ast +import http.client +import json +import os +import subprocess +import sys +import threading +import unittest +from unittest.mock import patch + +from src import backend_server + + +class ServerStartStopTest(unittest.TestCase): + def test_can_start_serve_and_stop(self): + # deliberately hits an unrelated route, not /api/system/pick-folder -- + # that one would pop a real, unmocked osascript dialog on this machine + # and hang waiting for a human who isn't there. + server = backend_server.create_server() + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + conn = http.client.HTTPConnection(host, port, timeout=5) + conn.request("POST", "/api/does-not-exist", body=b"{}") + resp = conn.getresponse() + resp.read() + conn.close() + self.assertEqual(resp.status, 404) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + self.assertFalse(thread.is_alive()) + + def test_unknown_route_returns_404(self): + server = backend_server.create_server() + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + conn = http.client.HTTPConnection(host, port, timeout=5) + conn.request("POST", "/api/does-not-exist") + resp = conn.getresponse() + body = json.loads(resp.read()) + conn.close() + self.assertEqual(resp.status, 404) + self.assertEqual(body["error"], "NOT_FOUND") + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +class PickFolderTest(unittest.TestCase): + def test_success_returns_selected_path(self): + with patch("src.backend_server.platform.system", return_value="Darwin"), \ + patch("src.backend_server.subprocess.run") as run: + run.return_value = subprocess.CompletedProcess( + args=["osascript"], returncode=0, stdout="/Users/x/some-project\n", stderr="" + ) + result = backend_server.pick_folder() + self.assertEqual(result, {"selected_path": "/Users/x/some-project"}) + + def test_user_cancel_returns_null_path_not_an_error(self): + with patch("src.backend_server.platform.system", return_value="Darwin"), \ + patch("src.backend_server.subprocess.run") as run: + run.return_value = subprocess.CompletedProcess( + args=["osascript"], returncode=1, stdout="", + stderr="execution error: User canceled. (-128)", + ) + result = backend_server.pick_folder() + self.assertEqual(result, {"selected_path": None}) + + def test_non_macos_raises_dialog_unavailable(self): + with patch("src.backend_server.platform.system", return_value="Linux"): + with self.assertRaises(backend_server.DialogUnavailableError): + backend_server.pick_folder() + + def test_osascript_failure_other_than_cancel_raises_dialog_unavailable(self): + with patch("src.backend_server.platform.system", return_value="Darwin"), \ + patch("src.backend_server.subprocess.run") as run: + run.return_value = subprocess.CompletedProcess( + args=["osascript"], returncode=1, stdout="", stderr="some other AppleScript error", + ) + with self.assertRaises(backend_server.DialogUnavailableError): + backend_server.pick_folder() + + +class PickFolderRouteTest(unittest.TestCase): + def setUp(self): + self.server = backend_server.create_server() + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + def _post(self, path): + host, port = self.server.server_address + conn = http.client.HTTPConnection(host, port, timeout=5) + conn.request("POST", path, body=b"{}") + resp = conn.getresponse() + body = json.loads(resp.read()) + conn.close() + return resp.status, body + + def test_route_returns_selected_path_on_success(self): + with patch("src.backend_server.platform.system", return_value="Darwin"), \ + patch("src.backend_server.subprocess.run") as run: + run.return_value = subprocess.CompletedProcess( + args=["osascript"], returncode=0, stdout="/tmp/chosen\n", stderr="" + ) + status, body = self._post("/api/system/pick-folder") + self.assertEqual(status, 200) + self.assertEqual(body, {"selected_path": "/tmp/chosen"}) + + def test_route_returns_422_dialog_unavailable_off_macos(self): + with patch("src.backend_server.platform.system", return_value="Linux"): + status, body = self._post("/api/system/pick-folder") + self.assertEqual(status, 422) + self.assertEqual(body["error"], "DIALOG_UNAVAILABLE") + + +class ZeroThirdPartyDependencyTest(unittest.TestCase): + """05 §5.7: no Flask/FastAPI/Express/Node.js -- checked statically so this + doesn't silently rot if someone adds an import later.""" + + def test_backend_server_only_imports_stdlib(self): + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backend_server.py") + with open(path, encoding="utf-8") as f: + tree = ast.parse(f.read(), filename=path) + + top_level_names = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + top_level_names.add(alias.name.split(".")[0]) + elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0: + top_level_names.add(node.module.split(".")[0]) + + self.assertTrue(top_level_names, "expected at least one import to check") + non_stdlib = {name for name in top_level_names if name not in sys.stdlib_module_names} + self.assertEqual(non_stdlib, set(), "non-stdlib imports found: {}".format(non_stdlib)) + + +if __name__ == "__main__": + unittest.main() -- libgit2 0.21.4