宜博 / lemongraph

"""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 base64
import http.client
import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from unittest.mock import patch

from src import backend_server, ws_subscribe


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(db_path=":memory:")
        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(db_path=":memory:")
        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 StaticFileServingTest(unittest.TestCase):
    """U-008: frontend shell served as static files. Actual client-side
    routing behavior was verified in a real browser this session (nav
    highlighting + content swap across all 3 routes, no page reload) --
    that's not something a stdlib-only test can exercise without a JS
    engine, so this only covers what the server is responsible for."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        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 _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = resp.read()
        content_type = resp.getheader("Content-Type")
        conn.close()
        return resp.status, content_type, body

    def test_root_serves_index_html(self):
        status, content_type, body = self._get("/")
        self.assertEqual(status, 200)
        self.assertIn("text/html", content_type)
        self.assertIn(b"<main id=\"app\">", body)
        self.assertIn(b'href="#/tasks"', body)

    def test_app_js_served_with_js_content_type(self):
        status, content_type, body = self._get("/app.js")
        self.assertEqual(status, 200)
        self.assertIn("javascript", content_type)
        self.assertIn(b"hashchange", body)
        self.assertIn(b'"#/tasks"', body)
        self.assertIn(b'"#/settings"', body)

    def test_style_css_served_with_css_content_type(self):
        status, content_type, body = self._get("/style.css")
        self.assertEqual(status, 200)
        self.assertIn("text/css", content_type)

    def test_unknown_get_path_returns_404(self):
        status, _, body = self._get("/nope")
        self.assertEqual(status, 404)
        self.assertEqual(json.loads(body)["error"], "NOT_FOUND")


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_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()

    def test_windows_success_returns_selected_path(self):
        with patch("src.backend_server.platform.system", return_value="Windows"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["powershell"], returncode=0, stdout="C:\\Users\\x\\some-project\r\n", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": "C:\\Users\\x\\some-project"})

    def test_windows_user_cancel_returns_null_path(self):
        with patch("src.backend_server.platform.system", return_value="Windows"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["powershell"], returncode=0, stdout="", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": None})

    def test_windows_missing_powershell_raises_dialog_unavailable(self):
        with patch("src.backend_server.platform.system", return_value="Windows"), \
             patch("src.backend_server.subprocess.run", side_effect=FileNotFoundError("no powershell")):
            with self.assertRaises(backend_server.DialogUnavailableError):
                backend_server.pick_folder()

    def test_linux_zenity_success_returns_selected_path(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["zenity"], returncode=0, stdout="/home/x/some-project\n", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": "/home/x/some-project"})
        run.assert_called_once()
        self.assertEqual(run.call_args[0][0][0], "zenity")

    def test_linux_zenity_cancel_returns_null_path(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["zenity"], returncode=1, stdout="", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": None})

    def test_linux_falls_back_to_kdialog_when_zenity_missing(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run") as run:
            run.side_effect = [
                FileNotFoundError("no zenity"),
                subprocess.CompletedProcess(args=["kdialog"], returncode=0, stdout="/home/x/proj\n", stderr=""),
            ]
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": "/home/x/proj"})
        self.assertEqual(run.call_count, 2)

    def test_linux_neither_zenity_nor_kdialog_raises_dialog_unavailable(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run", side_effect=FileNotFoundError("not found")):
            with self.assertRaises(backend_server.DialogUnavailableError):
                backend_server.pick_folder()


class PickFolderRouteTest(unittest.TestCase):
    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        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_when_no_dialog_tool_present(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run", side_effect=FileNotFoundError("not found")):
            status, body = self._post("/api/system/pick-folder")
        self.assertEqual(status, 422)
        self.assertEqual(body["error"], "DIALOG_UNAVAILABLE")


class ProjectCreateRouteTest(unittest.TestCase):
    """c_project_create (03-C §3.0) wired to a real HTTP route for the first
    time in U-022 -- project_scaffold.create_project() itself was already
    fully implemented and tested in U-002, so this only exercises the
    request/response/error-code plumbing, not the scaffolding logic again."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
        self.server = backend_server.create_server(db_path=":memory:")
        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, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_creates_project_for_existing_git_repo(self):
        status, body = self._post("/api/projects", {"target_folder_path": self.tmp.name})
        self.assertEqual(status, 200)
        self.assertTrue(body["has_git_repo"])
        self.assertTrue(body["graphlemon_subdir_initialized"])
        self.assertTrue(os.path.isdir(os.path.join(self.tmp.name, ".graphlemon")))

    def test_repeat_call_is_idempotent(self):
        status1, body1 = self._post("/api/projects", {"target_folder_path": self.tmp.name})
        status2, body2 = self._post("/api/projects", {"target_folder_path": self.tmp.name})
        self.assertEqual(status1, 200)
        self.assertEqual(status2, 200)
        self.assertEqual(body1["project_id"], body2["project_id"])

    def test_missing_target_folder_path_returns_400(self):
        status, body = self._post("/api/projects", {})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_INPUT")

    def test_nonexistent_path_returns_404(self):
        status, body = self._post("/api/projects", {"target_folder_path": "/no/such/path/at/all"})
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "PATH_NOT_FOUND")

    def test_non_git_folder_without_confirmation_returns_409(self):
        non_git = tempfile.mkdtemp()
        self.addCleanup(lambda: shutil.rmtree(non_git, ignore_errors=True))
        status, body = self._post("/api/projects", {"target_folder_path": non_git})
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "GIT_INIT_DECLINED")

    def test_non_git_folder_with_confirmation_initializes_git(self):
        non_git = tempfile.mkdtemp()
        self.addCleanup(lambda: shutil.rmtree(non_git, ignore_errors=True))
        status, body = self._post(
            "/api/projects", {"target_folder_path": non_git, "init_git_if_missing": True}
        )
        self.assertEqual(status, 200)
        self.assertTrue(body["has_git_repo"])
        self.assertTrue(os.path.isdir(os.path.join(non_git, ".git")))


class TaskAndBugReportRouteTest(unittest.TestCase):
    """U-023: c_task_create (03-C §3.1) + c_bug_report_create (03-C §3.1.1)
    wired to real HTTP routes for v_goal_composer/v_bug_feedback."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
        )
        self.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _post(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_creates_task(self):
        status, body = self._post(
            "/api/tasks", {"project_id": "proj-1", "goal_text": "make a snake game", "created_via": "human_console"}
        )
        self.assertEqual(status, 200)
        self.assertEqual(body["status"], "queued")
        self.assertIn("task_id", body)

    def test_empty_goal_text_returns_400(self):
        status, body = self._post("/api/tasks", {"project_id": "proj-1", "goal_text": ""})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_nonexistent_project_id_returns_400(self):
        status, body = self._post("/api/tasks", {"project_id": "no-such-project", "goal_text": "x"})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_creates_bug_report(self):
        status, body = self._post(
            "/api/bug-reports", {"project_id": "proj-1", "description": "snake clips through walls"}
        )
        self.assertEqual(status, 200)
        self.assertEqual(body["status"], "open")
        self.assertIn("bug_id", body)

    def test_bug_report_with_related_task_id(self):
        _, task_body = self._post("/api/tasks", {"project_id": "proj-1", "goal_text": "x"})
        status, body = self._post(
            "/api/bug-reports",
            {"project_id": "proj-1", "description": "x", "related_task_id": task_body["task_id"]},
        )
        self.assertEqual(status, 200)

    def test_empty_description_returns_400(self):
        status, body = self._post("/api/bug-reports", {"project_id": "proj-1", "description": ""})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_bug_report_with_bad_project_id_returns_400_not_500(self):
        # a raw sqlite3.IntegrityError propagating uncaught here was a real
        # bug this smoke test caught (stale/bad project_id crashed the
        # request instead of a clean 400)
        status, body = self._post("/api/bug-reports", {"project_id": "no-such-project", "description": "x"})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")


def _ws_handshake_request(path, key):
    return (
        "GET {} HTTP/1.1\r\n"
        "Host: 127.0.0.1\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        "Sec-WebSocket-Key: {}\r\n"
        "Sec-WebSocket-Version: 13\r\n"
        "\r\n"
    ).format(path, key).encode("ascii")


def _read_http_response_head(sock):
    """Reads until the blank line terminating an HTTP response's headers --
    good enough for both a 101 upgrade and a plain error response, since
    neither carries a body in this handler."""
    buf = b""
    while b"\r\n\r\n" not in buf:
        chunk = sock.recv(4096)
        if not chunk:
            break
        buf += chunk
    return buf.decode("ascii", errors="replace")


def _read_server_frame(sock):
    """Server->client frames are never masked (RFC 6455 §5.1) -- simpler
    decode than ws_subscribe.decode_client_frame, which expects masking."""
    first_two = _recv_exact(sock, 2)
    length = first_two[1] & 0x7F
    if length == 126:
        length = int.from_bytes(_recv_exact(sock, 2), "big")
    elif length == 127:
        length = int.from_bytes(_recv_exact(sock, 8), "big")
    payload = _recv_exact(sock, length)
    return json.loads(payload.decode("utf-8"))


def _recv_exact(sock, n):
    chunks = []
    remaining = n
    while remaining > 0:
        chunk = sock.recv(remaining)
        if not chunk:
            raise ConnectionError("socket closed")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


class WsSubscribeRouteTest(unittest.TestCase):
    """U-024: c_ws_subscribe end-to-end over a real raw socket -- the actual
    RFC 6455 handshake + framing, not mocked, since that's exactly the
    hand-rolled logic (no `websockets` dependency, 05 §5.7) most likely to
    have a real bug."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
        )
        self.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _connect(self):
        host, port = self.server.server_address
        return socket.create_connection((host, port), timeout=5)

    def test_handshake_succeeds_with_correct_accept_key(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1", key))
        head = _read_http_response_head(sock)
        self.assertIn("101 Switching Protocols", head.splitlines()[0])
        expected_accept = ws_subscribe.compute_accept_key(key)
        self.assertIn("Sec-WebSocket-Accept: " + expected_accept, head)

    def test_nonexistent_project_rejected_before_upgrade(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/no-such-project", key))
        head = _read_http_response_head(sock)
        self.assertIn("404", head.splitlines()[0])
        self.assertNotIn("101", head.splitlines()[0])

    def test_non_upgrade_request_rejected(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        sock.sendall(b"GET /api/ws/projects/proj-1 HTTP/1.1\r\nHost: x\r\n\r\n")
        head = _read_http_response_head(sock)
        self.assertIn("400", head.splitlines()[0])

    def test_new_execution_event_pushed_within_two_seconds(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1", key))
        _read_http_response_head(sock)

        start = time.monotonic()
        with self.server.conn_lock:
            self.server.conn.execute(
                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
                "VALUES ('evt-1', 'proj-1', NULL, 'budget_updated', '{\"x\": 1}')"
            )
            self.server.conn.commit()

        sock.settimeout(5)
        event = _read_server_frame(sock)
        elapsed = time.monotonic() - start
        self.assertLess(elapsed, 2.0)
        self.assertEqual(event["event_id"], "evt-1")
        self.assertEqual(event["event_type"], "budget_updated")
        self.assertEqual(event["payload"], {"x": 1})

    def test_task_id_filter_excludes_unrelated_events(self):
        with self.server.conn_lock:
            self.server.conn.execute(
                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
                "VALUES ('task-1', 'proj-1', 'g', 'goal', 'queued', 'planner')"
            )
            self.server.conn.commit()

        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1?task_id=task-1", key))
        _read_http_response_head(sock)

        with self.server.conn_lock:
            self.server.conn.execute(
                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
                "VALUES ('evt-unrelated', 'proj-1', NULL, 'budget_updated', '{}')"
            )
            self.server.conn.execute(
                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
                "VALUES ('evt-related', 'proj-1', 'task-1', 'task_status_changed', '{}')"
            )
            self.server.conn.commit()

        sock.settimeout(5)
        event = _read_server_frame(sock)
        self.assertEqual(event["event_id"], "evt-related")


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")
        # "src" is graphlemon's own package (e.g. project_scaffold/schema/paths,
        # wired in U-022) -- not a third-party dependency, so it's allowed;
        # everything else must be in the standard library.
        non_stdlib = {name for name in top_level_names if name not in sys.stdlib_module_names and name != "src"}
        self.assertEqual(non_stdlib, set(), "non-stdlib imports found: {}".format(non_stdlib))


if __name__ == "__main__":
    unittest.main()