Gabriel Vidal
3a0ef08dcf
- liveness falls back to psutil where there is no /proc (macOS) - GET /feed + /feed/file: the hub pulls transcripts it has no mount for - POST /pair trades a one-time code for the worker's bearer; /unpair drops it - /health reports worker identity + permission mode - worker/install-macos.sh: venv, launchd agent bound to the Tailscale IP, prints the pairing string (addr/code/claude login) - Dockerfile copies every sidecar module (claude_cli was missing, so the in-container runner could not import) Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
"""Pairing: one-time code → bearer, reuse refused, bad codes burn it."""
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
import pairing
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("WORKER_STATE_DIR", str(tmp_path / "state"))
|
|
|
|
def _auth(h):
|
|
if not pairing.token() or h != f"Bearer {pairing.token()}":
|
|
raise HTTPException(401, "unauthorized")
|
|
|
|
app = FastAPI()
|
|
app.include_router(pairing.make_router(
|
|
_auth, lambda: {"email": "g@orus.example", "loggedIn": True},
|
|
lambda: {"cwd": "/Users/g/projects/orus-monorepo"}))
|
|
return TestClient(app)
|
|
|
|
|
|
def test_pair_once_then_409(client):
|
|
code = pairing.new_code()
|
|
r = client.post("/pair", json={"code": code.lower(), "hubName": "homelab"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["token"] == pairing.token() and len(body["token"]) > 30
|
|
assert body["account"]["email"] == "g@orus.example"
|
|
assert body["cwd"].endswith("orus-monorepo")
|
|
assert body["workerId"] and body["hubName"] == "homelab"
|
|
again = client.post("/pair", json={"code": code})
|
|
assert again.status_code == 409
|
|
assert again.json()["detail"]["kind"] == "no_code"
|
|
|
|
|
|
def test_wrong_code_is_403_and_burns_after_max_attempts(client):
|
|
pairing.new_code()
|
|
for _ in range(pairing.MAX_ATTEMPTS):
|
|
r = client.post("/pair", json={"code": "WRONGWRONG"})
|
|
assert r.status_code == 403
|
|
assert not pairing._read("pairing-code")
|
|
|
|
|
|
def test_repair_rotates_the_token_and_keeps_the_id(client):
|
|
first = client.post("/pair", json={"code": pairing.new_code()}).json()
|
|
second = client.post("/pair", json={"code": pairing.new_code()}).json()
|
|
assert first["token"] != second["token"]
|
|
assert first["workerId"] == second["workerId"]
|
|
|
|
|
|
def test_unpair_drops_the_bearer(client):
|
|
tok = client.post("/pair", json={"code": pairing.new_code()}).json()["token"]
|
|
assert client.post("/unpair").status_code == 401
|
|
r = client.post("/unpair", headers={"Authorization": f"Bearer {tok}"})
|
|
assert r.status_code == 200 and pairing.token() == ""
|
|
|
|
|
|
def test_state_files_are_private(client):
|
|
client.post("/pair", json={"code": pairing.new_code()})
|
|
mode = (pairing.state_dir() / "token").stat().st_mode & 0o777
|
|
assert mode == 0o600
|