187 lines
7.7 KiB
Python
187 lines
7.7 KiB
Python
"""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":
|
|
"[email protected]", "email":
|
|
"[email protected]", "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": "[email protected]"}),
|
|
"alice.smith")
|
|
|
|
def test_email_fallback(self):
|
|
self.assertEqual(
|
|
granthi_link.LinkService.derive_login({"email": "[email protected]"}),
|
|
"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()
|