quill/tests/test_quill.py

import http.client
import json
import sys
import tempfile
import threading
import unittest
import urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
import quill_core  # noqa: E402
import server  # noqa: E402

FILES = {
    "README.md": "# demo\n\nA demo space.\n",
    "posts/2026-10-01-hello.md": "---\ntitle: Hello\nauthor: test-agent\n---\nFirst **post**. See [the other](second.md).\n",
    "posts/second.md": "# Second post\n\nPlain text <script>alert(1)</script>\n",
}


class FakeGitea(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_GET(self):
        p = urllib.parse.urlsplit(self.path)
        path = p.path
        def out(obj, raw=False):
            body = obj.encode() if raw else json.dumps(obj).encode()
            self.send_response(200); self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
        if path == "/api/v1/repos/demo/space":
            return out({"default_branch": "main"})
        if path == "/api/v1/repos/demo/space/branches/main":
            return out({"commit": {"id": "abc123def456", "message": "add posts", "author": {"name": "test-agent"}, "timestamp": "2026-10-01T00:00:00Z"}})
        if path.startswith("/api/v1/repos/demo/space/git/trees/"):
            return out({"tree": [{"path": k, "type": "blob"} for k in FILES]})
        if path.startswith("/api/v1/repos/demo/space/raw/"):
            f = urllib.parse.unquote(path[len("/api/v1/repos/demo/space/raw/"):])
            if f in FILES:
                return out(FILES[f], raw=True)
        self.send_response(404); self.send_header("Content-Length", "0"); self.end_headers()


def serve(handler):
    srv = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv


class QuillTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.gitea = serve(FakeGitea)
        cls.tmp = tempfile.TemporaryDirectory()
        cfg = {
            "name": "Test Quill", "gitea_url": f"http://127.0.0.1:{cls.gitea.server_port}",
            "git_web_base": "https://git.example.test", "access": "code",
            "view_code_sha256": server.hash_code("letmein"), "data_dir": cls.tmp.name,
            "nav": [{"label": "Git", "href": "https://git.example.test/"}],
            "spaces": [{"key": "demo", "name": "Demo", "repo": "demo/space", "description": "d"}],
        }
        p = Path(cls.tmp.name) / "config.json"
        p.write_text(json.dumps(cfg))
        server.CONFIG = server.load_config(p)
        server._SECRET = server._secret(Path(cls.tmp.name))
        cls.srv = serve(server.Handler)

    @classmethod
    def tearDownClass(cls):
        cls.srv.shutdown(); cls.gitea.shutdown(); cls.tmp.cleanup()

    def req(self, method, path, body=None, headers=None):
        c = http.client.HTTPConnection("127.0.0.1", self.srv.server_port, timeout=10)
        c.request(method, path, body=body, headers=headers or {})
        r = c.getresponse()
        return r.status, dict(r.getheaders()), r.read()

    def test_core_listing_and_post(self):
        status, data = quill_core.api("/v1/quill/spaces/demo", server.gitea_get)
        self.assertEqual(status, 200)
        titles = [p["title"] for p in data["posts"]]
        self.assertIn("Hello", titles)
        self.assertIn("Second post", titles)
        self.assertTrue(all("html" not in p for p in data["posts"]))
        slug = next(p["slug"] for p in data["posts"] if p["title"] == "Second post")
        status, data = quill_core.api("/v1/quill/spaces/demo/posts/" + slug, server.gitea_get)
        self.assertEqual(status, 200)
        self.assertNotIn("<script>", data["post"]["html"])
        self.assertTrue(data["post"]["git_url"].startswith("https://git.example.test/demo/space/src/branch/main/"))

    def test_unknown_space(self):
        self.assertEqual(quill_core.api("/v1/quill/spaces/nope", server.gitea_get)[0], 404)

    def test_bad_config(self):
        with self.assertRaises(ValueError):
            quill_core.configure([{"key": "Bad Key", "repo": "a/b"}])
        quill_core.configure([{"key": "demo", "name": "Demo", "repo": "demo/space"}], "https://git.example.test")

    def test_view_code_flow(self):
        s, h, _ = self.req("GET", "/quill")
        self.assertEqual(s, 302)
        self.assertTrue(h["Location"].startswith("/view?next="))
        self.assertEqual(self.req("GET", "/v1/quill")[0], 401)
        self.assertEqual(self.req("GET", "/v1/quill", headers={"X-View-Code": "letmein"})[0], 200)
        s, _, _ = self.req("POST", "/view", body="code=wrong&next=/quill", headers={"Content-Type": "application/x-www-form-urlencoded"})
        self.assertEqual(s, 401)
        s, h, _ = self.req("POST", "/view", body="code=letmein&next=/quill/demo", headers={"Content-Type": "application/x-www-form-urlencoded"})
        self.assertEqual(s, 303)
        self.assertEqual(h["Location"], "/quill/demo")
        cookie = h["Set-Cookie"].split(";")[0]
        s, _, body = self.req("GET", "/quill/demo", headers={"Cookie": cookie})
        self.assertEqual(s, 200)
        self.assertIn(b"Test Quill", body)
        self.assertIn(b"https://git.example.test/", body)
        s, _, body = self.req("GET", "/v1/quill/spaces", headers={"Cookie": cookie})
        self.assertEqual(json.loads(body)["spaces"][0]["key"], "demo")

    def test_open_redirect_blocked(self):
        s, h, _ = self.req("POST", "/view", body="code=letmein&next=//evil.example", headers={"Content-Type": "application/x-www-form-urlencoded"})
        self.assertEqual(h["Location"], "/quill")

    def test_read_only(self):
        for m in ("PUT", "PATCH", "DELETE"):
            self.assertEqual(self.req(m, "/v1/quill/spaces/demo", body=b"{}")[0], 405)
        self.assertEqual(self.req("POST", "/v1/quill", body=b"{}")[0], 405)

    def test_health(self):
        self.assertEqual(self.req("GET", "/health")[0], 200)


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