"""Unit tests for granthi-link: login derivation, link/repos flows against a stub HTTP server that plays both Zitadel userinfo and the Gitea API.""" import json import os import sys import threading import unittest import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "server")) import granthi_link # noqa: E402 class StubUpstream(BaseHTTPRequestHandler): """Plays Zitadel (/oidc/v1/userinfo) and Gitea (everything else).""" state = None # dict injected per-test def _json(self, status, obj): payload = json.dumps(obj).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def do_GET(self): st = self.state if self.path == "/oidc/v1/userinfo": auth = self.headers.get("Authorization", "") if auth == "Bearer good-token": return self._json(200, {"sub": "123", "preferred_username": "Alice.Smith@org.example", "email": "alice@example.com", "name": "Alice"}) return self._json(401, {"error": "invalid token"}) if self.path.startswith("/api/v1/users/"): login = self.path.rsplit("/", 1)[1] if login in st["users"]: return self._json(200, {"login": login}) return self._json(404, {"message": "not found"}) self._json(404, {}) def do_POST(self): st = self.state length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length) or b"{}") if self.path == "/api/v1/admin/users": st["users"].add(body["username"]) st["created"].append(body) return self._json(201, {"login": body["username"]}) if self.path.startswith("/api/v1/users/") and self.path.endswith("/tokens"): # must arrive with basic auth + Sudo (the verified 1.27 mechanism) st["token_reqs"].append({ "auth": self.headers.get("Authorization", ""), "sudo": self.headers.get("Sudo", ""), "body": body}) if not self.headers.get("Authorization", "").startswith("Basic "): return self._json(401, {"message": "auth required"}) return self._json(201, {"sha1": "MINTED", "name": body["name"]}) if self.path == "/api/v1/user/repos": if body["name"] in st["repos"]: return self._json(409, {"message": "exists"}) st["repos"].add(body["name"]) return self._json(201, {"name": body["name"], "private": body.get("private"), "full_name": f"alice/{body['name']}"}) self._json(404, {}) def log_message(self, *a): pass class ServiceTestBase(unittest.TestCase): def setUp(self): StubUpstream.state = {"users": set(), "created": [], "repos": set(), "token_reqs": []} self.upstream = ThreadingHTTPServer(("127.0.0.1", 0), StubUpstream) threading.Thread(target=self.upstream.serve_forever, daemon=True).start() self.addCleanup(self.upstream.shutdown) base = f"http://127.0.0.1:{self.upstream.server_address[1]}" self.svc = granthi_link.LinkService({ "gitea_base": base, "public_gitea_base": "http://public.example:3041", "zitadel_userinfo": f"{base}/oidc/v1/userinfo", "admin_token": "ADMTOK", "admin_login": "root", "admin_password": "rootpw", "test_mode": False, }) class TestDeriveLogin(unittest.TestCase): def test_strips_domain_and_sanitizes(self): self.assertEqual( granthi_link.LinkService.derive_login( {"preferred_username": "Alice.Smith@org.example"}), "alice.smith") def test_email_fallback(self): self.assertEqual( granthi_link.LinkService.derive_login({"email": "Bob+x@e.com"}), "bob-x") def test_empty_returns_none(self): self.assertIsNone(granthi_link.LinkService.derive_login({})) class TestLink(ServiceTestBase): def test_link_creates_user_and_mints_token(self): status, resp = self.svc.link({"zitadel_access_token": "good-token", "device_name": "mac studio"}) self.assertEqual(status, 200) self.assertEqual(resp["login"], "alice.smith") self.assertEqual(resp["token"], "MINTED") self.assertEqual(resp["gitea_base"], "http://public.example:3041") st = StubUpstream.state self.assertEqual(len(st["created"]), 1) created = st["created"][0] self.assertFalse(created["must_change_password"]) self.assertEqual(created["visibility"], "private") self.assertGreaterEqual(len(created["password"]), 30) req = st["token_reqs"][0] self.assertTrue(req["auth"].startswith("Basic ")) self.assertEqual(req["sudo"], "alice.smith") self.assertEqual(sorted(req["body"]["scopes"]), ["write:repository", "write:user"]) def test_link_existing_user_skips_create(self): StubUpstream.state["users"].add("alice.smith") status, resp = self.svc.link({"zitadel_access_token": "good-token", "device_name": "d"}) self.assertEqual(status, 200) self.assertEqual(StubUpstream.state["created"], []) def test_link_bad_token_401(self): status, resp = self.svc.link({"zitadel_access_token": "bad", "device_name": "d"}) self.assertEqual(status, 401) def test_link_missing_token_400(self): status, _ = self.svc.link({"device_name": "d"}) self.assertEqual(status, 400) def test_test_mode_stub_only_when_enabled(self): # disabled -> stub ignored, token required status, _ = self.svc.link({"test_userinfo": {"sub": "1", "preferred_username": "x"}}) self.assertEqual(status, 400) self.svc.cfg["test_mode"] = True status, resp = self.svc.link({"test_userinfo": { "sub": "1", "preferred_username": "evetest"}, "device_name": "d"}) self.assertEqual(status, 200) self.assertEqual(resp["login"], "evetest") class TestRepos(ServiceTestBase): def test_repo_create_returns_public_clone_url(self): status, resp = self.svc.repos({"token": "USERTOK", "name": "notes", "private": True}) self.assertEqual(status, 200) self.assertEqual(resp["clone_url"], "http://public.example:3041/alice/notes.git") def test_repo_conflict_409(self): self.svc.repos({"token": "T", "name": "notes"}) status, _ = self.svc.repos({"token": "T", "name": "notes"}) self.assertEqual(status, 409) def test_missing_fields_400(self): status, _ = self.svc.repos({"name": "x"}) self.assertEqual(status, 400) class TestHealth(ServiceTestBase): def test_health_endpoint(self): granthi_link.Handler.service = self.svc srv = ThreadingHTTPServer(("127.0.0.1", 0), granthi_link.Handler) threading.Thread(target=srv.serve_forever, daemon=True).start() self.addCleanup(srv.shutdown) with urllib.request.urlopen( f"http://127.0.0.1:{srv.server_address[1]}/health") as r: body = json.loads(r.read()) self.assertEqual(body["status"], "ok") self.assertEqual(body["service"], "granthi-link") if __name__ == "__main__": unittest.main()