commit 20775917555312a2645f5eb1d20a2b6b3c0e4b4c Author: Nirav Patel Date: Wed Aug 19 00:09:19 2026 -0400 granthi-sync v1: granthi-link provisioning service + client daemon + tests Co-Authored-By: Claude Fable 5 diff --git a/README.md b/README.md new file mode 100644 index 0000000..cac3d62 --- /dev/null +++ b/README.md @@ -0,0 +1,122 @@ +# granthi-sync v1 + +The signup → download → link-folders → cloud product spine for the Granthi +forge, tested against the BETA forge (granthi-beta.shre.ai). Python 3 stdlib + +git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh. + +``` + ┌──────────────┐ device flow ┌─────────────────┐ + │ granthi-sync │ ───────────────▶ │ shre-id Zitadel │ + │ (client, │ ◀─────────────── │ id.shre.ai │ + │ Mac/laptop) │ access token └─────────────────┘ + │ │ + │ │ POST /v1/link {zitadel_access_token, device_name} + │ │ ───────────────▶ ┌───────────────────────────────┐ + │ │ ◀─────────────── │ granthi-link :3042 │ + │ │ {login, token} │ (granthi VPS, tailnet-only) │ + │ │ │ · userinfo validation │ + │ │ POST /v1/repos │ · ensure Gitea user (admin) │ + │ │ ───────────────▶ │ · mint scoped user token │ + │ │ └──────────────┬────────────────┘ + │ │ git push/fetch (user token │ admin API + │ │ via credential helper) ▼ + │ │ ───────────────▶ ┌───────────────────────────────┐ + └──────────────┘ │ BETA forge :3041 │ + │ granthi-beta.shre.ai │ + └───────────────────────────────┘ +``` + +## Components + +### `server/granthi_link.py` — provisioning service (granthi VPS) + +* `GET /health` +* `POST /v1/link {zitadel_access_token, device_name}` → validates the token + against `https://id.shre.ai/oidc/v1/userinfo`, derives a login from + `preferred_username` (email local-part fallback), ensures the Gitea user + exists, mints a token scoped `write:repository,write:user`, returns + `{gitea_base, login, token, token_name}`. +* `POST /v1/repos {token, name, private}` → creates the user repo with the + USER token; clone/html URLs are rebased onto `public_gitea_base` because the + container `ROOT_URL` (https://granthi-beta.shre.ai) does not resolve for + tailnet-only clients. + +Deployment: `/opt/granthi-link/{granthi_link.py,config.json}` + systemd unit +`granthi-link.service`; binds `127.0.0.1:3042` **and** `100.111.127.127:3042` +(tailnet). **Not publicly exposed** — see promotion window. + +Empirically verified mechanics on Gitea **1.27.1** (beta forge): + +* Token minting: `POST /api/v1/users/{login}/tokens` returns **401 for + token-authenticated sudo** (both `Sudo:` header and `?sudo=`); it only works + with **admin basic auth + `Sudo: ` header** (201). The config + therefore carries `admin_login`/`admin_password` (root-only, 0600) in + addition to `admin_token` (minted via + `gitea admin user generate-access-token`, used for all other admin calls). +* User creation: `source_id` is omitted → **local user** with a random + 30-char password, `must_change_password=false`, `visibility=private`. + Rationale: `/api/v1/admin/identity-auth-sources` 404s on 1.27.1; the + shre-id OAuth2 source is ID 1 (via `gitea admin auth list`), but users + attached to an OAuth2 source cannot basic-auth and admin-created users get + no `external_login_user` row anyway — first OIDC web login links by email + regardless of this choice. +* `test_mode` (config flag, **never in production**): allows `/v1/link` to + accept `test_userinfo` in the body instead of a Zitadel round-trip, so E2E + can exercise the ensure-user + mint path headlessly. + +### `client/granthi_sync_client.py` (+ `bin/granthi-sync`) — client daemon + +* `link [--server URL] [--token TOK]` — Zitadel **device flow** (native app + `granthi-sync-device`, client_id `386909715541590022`, project + granthi-forge `386906525790109702`; grants: device_code + refresh_token): + prints the verification URL + user code, polls the token endpoint + (`authorization_pending`/`slow_down` handled), then calls `/v1/link`. + `--token` skips the device flow with a ready Zitadel token (headless/dev). + Result stored in `~/.granthi-sync/config.json` (0600). +* `add [--name N] [--private|--public]` — `git init -b main` if + needed, creates the cloud repo via `/v1/repos`, adds remote `granthi`, + initial commit + push. The token is delivered by a **git credential + helper** (the client's hidden `git-credential` subcommand reading the 0600 + config) — never embedded in the remote URL (estate rule). +* `watch [--interval 30] [--once]` — per folder: autocommit + (`sync: `) → fetch → ff-pull if remote strictly ahead → push if + local strictly ahead. **DIVERGED → log + record + SKIP. Never force, never + merge** — the same policy as the mesh. SIGTERM-clean. +* `status` — table of linked folders, last sync, divergence flags. +* Run as a daemon on macOS with `client/launchd/ai.granthi.sync.plist` + (edit the script path, then `launchctl bootstrap gui/$UID `). + +## Tests + +* `python3 -m unittest discover -s tests` — 24 tests: autocommit/ff/diverged + logic against real temp git repos (including "diverged never touches the + remote"), config 0600 handling, mocked device-flow polling, credential + helper protocol, and the full `/v1/link` + `/v1/repos` service flows + against an in-process stub playing Zitadel + Gitea. +* Live E2E against the beta forge is recorded in the delivery notes + (link → add → watch ff/push → forced divergence → DIVERGED skip verified + via API, remote sha untouched). + +## Invite-only story (v1) + +There is no open signup. An operator invites a user by creating them in +shre-id (Zitadel org). The user downloads the client, runs +`granthi-sync link`, signs in at id.shre.ai with the printed device code, and +the provisioning service creates their forge account + scoped token on the +fly — the forge never sees a password and the user never sees the forge admin. +Every linked folder becomes a private repo under their account. + +## Promotion window (beta → prod) + +1. **Expose :3042** behind cloudflared (granthi.shre.ai vhost or + link.granthi.shre.ai) — today it is tailnet-only by design. +2. **Swap forge base URLs** in `/opt/granthi-link/config.json`: + `gitea_base` → prod forge, `public_gitea_base` → + `https://granthi.shre.ai`; the client default server URL moves to the + public endpoint. +3. The `granthi-web` OIDC app already lists the prod callback; the device + app is host-independent. Rotate the beta admin token/password out of the + config when pointing at prod (prod forge is READ-ONLY to this estate — + promotion is an operator action, not an agent action). +4. Add rate limiting / abuse controls before public exposure (one token mint + per link call today). diff --git a/bin/granthi-sync b/bin/granthi-sync new file mode 100755 index 0000000..006c05c --- /dev/null +++ b/bin/granthi-sync @@ -0,0 +1,4 @@ +#!/bin/sh +# Thin wrapper: run the granthi-sync client from anywhere. +DIR="$(cd "$(dirname "$0")/.." && pwd)" +exec python3 "$DIR/client/granthi_sync_client.py" "$@" diff --git a/client/__pycache__/granthi_sync_client.cpython-314.pyc b/client/__pycache__/granthi_sync_client.cpython-314.pyc new file mode 100644 index 0000000..644b42f Binary files /dev/null and b/client/__pycache__/granthi_sync_client.cpython-314.pyc differ diff --git a/client/granthi_sync_client.py b/client/granthi_sync_client.py new file mode 100644 index 0000000..70b373b --- /dev/null +++ b/client/granthi_sync_client.py @@ -0,0 +1,420 @@ +#!/usr/bin/env python3 +"""granthi-sync client daemon. + +Commands: + link [--server URL] [--token ZITADEL_TOKEN] [--device NAME] + Zitadel OAuth device flow against https://id.shre.ai (or --token for + the headless path), then POST /v1/link on the granthi-link service. + Stores {server, gitea_base, login, token, token_name} in + ~/.granthi-sync/config.json (0600). + add [--name N] [--private/--public] + git init (branch main) if needed, create the cloud repo via + /v1/repos, add remote 'granthi', initial commit + push. The token + is supplied via a git credential helper (this script's hidden + `git-credential` subcommand), never embedded in the remote URL. + watch [--interval 30] [--once] + Poll linked folders: autocommit local changes, fetch, ff-pull when + the remote is strictly ahead, push when local is strictly ahead. + DIVERGED branches are logged + recorded and SKIPPED -- never force, + never merge (same policy as the estate's gitea_sync.py mesh). + SIGTERM-clean. + status + Table of linked folders, last sync, divergence flags. + +Stdlib + git CLI only. +""" + +import argparse +import base64 +import json +import os +import re +import signal +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from datetime import datetime, timezone + +VERSION = "1.0.0" + +CONFIG_DIR = os.path.expanduser(os.environ.get("GRANTHI_SYNC_HOME", "~/.granthi-sync")) +CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json") + +ZITADEL_BASE = "https://id.shre.ai" +DEVICE_CLIENT_ID = "386909715541590022" +# ^ Zitadel native app "granthi-sync-device" (appId 386909715541524486) in +# project granthi-forge (386906525790109702); device-code + refresh grants. +DEFAULT_SERVER = "http://100.111.127.127:3042" +DEVICE_SCOPE = "openid profile email" + + +def log(msg): + ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + print(f"[{ts}] {msg}", flush=True) + + +# -------------------------------------------------------------------------- +# HTTP helpers (patchable in tests) +# -------------------------------------------------------------------------- + +def http_json(method, url, headers=None, body=None, form=None, timeout=30): + """Returns (status, parsed-json-or-{'raw': text}).""" + data = None + hdrs = dict(headers or {}) + if form is not None: + data = urllib.parse.urlencode(form).encode() + hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded") + elif body is not None: + data = json.dumps(body).encode() + hdrs.setdefault("Content-Type", "application/json") + req = urllib.request.Request(url, data=data, headers=hdrs, method=method) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw, status = resp.read(), resp.status + except urllib.error.HTTPError as e: + raw, status = e.read(), e.code + except (urllib.error.URLError, OSError) as e: + return 599, {"error": str(e)} + try: + return status, json.loads(raw) if raw else {} + except ValueError: + return status, {"raw": raw.decode(errors="replace")} + + +# -------------------------------------------------------------------------- +# Config +# -------------------------------------------------------------------------- + +def load_config(): + try: + with open(CONFIG_PATH) as f: + return json.load(f) + except FileNotFoundError: + return {} + + +def save_config(cfg): + os.makedirs(CONFIG_DIR, mode=0o700, exist_ok=True) + tmp = CONFIG_PATH + ".tmp" + with open(tmp, "w") as f: + json.dump(cfg, f, indent=2) + os.chmod(tmp, 0o600) + os.replace(tmp, CONFIG_PATH) + + +# -------------------------------------------------------------------------- +# git helpers +# -------------------------------------------------------------------------- + +def git(folder, *args, check=True): + """Run git in folder. Returns (rc, stdout). Never uses --force.""" + proc = subprocess.run(["git", "-C", folder] + list(args), + capture_output=True, text=True) + if check and proc.returncode != 0: + raise RuntimeError( + f"git {' '.join(args)} failed in {folder}: {proc.stderr.strip()}") + return proc.returncode, proc.stdout.strip() + + +def ensure_repo(folder): + if os.path.isdir(os.path.join(folder, ".git")): + return + rc, _ = git(folder, "init", "-b", "main", check=False) + if rc != 0: # git < 2.28 fallback + git(folder, "init") + git(folder, "symbolic-ref", "HEAD", "refs/heads/main") + + +def autocommit(folder): + """Commit all local changes as 'sync: '. Returns True if a + commit was made.""" + _, status = git(folder, "status", "--porcelain") + if not status: + return False + git(folder, "add", "-A") + ts = datetime.now(timezone.utc).isoformat(timespec="seconds") + git(folder, "commit", "-m", f"sync: {ts}") + return True + + +def sync_state(folder, remote="granthi", branch="main"): + """Classify local vs remote after a fetch. + + Returns one of: 'no-remote-branch', 'in-sync', 'local-ahead', + 'remote-ahead-ff', 'diverged'. + """ + rc, _ = git(folder, "rev-parse", "--verify", "--quiet", + f"refs/remotes/{remote}/{branch}", check=False) + if rc != 0: + return "no-remote-branch" + _, local = git(folder, "rev-parse", branch) + _, remote_sha = git(folder, "rev-parse", f"refs/remotes/{remote}/{branch}") + if local == remote_sha: + return "in-sync" + rc, base = git(folder, "merge-base", branch, + f"refs/remotes/{remote}/{branch}", check=False) + if rc != 0: + return "diverged" # unrelated histories: treat as divergence + base = base.strip() + if base == remote_sha: + return "local-ahead" + if base == local: + return "remote-ahead-ff" + return "diverged" + + +def sync_folder(folder, remote="granthi", branch="main"): + """One sync pass for one folder. Returns (outcome, detail). + + outcome in {'synced', 'pushed', 'clean', 'diverged', 'error'}. + Policy: ff-only pulls, plain pushes. NEVER --force, NEVER merge. + """ + try: + committed = autocommit(folder) + git(folder, "fetch", remote) + state = sync_state(folder, remote, branch) + ff_pulled = False + if state == "remote-ahead-ff": + git(folder, "merge", "--ff-only", f"refs/remotes/{remote}/{branch}") + ff_pulled = True + state = sync_state(folder, remote, branch) + if state == "diverged": + return "diverged", "local and remote both advanced; skipping (no force, no merge)" + if state in ("local-ahead", "no-remote-branch"): + git(folder, "push", "-u", remote, branch) + return "pushed", "committed+pushed" if committed else "pushed" + # state == "in-sync" + if ff_pulled: + return "synced", "ff-pulled" + return "clean", "in-sync" + except RuntimeError as e: + return "error", str(e) + + +# -------------------------------------------------------------------------- +# credential helper (git calls back into this script; token never in URL) +# -------------------------------------------------------------------------- + +def cmd_git_credential(argv): + op = argv[0] if argv else "get" + if op != "get": + return 0 # ignore store/erase + attrs = {} + for line in sys.stdin: + line = line.strip() + if not line: + break + k, _, v = line.partition("=") + attrs[k] = v + cfg = load_config() + gitea = cfg.get("gitea_base", "") + want = urllib.parse.urlparse(gitea) + if attrs.get("host") == want.netloc: + print(f"username={cfg.get('login', '')}") + print(f"password={cfg.get('token', '')}") + return 0 + + +def install_credential_helper(folder): + helper = f"!{sys.executable} {os.path.abspath(__file__)} git-credential" + git(folder, "config", "credential.helper", helper) + + +# -------------------------------------------------------------------------- +# commands +# -------------------------------------------------------------------------- + +def device_flow(): + """Zitadel OAuth 2.0 device authorization grant. Returns access token.""" + status, resp = http_json( + "POST", f"{ZITADEL_BASE}/oauth/v2/device_authorization", + form={"client_id": DEVICE_CLIENT_ID, "scope": DEVICE_SCOPE}) + if status != 200: + raise SystemExit(f"device authorization failed (HTTP {status}): {resp}") + print(f"\nTo link this device, open:\n\n {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n") + print(f"and enter code: {resp['user_code']}\n") + interval = int(resp.get("interval", 5)) + deadline = time.time() + int(resp.get("expires_in", 300)) + while time.time() < deadline: + time.sleep(interval) + status, tok = http_json( + "POST", f"{ZITADEL_BASE}/oauth/v2/token", + form={"grant_type": "urn:ietf:params:oauth:grant-type:device_code", + "device_code": resp["device_code"], + "client_id": DEVICE_CLIENT_ID}) + if status == 200: + return tok["access_token"] + err = tok.get("error", "") + if err == "authorization_pending": + continue + if err == "slow_down": + interval += 5 + continue + raise SystemExit(f"device flow failed: {tok}") + raise SystemExit("device flow timed out (code expired)") + + +def cmd_link(args): + zitadel_token = args.token or device_flow() + status, resp = http_json( + "POST", f"{args.server.rstrip('/')}/v1/link", + body={"zitadel_access_token": zitadel_token, + "device_name": args.device}) + if status != 200: + raise SystemExit(f"link failed (HTTP {status}): {resp}") + cfg = load_config() + cfg.update({"server": args.server.rstrip("/"), + "gitea_base": resp["gitea_base"], + "login": resp["login"], + "token": resp["token"], + "token_name": resp["token_name"]}) + cfg.setdefault("folders", {}) + save_config(cfg) + log(f"linked as {resp['login']} on {resp['gitea_base']} " + f"(token {resp['token_name']}); config: {CONFIG_PATH}") + return 0 + + +def cmd_add(args): + cfg = load_config() + if "token" not in cfg: + raise SystemExit("not linked yet -- run: granthi-sync link") + folder = os.path.abspath(args.folder) + if not os.path.isdir(folder): + raise SystemExit(f"no such folder: {folder}") + name = args.name or re.sub(r"[^a-zA-Z0-9._-]+", "-", os.path.basename(folder)) + ensure_repo(folder) + install_credential_helper(folder) + status, resp = http_json( + "POST", f"{cfg['server']}/v1/repos", + body={"token": cfg["token"], "name": name, "private": args.private}) + if status == 200: + clone_url = resp["clone_url"] + elif status == 409: + clone_url = f"{cfg['gitea_base']}/{cfg['login']}/{name}.git" + log(f"cloud repo {name} already exists; reusing") + else: + raise SystemExit(f"repo create failed (HTTP {status}): {resp}") + rc, _ = git(folder, "remote", "get-url", "granthi", check=False) + if rc == 0: + git(folder, "remote", "set-url", "granthi", clone_url) + else: + git(folder, "remote", "add", "granthi", clone_url) + autocommit(folder) + git(folder, "push", "-u", "granthi", "main") + cfg.setdefault("folders", {})[folder] = { + "name": name, "branch": "main", + "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "diverged": False} + save_config(cfg) + log(f"linked folder {folder} -> {clone_url}") + return 0 + + +_STOP = False + + +def _sigterm(signum, frame): + global _STOP + _STOP = True + log(f"signal {signum} received; finishing current pass then exiting") + + +def watch_pass(): + cfg = load_config() + for folder, meta in sorted(cfg.get("folders", {}).items()): + if not os.path.isdir(folder): + log(f"SKIP {folder}: missing") + continue + outcome, detail = sync_folder(folder, branch=meta.get("branch", "main")) + meta["diverged"] = outcome == "diverged" + if outcome in ("pushed", "synced"): + meta["last_sync"] = datetime.now(timezone.utc).isoformat(timespec="seconds") + if outcome == "diverged": + log(f"DIVERGED {folder}: {detail}") + elif outcome == "error": + log(f"ERROR {folder}: {detail}") + else: + log(f"{outcome} {folder}: {detail}") + save_config(cfg) + + +def cmd_watch(args): + signal.signal(signal.SIGTERM, _sigterm) + signal.signal(signal.SIGINT, _sigterm) + log(f"granthi-sync watch starting (interval {args.interval}s)") + while True: + watch_pass() + if args.once or _STOP: + break + deadline = time.time() + args.interval + while time.time() < deadline and not _STOP: + time.sleep(1) + if _STOP: + break + log("watch stopped cleanly") + return 0 + + +def cmd_status(args): + cfg = load_config() + if "login" in cfg: + print(f"linked: {cfg['login']} @ {cfg.get('gitea_base')} " + f"(token {cfg.get('token_name')})") + else: + print("not linked") + folders = cfg.get("folders", {}) + if not folders: + print("no folders linked") + return 0 + rows = [("FOLDER", "REPO", "LAST SYNC", "DIVERGED")] + for folder, meta in sorted(folders.items()): + rows.append((folder, meta.get("name", "?"), + meta.get("last_sync", "never"), + "YES" if meta.get("diverged") else "no")) + widths = [max(len(r[i]) for r in rows) for i in range(4)] + for r in rows: + print(" ".join(c.ljust(w) for c, w in zip(r, widths))) + return 0 + + +def main(argv=None): + p = argparse.ArgumentParser(prog="granthi-sync", + description="Granthi folder-to-cloud sync client") + p.add_argument("--version", action="version", version=VERSION) + sub = p.add_subparsers(dest="cmd", required=True) + + sp = sub.add_parser("link", help="link this device to your granthi account") + sp.add_argument("--server", default=DEFAULT_SERVER) + sp.add_argument("--token", help="ready Zitadel access token (headless path)") + sp.add_argument("--device", default=os.uname().nodename.split(".")[0]) + sp.set_defaults(fn=cmd_link) + + sp = sub.add_parser("add", help="link a folder and push it to the cloud") + sp.add_argument("folder") + sp.add_argument("--name") + grp = sp.add_mutually_exclusive_group() + grp.add_argument("--private", dest="private", action="store_true", default=True) + grp.add_argument("--public", dest="private", action="store_false") + sp.set_defaults(fn=cmd_add) + + sp = sub.add_parser("watch", help="sync loop over linked folders") + sp.add_argument("--interval", type=int, default=30) + sp.add_argument("--once", action="store_true", help="single pass then exit") + sp.set_defaults(fn=cmd_watch) + + sp = sub.add_parser("status", help="show linked folders") + sp.set_defaults(fn=cmd_status) + + argv = list(sys.argv[1:] if argv is None else argv) + if argv and argv[0] == "git-credential": # hidden helper protocol + return cmd_git_credential(argv[1:]) + args = p.parse_args(argv) + return args.fn(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/client/launchd/ai.granthi.sync.plist b/client/launchd/ai.granthi.sync.plist new file mode 100644 index 0000000..c884406 --- /dev/null +++ b/client/launchd/ai.granthi.sync.plist @@ -0,0 +1,19 @@ + + + + + Labelai.granthi.sync + ProgramArguments + + /usr/bin/python3 + /PATH/TO/granthi-sync/client/granthi_sync_client.py + watch + --interval + 30 + + RunAtLoad + KeepAlive + StandardOutPath/tmp/granthi-sync.log + StandardErrorPath/tmp/granthi-sync.log + + diff --git a/server/__pycache__/granthi_link.cpython-314.pyc b/server/__pycache__/granthi_link.cpython-314.pyc new file mode 100644 index 0000000..ae8b1d1 Binary files /dev/null and b/server/__pycache__/granthi_link.cpython-314.pyc differ diff --git a/server/config.example.json b/server/config.example.json new file mode 100644 index 0000000..6b6b016 --- /dev/null +++ b/server/config.example.json @@ -0,0 +1,10 @@ +{ + "gitea_base": "http://127.0.0.1:3041", + "public_gitea_base": "http://100.111.127.127:3041", + "zitadel_userinfo": "https://id.shre.ai/oidc/v1/userinfo", + "admin_token": "MINT-VIA: docker exec -u git gitea-beta-gitea-1 gitea admin user generate-access-token --username nirpa --scopes write:admin,write:user,write:repository --raw", + "admin_login": "nirpa", + "admin_password": "FROM /opt/gitea-beta/.admin-creds (required: Gitea 1.27 token minting only works via basic auth + Sudo header)", + "binds": [["127.0.0.1", 3042], ["100.111.127.127", 3042]], + "test_mode": false +} diff --git a/server/granthi-link.service b/server/granthi-link.service new file mode 100644 index 0000000..dc48585 --- /dev/null +++ b/server/granthi-link.service @@ -0,0 +1,17 @@ +[Unit] +Description=granthi-link provisioning service (granthi-sync spine) +After=network-online.target docker.service +Wants=network-online.target + +[Service] +Type=simple +ExecStart=/usr/bin/python3 /opt/granthi-link/granthi_link.py /opt/granthi-link/config.json +Restart=on-failure +RestartSec=3 +User=root +NoNewPrivileges=true +ProtectSystem=full +ReadWritePaths=/opt/granthi-link + +[Install] +WantedBy=multi-user.target diff --git a/server/granthi_link.py b/server/granthi_link.py new file mode 100644 index 0000000..fff0c34 --- /dev/null +++ b/server/granthi_link.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +"""granthi-link: provisioning service for the granthi-sync product spine. + +Bridges shre-id (Zitadel) identity to a Granthi (Gitea) forge: + + POST /v1/link {zitadel_access_token, device_name} + -> validates the token against Zitadel userinfo + -> ensures a Gitea user exists (admin API) + -> mints a scoped Gitea token for that user (admin basic + auth + `Sudo:` header -- the only mechanism that works + on Gitea 1.27; token-authenticated sudo returns 401) + -> {gitea_base, login, token, token_name} + + POST /v1/repos {token, name, private} -> creates a user repo with the + USER token, returns clone URLs rebased onto the public + gitea_base (container ROOT_URL may not resolve for + clients on the tailnet). + + GET /health -> {"status": "ok", ...} + +Design decisions (documented per spec): + * User creation uses source_id 0 (local). Gitea 1.27.1 has no + /api/v1/admin/identity-auth-sources endpoint (404); the shre-id OAuth2 + source is ID 1 (discovered via `gitea admin auth list`), but users + created against an OAuth2 source cannot basic-auth and admin-created + users get no external_login_user row anyway, so SSO linking happens on + first OIDC web login (by email) regardless. Local + random password is + the simplest correct v1. + * Token minting NEEDS the admin password (basic auth + Sudo header). + The admin API token alone cannot mint user tokens on 1.27. The config + therefore carries admin_login/admin_password alongside admin_token; + config must be root-owned 0600. + * test_mode: when config "test_mode" is true, a /v1/link body may carry + "test_userinfo" (dict) instead of a Zitadel round-trip. This exists so + E2E can exercise the ensure-user+mint path without a human OAuth login. + NEVER enable in production config. + +Stdlib only. Python 3.9+. +""" + +import base64 +import json +import logging +import os +import re +import secrets +import signal +import string +import sys +import threading +import urllib.error +import urllib.request +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +VERSION = "1.0.0" +LOG = logging.getLogger("granthi-link") + +DEFAULT_CONFIG = "/opt/granthi-link/config.json" + +LOGIN_SAFE = re.compile(r"[^a-zA-Z0-9._-]+") + + +# -------------------------------------------------------------------------- +# HTTP helper (patchable in tests) +# -------------------------------------------------------------------------- + +def http_json(method, url, headers=None, body=None, timeout=20): + """Minimal JSON-over-HTTP helper. Returns (status, parsed_or_text).""" + data = None + hdrs = dict(headers or {}) + if body is not None: + data = json.dumps(body).encode() + hdrs.setdefault("Content-Type", "application/json") + req = urllib.request.Request(url, data=data, headers=hdrs, method=method) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read() + status = resp.status + except urllib.error.HTTPError as e: + raw = e.read() + status = e.code + except (urllib.error.URLError, OSError) as e: + return 599, {"error": str(e)} + try: + return status, json.loads(raw) if raw else {} + except ValueError: + return status, {"raw": raw.decode(errors="replace")} + + +def _basic(login, password): + tok = base64.b64encode(f"{login}:{password}".encode()).decode() + return f"Basic {tok}" + + +# -------------------------------------------------------------------------- +# Core logic (class so tests can instantiate with a stub config) +# -------------------------------------------------------------------------- + +class LinkService: + def __init__(self, config): + self.cfg = config + # gitea_base: URL this service uses to reach the forge (loopback ok) + # public_gitea_base: URL handed to clients (tailnet / public) + self.gitea = config["gitea_base"].rstrip("/") + self.public_gitea = config.get("public_gitea_base", self.gitea).rstrip("/") + self.userinfo_url = config.get( + "zitadel_userinfo", "https://id.shre.ai/oidc/v1/userinfo") + + # -- identity ---------------------------------------------------------- + def validate_token(self, access_token): + status, info = http_json( + "GET", self.userinfo_url, + headers={"Authorization": f"Bearer {access_token}"}) + if status != 200: + return None, f"zitadel userinfo rejected token (HTTP {status})" + if not isinstance(info, dict) or "sub" not in info: + return None, "zitadel userinfo returned no sub" + return info, None + + @staticmethod + def derive_login(userinfo): + cand = userinfo.get("preferred_username") or "" + if not cand: + email = userinfo.get("email") or "" + cand = email.split("@", 1)[0] + cand = cand.split("@", 1)[0] # strip org domain from zitadel logins + cand = LOGIN_SAFE.sub("-", cand).strip("-._").lower() + return cand or None + + # -- gitea admin ------------------------------------------------------- + def _admin_hdr(self): + return {"Authorization": f"token {self.cfg['admin_token']}"} + + def user_exists(self, login): + status, _ = http_json( + "GET", f"{self.gitea}/api/v1/users/{login}", headers=self._admin_hdr()) + return status == 200 + + def create_user(self, login, userinfo): + pw_alphabet = string.ascii_letters + string.digits + password = "".join(secrets.choice(pw_alphabet) for _ in range(30)) + email = userinfo.get("email") or f"{login}@users.noreply.granthi.shre.ai" + body = { + "username": login, + "email": email, + "password": password, + "must_change_password": False, + "visibility": "private", + "full_name": userinfo.get("name", ""), + # source_id intentionally omitted -> local user; see module doc. + } + status, resp = http_json( + "POST", f"{self.gitea}/api/v1/admin/users", + headers=self._admin_hdr(), body=body) + if status != 201: + return f"gitea admin user create failed (HTTP {status}): {resp}" + return None + + def mint_token(self, login, device_name): + """Admin basic auth + Sudo header. Verified on Gitea 1.27.1: + token-auth sudo (header or ?sudo=) -> 401; basic+Sudo -> 201.""" + safe_dev = LOGIN_SAFE.sub("-", device_name or "device")[:40] + token_name = "granthi-sync-{}-{}".format( + safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")) + status, resp = http_json( + "POST", f"{self.gitea}/api/v1/users/{login}/tokens", + headers={ + "Authorization": _basic( + self.cfg["admin_login"], self.cfg["admin_password"]), + "Sudo": login, + }, + body={"name": token_name, + "scopes": ["write:repository", "write:user"]}) + if status != 201: + return None, None, f"token mint failed (HTTP {status}): {resp}" + return resp.get("sha1"), token_name, None + + # -- endpoints --------------------------------------------------------- + def link(self, body): + device_name = body.get("device_name") or "device" + if self.cfg.get("test_mode") and isinstance(body.get("test_userinfo"), dict): + LOG.warning("TEST-MODE link request (stubbed userinfo)") + userinfo = body["test_userinfo"] + else: + token = body.get("zitadel_access_token") + if not token: + return 400, {"error": "zitadel_access_token required"} + userinfo, err = self.validate_token(token) + if err: + return 401, {"error": err} + login = self.derive_login(userinfo) + if not login: + return 422, {"error": "could not derive a login from userinfo"} + if not self.user_exists(login): + err = self.create_user(login, userinfo) + if err: + return 502, {"error": err} + LOG.info("created gitea user %s", login) + gitea_token, token_name, err = self.mint_token(login, device_name) + if err: + return 502, {"error": err} + LOG.info("minted token %s for %s", token_name, login) + return 200, {"gitea_base": self.public_gitea, "login": login, + "token": gitea_token, "token_name": token_name} + + def repos(self, body): + token = body.get("token") + name = body.get("name") + if not token or not name: + return 400, {"error": "token and name required"} + status, resp = http_json( + "POST", f"{self.gitea}/api/v1/user/repos", + headers={"Authorization": f"token {token}"}, + body={"name": name, "private": bool(body.get("private", True)), + "default_branch": "main", "auto_init": False}) + if status == 409: + return 409, {"error": f"repo {name} already exists"} + if status != 201: + return 502, {"error": f"repo create failed (HTTP {status}): {resp}"} + full_name = resp.get("full_name", "") + return 200, { + "name": resp.get("name"), "full_name": full_name, + "private": resp.get("private"), + "clone_url": f"{self.public_gitea}/{full_name}.git", + "html_url": f"{self.public_gitea}/{full_name}", + } + + +# -------------------------------------------------------------------------- +# HTTP server plumbing +# -------------------------------------------------------------------------- + +class Handler(BaseHTTPRequestHandler): + service = None # set by serve() + server_version = f"granthi-link/{VERSION}" + + def _send(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): + if self.path == "/health": + self._send(200, {"status": "ok", "service": "granthi-link", + "version": VERSION}) + else: + self._send(404, {"error": "not found"}) + + def do_POST(self): + try: + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length) or b"{}") + except (ValueError, TypeError): + return self._send(400, {"error": "invalid JSON body"}) + if self.path == "/v1/link": + status, resp = self.service.link(body) + elif self.path == "/v1/repos": + status, resp = self.service.repos(body) + else: + return self._send(404, {"error": "not found"}) + self._send(status, resp) + + def log_message(self, fmt, *args): # route to logging, redact nothing else + LOG.info("%s %s", self.address_string(), fmt % args) + + +def serve(config): + service = LinkService(config) + Handler.service = service + servers = [] + for host, port in config.get("binds", [["127.0.0.1", 3042]]): + srv = ThreadingHTTPServer((host, int(port)), Handler) + servers.append(srv) + t = threading.Thread(target=srv.serve_forever, daemon=True) + t.start() + LOG.info("listening on %s:%s", host, port) + + stop = threading.Event() + + def on_term(signum, frame): + LOG.info("signal %s, shutting down", signum) + stop.set() + + signal.signal(signal.SIGTERM, on_term) + signal.signal(signal.SIGINT, on_term) + stop.wait() + for srv in servers: + srv.shutdown() + + +def main(): + logging.basicConfig(level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s") + path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CONFIG + with open(path) as f: + config = json.load(f) + for key in ("gitea_base", "admin_token", "admin_login", "admin_password"): + if key not in config: + sys.exit(f"config missing required key: {key}") + st = os.stat(path) + if st.st_mode & 0o077: + LOG.warning("config %s is group/world readable -- chmod 600 it", path) + serve(config) + + +if __name__ == "__main__": + main() diff --git a/tests/__pycache__/test_client.cpython-314.pyc b/tests/__pycache__/test_client.cpython-314.pyc new file mode 100644 index 0000000..bfe7bef Binary files /dev/null and b/tests/__pycache__/test_client.cpython-314.pyc differ diff --git a/tests/__pycache__/test_server.cpython-314.pyc b/tests/__pycache__/test_server.cpython-314.pyc new file mode 100644 index 0000000..6afe809 Binary files /dev/null and b/tests/__pycache__/test_server.cpython-314.pyc differ diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..1c78aae --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,197 @@ +"""Unit tests for the granthi-sync client: autocommit / ff / diverged logic, +config handling, device-flow polling (mocked HTTP). Stdlib unittest only.""" + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "client")) + +# Point client config at a temp home BEFORE import side effects. +_TMP_HOME = tempfile.mkdtemp(prefix="granthi-test-home-") +os.environ["GRANTHI_SYNC_HOME"] = _TMP_HOME + +import granthi_sync_client as client # noqa: E402 + +GIT_ENV = { + "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t", + "HOME": _TMP_HOME, "PATH": os.environ["PATH"], +} + + +def run_git(cwd, *args): + return subprocess.run(["git", "-C", cwd] + list(args), check=True, + capture_output=True, text=True, env=GIT_ENV).stdout.strip() + + +class GitScenarioBase(unittest.TestCase): + """bare 'cloud' repo + two working clones to simulate device vs remote.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="granthi-test-") + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.bare = os.path.join(self.tmp, "cloud.git") + subprocess.run(["git", "init", "--bare", "-b", "main", self.bare], + check=True, capture_output=True, env=GIT_ENV) + self.local = os.path.join(self.tmp, "local") + os.makedirs(self.local) + client.ensure_repo(self.local) + run_git(self.local, "remote", "add", "granthi", self.bare) + # git() in the client inherits our env via subprocess default; set + # identity locally in the repo so commits work. + run_git(self.local, "config", "user.name", "t") + run_git(self.local, "config", "user.email", "t@t") + + def write(self, repo, name, content): + with open(os.path.join(repo, name), "w") as f: + f.write(content) + + def other_clone(self): + other = os.path.join(self.tmp, "other") + subprocess.run(["git", "clone", self.bare, other], check=True, + capture_output=True, env=GIT_ENV) + run_git(other, "config", "user.name", "o") + run_git(other, "config", "user.email", "o@o") + return other + + +class TestAutocommit(GitScenarioBase): + def test_autocommit_commits_changes(self): + self.write(self.local, "a.txt", "one") + self.assertTrue(client.autocommit(self.local)) + msg = run_git(self.local, "log", "-1", "--format=%s") + self.assertTrue(msg.startswith("sync: "), msg) + + def test_autocommit_noop_when_clean(self): + self.write(self.local, "a.txt", "one") + client.autocommit(self.local) + self.assertFalse(client.autocommit(self.local)) + + +class TestSyncFolder(GitScenarioBase): + def test_initial_push(self): + self.write(self.local, "a.txt", "one") + outcome, _ = client.sync_folder(self.local) + self.assertEqual(outcome, "pushed") + self.assertIn("a.txt", run_git(self.local, "ls-tree", "--name-only", + "granthi/main")) + + def test_ff_pull_when_remote_ahead(self): + self.write(self.local, "a.txt", "one") + client.sync_folder(self.local) + other = self.other_clone() + self.write(other, "b.txt", "from-other") + run_git(other, "add", "-A") + run_git(other, "commit", "-m", "remote change") + run_git(other, "push", "origin", "main") + outcome, detail = client.sync_folder(self.local) + self.assertEqual((outcome, detail), ("synced", "ff-pulled")) + self.assertTrue(os.path.exists(os.path.join(self.local, "b.txt"))) + + def test_diverged_is_skipped_never_forced(self): + self.write(self.local, "a.txt", "one") + client.sync_folder(self.local) + other = self.other_clone() + self.write(other, "b.txt", "remote side") + run_git(other, "add", "-A") + run_git(other, "commit", "-m", "remote change") + run_git(other, "push", "origin", "main") + remote_sha = run_git(other, "rev-parse", "HEAD") + self.write(self.local, "a.txt", "local side") # divergence + outcome, _ = client.sync_folder(self.local) + self.assertEqual(outcome, "diverged") + # remote must be untouched (not forced, not merged) + bare_sha = run_git(self.bare, "rev-parse", "main") + self.assertEqual(bare_sha, remote_sha) + + def test_clean_when_in_sync(self): + self.write(self.local, "a.txt", "one") + client.sync_folder(self.local) + outcome, _ = client.sync_folder(self.local) + self.assertEqual(outcome, "clean") + + +class TestConfig(unittest.TestCase): + def test_save_creates_0600(self): + client.save_config({"login": "x", "folders": {}}) + st = os.stat(client.CONFIG_PATH) + self.assertEqual(st.st_mode & 0o777, 0o600) + self.assertEqual(client.load_config()["login"], "x") + + def test_load_missing_returns_empty(self): + with mock.patch.object(client, "CONFIG_PATH", "/nonexistent/nope.json"): + self.assertEqual(client.load_config(), {}) + + +class TestDeviceFlow(unittest.TestCase): + def test_device_flow_polls_until_token(self): + calls = [] + + def fake_http(method, url, headers=None, body=None, form=None, timeout=30): + calls.append(url) + if url.endswith("/device_authorization"): + return 200, {"device_code": "dc", "user_code": "AB-CD", + "verification_uri": "https://id/device", + "verification_uri_complete": "https://id/device?u=AB-CD", + "interval": 0, "expires_in": 300} + if len([c for c in calls if c.endswith("/token")]) < 3: + return 400, {"error": "authorization_pending"} + return 200, {"access_token": "ZTOK"} + + with mock.patch.object(client, "http_json", fake_http), \ + mock.patch.object(client.time, "sleep"): + tok = client.device_flow() + self.assertEqual(tok, "ZTOK") + self.assertEqual(len([c for c in calls if c.endswith("/token")]), 3) + + def test_device_flow_slow_down_backs_off(self): + state = {"n": 0} + + def fake_http(method, url, headers=None, body=None, form=None, timeout=30): + if url.endswith("/device_authorization"): + return 200, {"device_code": "dc", "user_code": "AB", + "verification_uri": "u", "interval": 1, + "expires_in": 300} + state["n"] += 1 + if state["n"] == 1: + return 400, {"error": "slow_down"} + return 200, {"access_token": "T"} + + sleeps = [] + with mock.patch.object(client, "http_json", fake_http), \ + mock.patch.object(client.time, "sleep", sleeps.append): + self.assertEqual(client.device_flow(), "T") + self.assertIn(6, sleeps) # 1 + 5 backoff after slow_down + + +class TestCredentialHelper(GitScenarioBase): + def test_helper_emits_creds_for_matching_host(self): + client.save_config({"gitea_base": "http://100.111.127.127:3041", + "login": "alice", "token": "sekrit", "folders": {}}) + stdin = "protocol=http\nhost=100.111.127.127:3041\n\n" + out = subprocess.run( + [sys.executable, client.__file__, "git-credential", "get"], + input=stdin, capture_output=True, text=True, + env={**GIT_ENV, "GRANTHI_SYNC_HOME": _TMP_HOME}) + self.assertIn("username=alice", out.stdout) + self.assertIn("password=sekrit", out.stdout) + + def test_helper_silent_for_other_host(self): + client.save_config({"gitea_base": "http://100.111.127.127:3041", + "login": "alice", "token": "sekrit", "folders": {}}) + out = subprocess.run( + [sys.executable, client.__file__, "git-credential", "get"], + input="protocol=https\nhost=github.com\n\n", + capture_output=True, text=True, + env={**GIT_ENV, "GRANTHI_SYNC_HOME": _TMP_HOME}) + self.assertNotIn("password=", out.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..3d02b5a --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,186 @@ +"""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()