Gabriel Vidal
deb01427d5
newproject/: the policy (guestDefaults.newProjects — quota, templates, the ones the owner approves first, visibility, level, domain — and a per-guest createProjects + address bucket), the New project form built from it, and the flow: POST /api/new-project/form (the run's guesses as defaults) → submit → shape/policy and host checks (a problem sends a corrected form back) → an owner ask for a backend template → sidecar /scaffold → grant → /relocate + resume the same session inside the project with an app-authored turn. SpawnBody.newProject starts a guest's run in ~/projects/.new-project/<uid> (staging_scope: /tmp only); /api/me reports newProjects; the thread follows ConversationDetail.continuedIn. Fix: a guest's turn into an existing conversation resumed with no cwd, so the sidecar ran it in the lab checkout (its CLAUDE.md loaded) — it now resumes where the session lives. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
287 lines
14 KiB
Python
287 lines
14 KiB
Python
"""New projects (newproject/): the policy, the form, the submit → check →
|
|
approve → scaffold → grant → move-and-resume flow, and the New project spawn.
|
|
|
|
Imports ``main`` against throwaway paths (no startup event → no threads); the
|
|
host sidecar and the run launcher are stubbed."""
|
|
|
|
import os
|
|
import pathlib
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
TMP = pathlib.Path(tempfile.mkdtemp())
|
|
for k, v in {"DB_PATH": TMP / "x.db", "TRANSCRIPTS_DIR": TMP / "t",
|
|
"WORKSPACE": TMP, "REPO_DIR": TMP, "ARCHIVE_DIR": TMP / "arch",
|
|
"META_PATH": TMP / "meta.json", "WORKERS_PATH": TMP / "w.json",
|
|
"WORKERS_TRANSCRIPTS_DIR": TMP / "mirror",
|
|
"CRON_PATH": TMP / "cron.json", "NOTIFY_PATH": TMP / "n.json",
|
|
"FORMS_PATH": TMP / "f.json", "UI_STATE_PATH": TMP / "ui.json",
|
|
"NOTIF_READ_PATH": TMP / "nr.json",
|
|
"MAIL_TRIGGER_SENDERS_PATH": TMP / "mt.json",
|
|
"USERS_PATH": TMP / "users.json", "AUTH_PATH": TMP / "auth.json",
|
|
"NEW_PROJECTS_PATH": TMP / "np.json",
|
|
"PROJECTS_DIR": TMP / "projects", "TEMPLATES_DIR": TMP / "projects" / "templates",
|
|
"INTERNAL_API_TOKEN": "tok", "GUEST_HOSTS": "claude.example.test",
|
|
"SIDECAR_TOKEN": "", "PI_TRANSCRIPTS_DIR": ""}.items():
|
|
os.environ[k] = str(v)
|
|
for t, marker in (("react-vite", "package.json"), ("vite-react-pocketbase", "package.json"),
|
|
("datakit", "pyproject.toml"), ("homelab-service", "docker-compose.yml"),
|
|
("skill-cli", "SKILL.md")):
|
|
(TMP / "projects" / "templates" / t).mkdir(parents=True)
|
|
(TMP / "projects" / "templates" / t / marker).write_text("{}")
|
|
(TMP / "projects" / "film").mkdir(parents=True)
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
import main # noqa: E402
|
|
import runs.routes as run_routes # noqa: E402
|
|
import runs.service as run_service # noqa: E402
|
|
from newproject import form as np_form # noqa: E402
|
|
from newproject import policy, service # noqa: E402
|
|
from users.caller import OWNER, SESSION_COOKIE, guest_from_record # noqa: E402
|
|
|
|
DEPS = main.app.state.ai
|
|
OWNER_H = {"X-Internal-Auth": "tok"}
|
|
SID = "11111111-2222-4333-8444-555555555555"
|
|
|
|
|
|
class FakeHost:
|
|
"""The host sidecar: /scaffold/check, /scaffold, /relocate."""
|
|
|
|
def __init__(self, problems=None):
|
|
self.calls = []
|
|
self.problems = problems or []
|
|
|
|
def __call__(self, deps, path, payload, timeout):
|
|
self.calls.append((path, payload))
|
|
if path == "/scaffold/check":
|
|
return {"ok": not self.problems,
|
|
"result": {"ok": not self.problems, "problems": self.problems}}
|
|
if path == "/scaffold":
|
|
v = payload["vars"]
|
|
host = f"{v['LABEL']}.{v['SUBDOMAIN']}.gabvdl.xyz"
|
|
res = {"dir": f"/home/x/projects/{v['NAME']}", "host": host,
|
|
"url": f"https://{host}", "repo": f"https://git/{v.get('GITEA_OWNER', 'g')}/{v['NAME']}"}
|
|
if payload.get("password"):
|
|
res["password"] = "abcd-efgh"
|
|
return {"ok": True, "result": res,
|
|
"log": "noise\nSCAFFOLD SUMMARY\nfiles…\nSCAFFOLD_RESULT={}"}
|
|
if path == "/relocate":
|
|
return {"copied": True}
|
|
raise AssertionError(path)
|
|
|
|
|
|
def _wait(fid, *states):
|
|
for _ in range(200):
|
|
r = DEPS.new_projects.get(fid)
|
|
if r and r["status"] in states:
|
|
return r
|
|
time.sleep(0.02)
|
|
raise AssertionError(f"{fid} stuck at {DEPS.new_projects.get(fid)}")
|
|
|
|
|
|
class NewProjects(unittest.TestCase):
|
|
def setUp(self):
|
|
for u in DEPS.users.all():
|
|
DEPS.users.delete(u["id"])
|
|
DEPS.new_projects._data["requests"].clear()
|
|
DEPS.users.set_defaults({"newProjects": {
|
|
"quota": 2, "templates": ["react-vite", "vite-react-pocketbase"],
|
|
"approve": ["vite-react-pocketbase"], "visibility": ["public", "password", "geo"],
|
|
"visibilityDefault": "public", "level": "ship"}})
|
|
self.dom = DEPS.users.create("dom@example.com", "Dom", projects={"film": "chat"},
|
|
createProjects=True)
|
|
DEPS.meta_store.update(SID, {"owner": self.dom["id"], "projects": [],
|
|
"newProject": True})
|
|
self.host = FakeHost()
|
|
self._patch(service, "_sidecar", self.host)
|
|
self.sent = []
|
|
self._patch(run_service, "_send_message",
|
|
lambda deps, prompt, **kw: self.sent.append((prompt, kw)) or
|
|
{"sessionId": kw.get("session_id") or "new"})
|
|
self.c = TestClient(main.app)
|
|
|
|
def tearDown(self):
|
|
import threading
|
|
for t in threading.enumerate():
|
|
if t.name.startswith("new-project-"):
|
|
t.join(5)
|
|
|
|
def _patch(self, mod, name, value):
|
|
orig = getattr(mod, name)
|
|
setattr(mod, name, value)
|
|
self.addCleanup(setattr, mod, name, orig)
|
|
|
|
def caller(self):
|
|
return guest_from_record(DEPS.users.get(self.dom["id"]))
|
|
|
|
def open(self, **guesses):
|
|
r = self.c.post("/api/new-project/form", headers=OWNER_H,
|
|
json={"sessionId": SID, **guesses})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
return r.json()
|
|
|
|
def submit(self, fid, **answers):
|
|
g = TestClient(main.app, base_url="https://claude.example.test")
|
|
g.cookies.set(SESSION_COOKIE, DEPS.auth.open_session(self.dom["id"]))
|
|
base = {"name": "planning", "title": "Planning", "description": "",
|
|
"template": "react-vite", "label": "planning", "visibility": "public"}
|
|
r = g.post(f"/api/forms/{fid}/submit", json={"answers": {**base, **answers}},
|
|
headers={"Origin": "https://claude.example.test"})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
|
|
# ── policy ───────────────────────────────────────────────────────────────
|
|
def test_access(self):
|
|
self.assertTrue(policy.access(DEPS, OWNER)["allowed"])
|
|
a = policy.access(DEPS, self.caller())
|
|
self.assertEqual((a["allowed"], a["remaining"], a["bucket"]), (True, 2, "dom"))
|
|
DEPS.users.update(self.dom["id"], {"createProjects": False})
|
|
self.assertEqual(policy.access(DEPS, self.caller())["reason"], "not enabled")
|
|
DEPS.users.update(self.dom["id"], {"createProjects": True, "bucket": "dominique"})
|
|
self.assertEqual(policy.access(DEPS, self.caller())["bucket"], "dominique")
|
|
with self.assertRaises(ValueError):
|
|
DEPS.users.update(self.dom["id"], {"bucket": "not a label"})
|
|
|
|
def test_app_templates_skip_other_generators(self):
|
|
self.assertEqual(policy.app_templates(),
|
|
["datakit", "react-vite", "vite-react-pocketbase"])
|
|
self.assertEqual(policy.options(DEPS, "owner")["templates"],
|
|
["datakit", "react-vite", "vite-react-pocketbase"])
|
|
|
|
def test_guest_form_is_pinned_to_their_bucket(self):
|
|
opts = policy.options(DEPS, self.dom["id"])
|
|
f = {x.get("key"): x for x in np_form.fields(opts, {"name": "My Planning!",
|
|
"template": "nope"})}
|
|
self.assertEqual(f["name"]["default"], "my-planning")
|
|
self.assertEqual(f["template"]["default"], "react-vite") # fell back
|
|
self.assertNotIn("bucket", f)
|
|
self.assertIn("my-planning.dom.gabvdl.xyz", f["label"]["help"])
|
|
self.assertIn("approves it first", f["template"]["options"][1]["label"])
|
|
owner = {x.get("key"): x for x in np_form.fields(policy.options(DEPS, "owner"), {})}
|
|
self.assertEqual(owner["bucket"]["default"], "dev")
|
|
|
|
# ── the flow ─────────────────────────────────────────────────────────────
|
|
def test_frontend_project_is_created_granted_and_moved(self):
|
|
out = self.open(name="planning", title="Planning", summary="A shoot planner")
|
|
self.assertEqual(out["requester"], self.dom["id"])
|
|
self.submit(out["formId"])
|
|
req = _wait(out["formId"], "created", "failed")
|
|
self.assertEqual(req["status"], "created", req.get("error"))
|
|
self.assertEqual(DEPS.users.get(self.dom["id"])["projects"]["planning"], "ship")
|
|
scaffold = dict(self.host.calls)["/scaffold"]
|
|
self.assertEqual(scaffold["vars"]["SUBDOMAIN"], "dom")
|
|
self.assertEqual(scaffold["vars"]["GITEA_OWNER"], "dom")
|
|
self.assertFalse(scaffold["password"])
|
|
self.assertEqual(dict(self.host.calls)["/relocate"]["cwd"], "~/projects/planning")
|
|
prompt, kw = self.sent[-1]
|
|
self.assertEqual(kw["session_id"], SID)
|
|
self.assertEqual(kw["cwd"], "~/projects/planning")
|
|
self.assertIn("~/projects/planning", kw["scope"]["writeDirs"])
|
|
self.assertTrue(prompt.startswith(service.APP_TAG))
|
|
self.assertIn("https://planning.dom.gabvdl.xyz", prompt)
|
|
self.assertIn("SCAFFOLD SUMMARY", prompt)
|
|
self.assertEqual(DEPS.meta_store.peek(SID)["projects"], ["planning"])
|
|
self.assertEqual(DEPS.new_projects.used(self.dom["id"]), 1)
|
|
|
|
def test_password_site_hands_the_password_over(self):
|
|
out = self.open(name="secret")
|
|
self.submit(out["formId"], name="secret", label="secret", visibility="password")
|
|
_wait(out["formId"], "created")
|
|
self.assertTrue(dict(self.host.calls)["/scaffold"]["password"])
|
|
self.assertIn("abcd-efgh", self.sent[-1][0])
|
|
|
|
def test_host_problem_sends_the_form_back(self):
|
|
self.host.problems = [{"field": "label",
|
|
"message": "https://planning.dom.gabvdl.xyz is already in use"}]
|
|
out = self.open(name="planning")
|
|
self.submit(out["formId"])
|
|
req = _wait(out["formId"], "retried")
|
|
nxt = DEPS.form_store.get(req["next"])
|
|
self.assertEqual(nxt["sessionId"], SID)
|
|
self.assertIn("already in use", nxt["fields"][0]["text"])
|
|
self.assertEqual(DEPS.new_projects.get(req["next"])["status"], "pending")
|
|
self.assertNotIn("/scaffold", dict(self.host.calls))
|
|
|
|
def test_bad_answers_never_reach_the_host(self):
|
|
out = self.open()
|
|
self.submit(out["formId"], name="Bad Name", label="ok")
|
|
req = _wait(out["formId"], "retried")
|
|
self.assertEqual(self.host.calls, [])
|
|
self.assertIn("can't be a project name", req["error"])
|
|
|
|
def test_backend_template_waits_for_the_owner(self):
|
|
out = self.open()
|
|
self.submit(out["formId"], template="vite-react-pocketbase")
|
|
req = _wait(out["formId"], "approval")
|
|
self.assertNotIn("/scaffold", dict(self.host.calls))
|
|
r = self.c.post(f"/api/ask/{req['askId']}/answer", headers=OWNER_H,
|
|
json={"label": service.APPROVE})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertEqual(_wait(out["formId"], "created")["status"], "created")
|
|
|
|
def test_declined_project_is_told(self):
|
|
out = self.open()
|
|
self.submit(out["formId"], template="vite-react-pocketbase")
|
|
req = _wait(out["formId"], "approval")
|
|
self.c.post(f"/api/ask/{req['askId']}/answer", headers=OWNER_H,
|
|
json={"label": service.DECLINE})
|
|
self.assertEqual(_wait(out["formId"], "declined")["status"], "declined")
|
|
self.assertIn("declined", self.sent[-1][0])
|
|
self.assertEqual(DEPS.new_projects.used(self.dom["id"]), 0)
|
|
|
|
def test_quota_and_permission_gate_the_form(self):
|
|
DEPS.users.update(self.dom["id"], {"createProjects": False})
|
|
r = self.c.post("/api/new-project/form", headers=OWNER_H, json={"sessionId": SID})
|
|
self.assertEqual(r.status_code, 403)
|
|
DEPS.users.update(self.dom["id"], {"createProjects": True})
|
|
for n in ("a1", "a2"):
|
|
DEPS.new_projects.create(n, SID, self.dom["id"], {})
|
|
DEPS.new_projects.update(n, {"status": "created"})
|
|
r = self.c.post("/api/new-project/form", headers=OWNER_H, json={"sessionId": SID})
|
|
self.assertEqual(r.status_code, 403)
|
|
self.assertIn("the most you can", r.json()["detail"]["message"])
|
|
|
|
def test_guest_cannot_open_forms_directly(self):
|
|
g = TestClient(main.app, base_url="https://claude.example.test")
|
|
g.cookies.set(SESSION_COOKIE, DEPS.auth.open_session(self.dom["id"]))
|
|
self.assertEqual(g.post("/api/new-project/form", json={"sessionId": SID}).status_code,
|
|
403)
|
|
|
|
# ── the spawn ────────────────────────────────────────────────────────────
|
|
def test_new_project_spawn_needs_no_project(self):
|
|
seen = []
|
|
self._patch(run_routes, "_send_message",
|
|
lambda deps, prompt, **kw: seen.append((prompt, kw)) or {"sessionId": "s"})
|
|
self._patch(run_routes.deliver, "owner_ping_start", lambda *a, **k: None)
|
|
g = TestClient(main.app, base_url="https://claude.example.test")
|
|
g.cookies.set(SESSION_COOKIE, DEPS.auth.open_session(self.dom["id"]))
|
|
r = g.post("/api/spawn", json={"prompt": "a site to plan our shoots",
|
|
"newProject": True, "projects": []},
|
|
headers={"Origin": "https://claude.example.test"})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
prompt, kw = seen[-1]
|
|
self.assertEqual(kw["cwd"], "~/projects/.new-project/dom")
|
|
self.assertTrue((TMP / "projects" / ".new-project" / "dom").is_dir())
|
|
self.assertEqual(kw["scope"]["writeDirs"], ["/tmp"])
|
|
self.assertIn("shared:new-frontend-project", kw["scope"]["systemPrompt"])
|
|
self.assertIn("New project", prompt)
|
|
self.assertEqual(kw["stamp"]["owner"], "dom")
|
|
DEPS.users.update(self.dom["id"], {"createProjects": False})
|
|
r = g.post("/api/spawn", json={"prompt": "x", "newProject": True},
|
|
headers={"Origin": "https://claude.example.test"})
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_me_reports_access(self):
|
|
g = TestClient(main.app, base_url="https://claude.example.test")
|
|
g.cookies.set(SESSION_COOKIE, DEPS.auth.open_session(self.dom["id"]))
|
|
np = g.get("/api/me").json()["newProjects"]
|
|
self.assertEqual((np["allowed"], np["remaining"], np["bucket"]), (True, 2, "dom"))
|
|
users = self.c.get("/api/users", headers=OWNER_H).json()["users"]
|
|
self.assertEqual(users[0]["createProjects"], True)
|
|
self.assertEqual(users[0]["createdProjects"], 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|