test_server.py

#!/usr/bin/env python3
import json
import os
import subprocess
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from pathlib import Path

from git_repository import diff_line_changes
from server import APIError, CodeBrowser, CodeBrowserHandler, ReusableThreadingHTTPServer


class CodeBrowserTest(unittest.TestCase):
    def setUp(self):
        self.temporary_directory = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary_directory.name) / "repo"
        self.root.mkdir()
        self._git("init")
        self._git("config", "user.name", "Code Browser Test")
        self._git("config", "user.email", "browser@example.test")

        (self.root / ".gitignore").write_text("ignored.log\n", encoding="utf-8")
        (self.root / ".hidden").write_text("visible to explorer\n", encoding="utf-8")
        (self.root / "root.py").write_text("print('initial')\n", encoding="utf-8")
        (self.root / "rename-me.txt").write_text("rename content\n", encoding="utf-8")
        self.long_lines = ["file start sentinel\n", *[f"unchanged line {line}\n" for line in range(2, 21)]]
        (self.root / "long.txt").write_text("".join(self.long_lines), encoding="utf-8")
        (self.root / "adir").mkdir()
        (self.root / "adir" / "inside.c").write_text("int value;\n", encoding="utf-8")
        self._git("add", ".gitignore", "root.py", "rename-me.txt", "long.txt", "adir/inside.c")
        self._git("commit", "-m", "Initial files")
        self.initial_oid = self._git("rev-parse", "HEAD").stdout.strip()

        os.rename(self.root / "rename-me.txt", self.root / "renamed.txt")
        self.long_lines[9] = "committed change\n"
        (self.root / "long.txt").write_text("".join(self.long_lines), encoding="utf-8")
        self._git("add", "-A")
        self._git("commit", "-m", "Rename file")
        self.rename_oid = self._git("rev-parse", "HEAD").stdout.strip()

        (self.root / "root.py").write_text("print('working')\n", encoding="utf-8")
        self.long_lines[14] = "working change\n"
        (self.root / "long.txt").write_text("".join(self.long_lines), encoding="utf-8")
        (self.root / "staged.txt").write_text("staged content\n", encoding="utf-8")
        self._git("add", "staged.txt")
        (self.root / "untracked.txt").write_text("untracked content\n", encoding="utf-8")
        (self.root / "ignored.log").write_text("ignored content\n", encoding="utf-8")
        (self.root / "binary.bin").write_bytes(b"\x00\x01binary\xff")
        self.svg_content = b'<svg xmlns="http://www.w3.org/2000/svg" width="10" height="10"><rect width="10" height="10"/></svg>'
        (self.root / "diagram.svg").write_bytes(self.svg_content)
        (self.root / "photo.png").write_bytes(b"\x89PNG\r\n\x1a\n")
        (self.root / "document.pdf").write_bytes(b"%PDF-1.4\n%%EOF\n")

        self.outside = Path(self.temporary_directory.name) / "outside.txt"
        self.outside.write_text("outside\n", encoding="utf-8")
        (self.root / "escape-link").symlink_to(self.outside)
        self.browser = CodeBrowser(self.root)

    def tearDown(self):
        self.temporary_directory.cleanup()

    def _git(self, *arguments):
        return subprocess.run(
            ["git", "-C", str(self.root), *arguments],
            check=True,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

    def test_tree_is_lazy_sorted_and_includes_hidden_and_ignored_entries(self):
        result = self.browser.tree("")
        entries = result["entries"]
        by_name = {entry["name"]: entry for entry in entries}

        self.assertEqual(result["path"], "")
        self.assertIn(".hidden", by_name)
        self.assertIn("ignored.log", by_name)
        self.assertTrue(by_name[".hidden"]["hidden"])
        self.assertEqual(by_name["escape-link"]["type"], "symlink")
        self.assertNotIn("inside.c", by_name)
        types = [entry["type"] for entry in entries]
        first_non_directory = next((index for index, value in enumerate(types) if value != "directory"), len(types))
        self.assertTrue(all(value == "directory" for value in types[:first_non_directory]))
        names_within_groups = [entry["name"] for entry in entries if entry["type"] == "directory"]
        self.assertEqual(names_within_groups, sorted(names_within_groups, key=str.casefold))

    def test_paths_cannot_traverse_or_follow_an_escaping_symlink(self):
        with self.assertRaises(APIError) as traversal:
            self.browser.file("../outside.txt")
        self.assertEqual(traversal.exception.code, "path_escape")

        with self.assertRaises(APIError) as symlink:
            self.browser.file("escape-link")
        self.assertEqual(symlink.exception.code, "path_escape")

    def test_file_returns_text_and_media_metadata_and_rejects_other_binary(self):
        result = self.browser.file("root.py")
        self.assertEqual(result["kind"], "text")
        self.assertEqual(result["language"], "python")
        self.assertEqual(result["name"], "root.py")
        self.assertEqual(result["content"], "print('working')\n")
        self.assertEqual(result["changes"], [{"line": 1, "kind": "modified"}])
        self.assertEqual(result["modifiedNs"], str((self.root / "root.py").stat().st_mtime_ns))

        image = self.browser.file("diagram.svg")
        self.assertEqual(image["kind"], "image")
        self.assertEqual(image["mediaType"], "image/svg+xml")
        self.assertEqual(image["modifiedNs"], str((self.root / "diagram.svg").stat().st_mtime_ns))
        self.assertNotIn("content", image)

        pdf = self.browser.file("document.pdf")
        self.assertEqual(pdf["kind"], "pdf")
        self.assertEqual(pdf["mediaType"], "application/pdf")

        with self.assertRaises(APIError) as binary:
            self.browser.file("binary.bin")
        self.assertEqual(binary.exception.status, 415)
        self.assertEqual(binary.exception.code, "binary_file")

    def test_media_only_opens_allowlisted_files(self):
        path, content_type, size = self.browser.media("diagram.svg")
        self.assertEqual(path, (self.root / "diagram.svg").resolve())
        self.assertEqual(content_type, "image/svg+xml")
        self.assertEqual(size, len(self.svg_content))

        with self.assertRaises(APIError) as unsupported:
            self.browser.media("binary.bin")
        self.assertEqual(unsupported.exception.status, 415)
        self.assertEqual(unsupported.exception.code, "unsupported_media")

    def test_git_status_contains_worktree_staged_and_untracked_changes(self):
        result = self.browser.git_status()
        entries = {entry["path"]: entry for entry in result["entries"]}

        self.assertTrue(result["branch"])
        self.assertEqual(entries["root.py"]["worktreeStatus"], "M")
        self.assertEqual(entries["root.py"]["kind"], "modified")
        self.assertEqual(entries["staged.txt"]["indexStatus"], "A")
        self.assertEqual(entries["untracked.txt"]["kind"], "untracked")
        self.assertNotIn("ignored.log", entries)

    def test_search_supports_project_and_file_scope(self):
        project = self.browser.search("working")
        by_path = {result["path"]: result for result in project["results"]}
        self.assertEqual(project["scope"], "project")
        self.assertEqual(by_path["root.py"]["line"], 1)
        self.assertEqual(by_path["long.txt"]["line"], 15)
        self.assertNotIn("binary.bin", by_path)
        self.assertNotIn("escape-link", by_path)
        self.assertFalse(self.browser.search("<rect")["results"])
        self.assertFalse(self.browser.search("ignored content")["results"])

        current_file = self.browser.search("WORKING", "long.txt")
        self.assertEqual(current_file["scope"], "file")
        self.assertEqual(current_file["path"], "long.txt")
        self.assertEqual(current_file["results"][0]["line"], 15)
        self.assertFalse(self.browser.search("WORKING", "long.txt", match_case=True)["results"])
        self.assertTrue(self.browser.search("work", "long.txt")["results"])
        self.assertFalse(self.browser.search("work", "long.txt", whole_word=True)["results"])
        self.assertTrue(self.browser.search("working", "long.txt", whole_word=True)["results"])

        subtree = CodeBrowser(self.root / "adir")
        subtree_results = subtree.search("int value")["results"]
        self.assertEqual(subtree_results[0]["path"], "inside.c")

        with self.assertRaises(APIError) as invalid:
            self.browser.search("   ")
        self.assertEqual(invalid.exception.code, "invalid_query")

    def test_git_status_and_diff_handle_a_staged_rename_and_deletion(self):
        os.rename(self.root / "renamed.txt", self.root / "working-renamed.txt")
        self._git("add", "-A", "renamed.txt", "working-renamed.txt")
        (self.root / "adir" / "inside.c").unlink()

        entries = {entry["path"]: entry for entry in self.browser.git_status()["entries"]}
        rename = entries["working-renamed.txt"]
        self.assertEqual(rename["oldPath"], "renamed.txt")
        self.assertEqual(rename["kind"], "renamed")
        self.assertEqual(entries["adir/inside.c"]["kind"], "deleted")
        diff = self.browser.git_diff("working-renamed.txt")["diff"]
        self.assertIn("rename from renamed.txt", diff)
        self.assertIn("rename to working-renamed.txt", diff)

    def test_git_diff_covers_tracked_and_untracked_files(self):
        tracked = self.browser.git_diff("root.py")
        self.assertIn("-print('initial')", tracked["diff"])
        self.assertIn("+print('working')", tracked["diff"])
        self.assertEqual(tracked["status"], "M")
        self.assertEqual(tracked["additions"], 1)
        self.assertEqual(tracked["deletions"], 1)

        untracked = self.browser.git_diff("untracked.txt")
        self.assertIn("--- /dev/null", untracked["diff"])
        self.assertIn("+++ b/untracked.txt", untracked["diff"])
        self.assertIn("+untracked content", untracked["diff"])
        self.assertEqual(untracked["status"], "U")
        self.assertEqual(untracked["kind"], "untracked")
        self.assertEqual(untracked["additions"], 1)
        self.assertEqual(untracked["deletions"], 0)

        binary = self.browser.git_diff("binary.bin")
        self.assertEqual(binary["diff"], "Binary files /dev/null and b/binary.bin differ\n")
        self.assertEqual(binary["additions"], 0)
        self.assertEqual(binary["deletions"], 0)

    def test_git_diff_context_defaults_to_full_and_supports_compact(self):
        full = self.browser.git_diff("long.txt")["diff"]
        compact = self.browser.git_diff("long.txt", "compact")["diff"]
        self.assertIn(" file start sentinel", full)
        self.assertNotIn(" file start sentinel", compact)

        commit_full = self.browser.git_commit_diff(self.rename_oid, "long.txt")["diff"]
        commit_compact = self.browser.git_commit_diff(self.rename_oid, "long.txt", "compact")["diff"]
        self.assertIn(" file start sentinel", commit_full)
        self.assertNotIn(" file start sentinel", commit_compact)

        with self.assertRaises(APIError) as invalid:
            self.browser.git_diff("long.txt", "everything")
        self.assertEqual(invalid.exception.code, "invalid_context")

    def test_diff_line_changes_classifies_replacements_additions_and_deletions(self):
        diff = """@@ -1,3 +1,4 @@
 same
-old one
+replacement
+addition
 tail
"""
        self.assertEqual(
            diff_line_changes(diff, 4),
            [
                {"line": 2, "kind": "modified"},
                {"line": 3, "kind": "added"},
            ],
        )

        deletion = """@@ -1,2 +1 @@
 keep
-gone
"""
        self.assertEqual(diff_line_changes(deletion, 1), [{"line": 1, "kind": "deleted"}])

    def test_commit_list_details_root_commit_and_diffs(self):
        commits = self.browser.git_commits(100)["commits"]
        self.assertEqual([item["subject"] for item in commits[:2]], ["Rename file", "Initial files"])

        initial = self.browser.git_commit(self.initial_oid[:8])
        initial_files = {item["path"]: item for item in initial["files"]}
        self.assertEqual(initial["oid"], self.initial_oid)
        self.assertEqual(initial_files["root.py"]["status"], "A")
        root_diff = self.browser.git_commit_diff(self.initial_oid, "root.py")
        self.assertIn("new file mode", root_diff["diff"])
        self.assertIn("+print('initial')", root_diff["diff"])
        self.assertEqual(root_diff["status"], "A")
        self.assertEqual(root_diff["additions"], 1)
        self.assertEqual(root_diff["deletions"], 0)

        renamed = self.browser.git_commit(self.rename_oid)
        rename_file = next(item for item in renamed["files"] if item["path"] == "renamed.txt")
        self.assertEqual(rename_file["oldPath"], "rename-me.txt")
        self.assertEqual(rename_file["status"], "R")
        rename_diff = self.browser.git_commit_diff(self.rename_oid, "renamed.txt")["diff"]
        self.assertIn("rename from rename-me.txt", rename_diff)
        self.assertIn("rename to renamed.txt", rename_diff)

        subtree = CodeBrowser(self.root / "adir")
        escaped_pathspec = subtree.git_commit_diff(self.initial_oid, ":(top)root.py")["diff"]
        self.assertEqual(escaped_pathspec, "")

    def test_http_api_returns_structured_security_errors(self):
        browser = self.browser

        class QuietHandler(CodeBrowserHandler):
            def log_message(self, format, *args):
                pass

        QuietHandler.browser = browser
        server = ReusableThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        base_url = f"http://127.0.0.1:{server.server_port}"
        try:
            with urllib.request.urlopen(base_url + "/api/info") as response:
                info = json.load(response)
            self.assertEqual(info["path"], str(self.root.resolve()))
            self.assertEqual(info["root"], str(self.root.resolve()))
            self.assertTrue(info["isGit"])
            self.assertTrue(info["gitAvailable"])

            with self.assertRaises(urllib.error.HTTPError) as traversal:
                urllib.request.urlopen(base_url + "/api/file?path=..%2Foutside.txt")
            self.assertEqual(traversal.exception.code, 403)
            payload = json.load(traversal.exception)
            self.assertEqual(payload["code"], "path_escape")
            self.assertIsInstance(payload["error"], str)

            with urllib.request.urlopen(base_url + "/api/media?path=diagram.svg") as response:
                self.assertEqual(response.headers["Content-Type"], "image/svg+xml")
                self.assertEqual(response.headers["X-Content-Type-Options"], "nosniff")
                self.assertIn("sandbox", response.headers["Content-Security-Policy"])
                self.assertEqual(response.read(), self.svg_content)

            with urllib.request.urlopen(base_url + "/api/media?path=document.pdf") as response:
                self.assertEqual(response.headers["Content-Type"], "application/pdf")
                self.assertEqual(response.read(), b"%PDF-1.4\n%%EOF\n")

            with self.assertRaises(urllib.error.HTTPError) as unsupported_media:
                urllib.request.urlopen(base_url + "/api/media?path=binary.bin")
            self.assertEqual(unsupported_media.exception.code, 415)
            self.assertEqual(json.load(unsupported_media.exception)["code"], "unsupported_media")

            with self.assertRaises(urllib.error.HTTPError) as static_escape:
                urllib.request.urlopen(base_url + "/../server.py")
            self.assertEqual(static_escape.exception.code, 404)

            static_routes = (
                ("/", "text/html"),
                ("/favicon.ico", "image/"),
                ("/favicon.svg", "image/svg+xml"),
                ("/code-lens-v4-16.png", "image/png"),
                ("/code-lens-v4-32.png", "image/png"),
                ("/code-lens-v4-180.png", "image/png"),
                ("/style.css", "text/css"),
                ("/static/js/api.js", "javascript"),
                ("/static/js/app.js", "javascript"),
                ("/static/js/highlighter.js", "javascript"),
                ("/static/js/icons.js", "javascript"),
                ("/static/js/sidebar.js", "javascript"),
                ("/static/js/viewer.js", "javascript"),
                ("/static/vendor/highlightjs/highlight.min.js", "javascript"),
                ("/static/vendor/highlightjs/cmake.min.js", "javascript"),
                ("/static/vendor/diff2html/diff2html-ui-base.min.js", "javascript"),
                ("/static/vendor/diff2html/diff2html.min.css", "text/css"),
                ("/static/vendor/pdfjs/pdf.min.mjs", "javascript"),
                ("/static/vendor/pdfjs/pdf.worker.min.mjs", "javascript"),
                ("/static/vendor/seti/seti.woff", "font/woff"),
                ("/static/vendor/seti/vs-seti-icon-theme.json", "application/json"),
            )
            for route, content_type in static_routes:
                with urllib.request.urlopen(base_url + route) as response:
                    self.assertIn(content_type, response.headers["Content-Type"])
                    self.assertEqual(response.headers["Cache-Control"], "no-store")
                    response.read()
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=2)


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