Gabriel Vidal
3d741b5b1e
Notifications follow the conversation's audience (users/access.audience):
whoever started it and the people it's shared with, plus the owner once they
follow a guest's conversation (writing in it, or PUT /api/conversation-follow),
minus whoever muted it (meta.follow). Pushes and asks fan out to each of them
on their own channels — the owner on the webhooks, a guest in-app + email —
and the feeds (/api/notifications, /api/notify-log, the phone's /unread)
list only what the caller hears, links rebased to the reader's host. Forms
carry forYou so "waiting for feedback" is per user.
/api/users carries spentUsd (this month), totalSpentUsd and conversations per
guest (users/runs.spend_by_user, the same sum the budget is checked against).
POST /api/users/{id}/impersonate opens a 4-hour guest session for the owner's
browser on the admin host (imp on the record: never extended, not counted in
the guest's sessions); /api/me says impersonating; logout ends it. A spawn
while impersonating skips the "guest started" ping.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
538 lines
29 KiB
Python
538 lines
29 KiB
Python
"""Guests (users/): the login lifecycle, the gate's caller resolution and
|
|
default-deny route table, the per-conversation filters, the server-pinned
|
|
spawn, sharing, per-user ui-state, the SSE filter and the author tag.
|
|
|
|
Imports ``main`` against throwaway paths (no startup event → no threads)."""
|
|
|
|
import os
|
|
import pathlib
|
|
import tempfile
|
|
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",
|
|
"PROJECTS_DIR": TMP / "projects",
|
|
"INTERNAL_API_TOKEN": "tok", "GUEST_HOSTS": "claude.example.test",
|
|
"SIDECAR_TOKEN": "", "PI_TRANSCRIPTS_DIR": ""}.items():
|
|
os.environ[k] = str(v)
|
|
(TMP / "projects" / "film").mkdir(parents=True)
|
|
(TMP / "projects" / "other").mkdir(parents=True)
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
import main # noqa: E402
|
|
import conversations.routes as conv_routes # noqa: E402
|
|
import runs.routes as run_routes # noqa: E402
|
|
from conversations.parse import _split_author # noqa: E402
|
|
from core.routes import _guest_filter # noqa: E402
|
|
from notifications.service import _flat_notifications # noqa: E402
|
|
from users import access # noqa: E402
|
|
from users.caller import OWNER, SESSION_COOKIE, guest_from_record # noqa: E402
|
|
from users.runs import author_tag, check_model # noqa: E402
|
|
from users.store import DEFAULT_GUEST_SHORTCUTS, DEFAULT_SHARED_SKILLS # noqa: E402
|
|
|
|
DEPS = main.app.state.ai
|
|
OWNER_H = {"X-Internal-Auth": "tok"}
|
|
|
|
|
|
def _card(sid, **meta):
|
|
return {"sessionId": sid, "endedAt": "2026-10-10T00:00:00Z", "cost": 1.0,
|
|
"meta": {"archived": False, **meta}}
|
|
|
|
|
|
class Guests(unittest.TestCase):
|
|
def setUp(self):
|
|
for u in DEPS.users.all():
|
|
DEPS.users.delete(u["id"])
|
|
self.dom = DEPS.users.create("Dom@Example.com", "Dom",
|
|
projects={"film": "chat"},
|
|
models=["sonnet", "haiku"])
|
|
self.claire = DEPS.users.create("claire@example.com", "Claire",
|
|
projects={"film": "view"})
|
|
self.c = TestClient(main.app)
|
|
self.dom_cookie = DEPS.auth.open_session(self.dom["id"])
|
|
DEPS.users.set_defaults({"promptShortcuts": list(DEFAULT_GUEST_SHORTCUTS),
|
|
"sharedSkills": list(DEFAULT_SHARED_SKILLS)})
|
|
DEPS.ui_state.delete(f"u.{self.dom['id']}.ai-agent-settings")
|
|
|
|
def guest(self, cookie=None, host="claude.example.test"):
|
|
c = TestClient(main.app, base_url=f"https://{host}")
|
|
c.cookies.set(SESSION_COOKIE, cookie or self.dom_cookie)
|
|
return c
|
|
|
|
# ── store + login ────────────────────────────────────────────────────────
|
|
def test_login_link_is_single_use_and_opens_a_session(self):
|
|
token = DEPS.auth.mint_link(self.dom["id"])
|
|
c = TestClient(main.app, base_url="https://claude.example.test")
|
|
r = c.post("/api/auth/login", json={"token": token})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertEqual(r.json()["id"], self.dom["id"])
|
|
self.assertIn(SESSION_COOKIE, r.cookies)
|
|
self.assertEqual(c.get("/api/me").json()["role"], "guest")
|
|
again = TestClient(main.app, base_url="https://claude.example.test")
|
|
self.assertEqual(again.post("/api/auth/login",
|
|
json={"token": token}).status_code, 401)
|
|
|
|
def test_owner_links_are_invitations(self):
|
|
# Copied / mailed by the owner: lives for days, still single-use; the
|
|
# login page's own link stays a quarter-hour one.
|
|
from users.store import INVITE_TTL_S, LINK_TTL_S
|
|
r = self.c.post(f"/api/users/{self.dom['id']}/login-link", json={"send": False},
|
|
headers=OWNER_H)
|
|
self.assertEqual(r.json()["expiresInS"], INVITE_TTL_S)
|
|
self.assertGreater(INVITE_TTL_S, LINK_TTL_S)
|
|
token = r.json()["url"].split("token=")[1]
|
|
self.assertEqual(DEPS.auth.consume_link(token), self.dom["id"])
|
|
self.assertIsNone(DEPS.auth.consume_link(token))
|
|
|
|
def test_signed_in_guest_can_still_use_sign_in_links(self):
|
|
# Dom's first sign-in worked but the page hung; every later attempt
|
|
# carried the live cookie and the gate refused /api/auth/* (403).
|
|
g = self.guest()
|
|
h = {"Origin": "https://claude.example.test"}
|
|
self.assertEqual(g.post("/api/auth/request-link", json={"email": "x@y.z"},
|
|
headers=h).status_code, 200)
|
|
token = DEPS.auth.mint_link(self.dom["id"])
|
|
r = g.post("/api/auth/login", json={"token": token}, headers=h)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
# the old session gave way to the new one
|
|
self.assertIsNone(DEPS.auth.session_user(self.dom_cookie))
|
|
self.assertEqual(g.post("/api/auth/login", json={"token": token},
|
|
headers=h).status_code, 401) # spent
|
|
|
|
def test_disabled_user_loses_the_session(self):
|
|
DEPS.users.update(self.dom["id"], {"disabled": True})
|
|
r = self.guest().get("/api/me")
|
|
self.assertEqual(r.status_code, 401) # guest surface, no valid session
|
|
|
|
def test_request_link_never_reveals_who_exists(self):
|
|
c = TestClient(main.app, base_url="https://claude.example.test")
|
|
for e in ("dom@example.com", "nobody@example.com"):
|
|
self.assertEqual(c.post("/api/auth/request-link",
|
|
json={"email": e}).json(), {"ok": True})
|
|
|
|
# ── the gate ─────────────────────────────────────────────────────────────
|
|
def test_owner_paths_unchanged(self):
|
|
r = self.c.get("/api/me", headers=OWNER_H)
|
|
self.assertEqual(r.json()["role"], "owner")
|
|
self.assertEqual(self.c.get("/api/me").status_code, 403) # untrusted ip
|
|
|
|
def test_guest_surface_is_never_owner(self):
|
|
c = TestClient(main.app, base_url="https://claude.example.test")
|
|
r = c.get("/api/me", headers=OWNER_H)
|
|
self.assertEqual(r.status_code, 401)
|
|
self.assertEqual(r.json()["detail"]["kind"], "login_required")
|
|
|
|
def test_guest_default_deny(self):
|
|
g = self.guest()
|
|
for path in ("/api/users", "/api/cron", "/api/bundle", "/api/accounts",
|
|
"/api/workers", "/api/memories", "/api/services",
|
|
"/api/project-env?slug=film", "/api/notifications/unread"):
|
|
self.assertEqual(g.get(path).status_code, 403, path)
|
|
self.assertEqual(g.put("/api/file", json={}).status_code, 403)
|
|
|
|
def test_guest_cross_site_post_refused(self):
|
|
g = self.guest()
|
|
r = g.post("/api/interrupt", json={"sessionId": "x"},
|
|
headers={"Origin": "https://evil.example"})
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_me_lists_grants(self):
|
|
me = self.guest().get("/api/me").json()
|
|
self.assertEqual(me["projects"], [{"slug": "film", "level": "chat"}])
|
|
self.assertEqual(me["surface"], "guest")
|
|
|
|
# ── filtering ────────────────────────────────────────────────────────────
|
|
def test_conversation_list_is_own_plus_shared(self):
|
|
cards = [_card("a" * 8, owner=self.dom["id"]),
|
|
_card("b" * 8),
|
|
_card("c" * 8, participants=[self.dom["id"]]),
|
|
_card("d" * 8, owner=self.claire["id"])]
|
|
orig = conv_routes._conversation_cards
|
|
conv_routes._conversation_cards = lambda deps, full: cards
|
|
try:
|
|
got = {c["sessionId"] for c in
|
|
self.guest().get("/api/conversations").json()["conversations"]}
|
|
every = self.c.get("/api/conversations", headers=OWNER_H).json()
|
|
finally:
|
|
conv_routes._conversation_cards = orig
|
|
self.assertEqual(got, {"a" * 8, "c" * 8})
|
|
self.assertEqual(every["count"], 4)
|
|
|
|
def test_cards_carry_owner_and_participants(self):
|
|
# The list filter reads the card's meta: _conv_meta must pass both
|
|
# through (it once rebuilt meta from a fixed key list and dropped them).
|
|
from conversations.cards import _conv_meta
|
|
DEPS.meta_store.update("s-card", {"owner": "dom", "participants": ["claire"]})
|
|
m = _conv_meta(DEPS, {"sessionId": "s-card"})
|
|
self.assertEqual((m["owner"], m["participants"]), ("dom", ["claire"]))
|
|
|
|
def test_unseen_conversation_is_404(self):
|
|
DEPS.meta_store.update("s-other", {"owner": self.claire["id"]})
|
|
r = self.guest().get("/api/conversation?id=-x/s-other.jsonl")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_every_route_reaches_its_own_handler(self):
|
|
# A route decorator once slipped onto a helper above the real handler,
|
|
# which then shadowed /api/conversation-commits for everyone.
|
|
for r in main.app.routes:
|
|
ep = getattr(r, "endpoint", None)
|
|
if getattr(r, "path", "").startswith("/api/") and ep is not None:
|
|
self.assertFalse(ep.__name__.startswith("_"), r.path)
|
|
r = self.c.get("/api/conversation-commits?id=-x/nope.jsonl", headers=OWNER_H)
|
|
self.assertNotEqual(r.status_code, 422, r.text)
|
|
|
|
def test_projects_scoped_to_grants(self):
|
|
r = self.guest().get("/api/project?slug=other")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_ui_state_is_namespaced(self):
|
|
g = self.guest()
|
|
g.put("/api/ui-state/ai-agent-settings", json={"value": "guest"})
|
|
self.assertEqual(DEPS.ui_state.get(f"u.{self.dom['id']}.ai-agent-settings"),
|
|
"guest")
|
|
self.assertIsNone(self.c.get("/api/ui-state/ai-agent-settings",
|
|
headers=OWNER_H).json()["value"])
|
|
|
|
# ── guest settings + shared skills ──────────────────────────────────────
|
|
def test_fresh_guest_settings_start_from_the_guest_shortcuts(self):
|
|
import json
|
|
g = self.guest()
|
|
seed = json.loads(g.get("/api/ui-state/ai-agent-settings").json()["value"])
|
|
self.assertEqual([s["id"] for s in seed["state"]["promptShortcuts"]],
|
|
[s["id"] for s in DEFAULT_GUEST_SHORTCUTS])
|
|
self.assertEqual(seed["version"], 3)
|
|
# the owner's edit reaches a guest who never saved anything…
|
|
self.c.put("/api/guest-defaults", headers=OWNER_H, json={"promptShortcuts": [
|
|
{"id": "x", "label": "X", "guideline": "do x", "runner": "mac",
|
|
"shared": False, "bogus": 1}]})
|
|
seed = json.loads(g.get("/api/ui-state/ai-agent-settings").json()["value"])
|
|
self.assertEqual(seed["state"]["promptShortcuts"], [
|
|
{"id": "x", "label": "X", "guideline": "do x", "shared": True,
|
|
"defaultOn": False}])
|
|
self.assertEqual(g.get("/api/me").json()["defaultShortcuts"][0]["id"], "x")
|
|
# …and their own save wins from then on; the owner never gets a seed
|
|
g.put("/api/ui-state/ai-agent-settings", json={"value": "mine"})
|
|
self.assertEqual(g.get("/api/ui-state/ai-agent-settings").json()["value"], "mine")
|
|
self.assertIsNone(self.c.get("/api/ui-state/ai-agent-settings",
|
|
headers=OWNER_H).json()["value"])
|
|
|
|
def test_guest_defaults_are_the_owners(self):
|
|
self.assertEqual(self.guest().get("/api/guest-defaults").status_code, 403)
|
|
r = self.c.put("/api/guest-defaults", headers=OWNER_H, json={
|
|
"sharedSkills": ["ask-form", "../etc", "film/cut", "a/b/c", "ask-form"]})
|
|
self.assertEqual(r.json()["sharedSkills"], ["ask-form", "film/cut"])
|
|
self.assertEqual(DEPS.users.defaults()["promptShortcuts"][0]["id"], "commit")
|
|
|
|
def test_guest_run_scope_carries_the_shared_skills(self):
|
|
calls = self._capture()
|
|
DEPS.users.set_defaults({"sharedSkills": ["ask-form", "film/cut"]})
|
|
self.guest().post("/api/spawn", json={"prompt": "hi", "projects": ["film"]})
|
|
scope = calls[-1][1]["scope"]
|
|
self.assertEqual(scope["skills"], ["ask-form", "film/cut"])
|
|
self.assertIn("`shared:cut`", scope["systemPrompt"])
|
|
DEPS.users.set_defaults({"sharedSkills": []})
|
|
self.guest().post("/api/spawn", json={"prompt": "hi", "projects": ["film"]})
|
|
self.assertNotIn("skills", calls[-1][1]["scope"])
|
|
|
|
def test_guest_skills_are_shared_plus_granted_projects(self):
|
|
def skill(root, name):
|
|
d = root / ".claude" / "skills" / name
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
(d / "SKILL.md").write_text(f"---\nname: {name}\ndescription: {name} it\n---\n")
|
|
skill(TMP, "ask-form")
|
|
skill(TMP, "kept-home")
|
|
skill(TMP / "projects" / "film", "cut")
|
|
skill(TMP / "projects" / "other", "secret")
|
|
DEPS.users.set_defaults({"sharedSkills": ["ask-form", "missing"]})
|
|
names = {s["name"]: s["source"] for s in
|
|
self.guest().get("/api/guest-skills").json()["skills"]}
|
|
self.assertEqual(names, {"shared:ask-form": "shared", "cut": "film"})
|
|
cards = {s["name"]: s for s in
|
|
self.c.get("/api/skills", headers=OWNER_H).json()["skills"]}
|
|
self.assertTrue(cards["ask-form"]["sharedWithGuests"])
|
|
self.assertFalse(cards["kept-home"]["sharedWithGuests"])
|
|
self.assertEqual(cards["cut"]["shareRef"], "film/cut")
|
|
|
|
def test_sse_filter(self):
|
|
dom = guest_from_record(self.dom)
|
|
DEPS.meta_store.update("mine", {"owner": self.dom["id"]})
|
|
flt = _guest_filter(DEPS, dom)
|
|
ev = flt({"type": "transcript", "ids": ["-p/mine.jsonl", "-p/theirs.jsonl",
|
|
"-p/mine/subagents/agent-1.jsonl"]})
|
|
self.assertEqual(ev["ids"], ["-p/mine.jsonl", "-p/mine/subagents/agent-1.jsonl"])
|
|
self.assertIsNone(flt({"type": "cron"}))
|
|
self.assertIsNone(flt({"type": "meta", "id": "theirs"}))
|
|
|
|
# ── runs ─────────────────────────────────────────────────────────────────
|
|
def _capture(self):
|
|
calls = []
|
|
orig = run_routes._send_message
|
|
run_routes._send_message = lambda deps, prompt, **kw: (
|
|
calls.append((prompt, kw)) or {"sessionId": "new", "pid": 1})
|
|
self.addCleanup(setattr, run_routes, "_send_message", orig)
|
|
return calls
|
|
|
|
def test_guest_spawn_is_pinned_server_side(self):
|
|
calls = self._capture()
|
|
r = self.guest().post("/api/spawn", json={
|
|
"prompt": "hello", "projects": ["film"], "cwd": "/",
|
|
"account": "work", "worker": "mac", "harness": "pi", "model": "sonnet"})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
prompt, kw = calls[0]
|
|
self.assertTrue(prompt.startswith('<ai-agent-from user="dom"'))
|
|
self.assertEqual(kw["cwd"], "~/projects/film")
|
|
self.assertEqual((kw["account"], kw["worker"], kw["harness"]),
|
|
("personal", "local", "claude"))
|
|
self.assertEqual(kw["stamp"], {"owner": "dom", "projects": ["film"]})
|
|
self.assertIn("~/projects/film", kw["scope"]["writeDirs"])
|
|
self.assertFalse(kw["scope"]["remoteControl"])
|
|
|
|
def test_guest_spawn_with_extra_projects(self):
|
|
DEPS.users.update(self.dom["id"], {"projects": {"film": "chat", "other": "ship"}})
|
|
calls = self._capture()
|
|
g = self.guest()
|
|
r = g.post("/api/spawn", json={"prompt": "x", "projects": ["other", "film", "other"]})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
_, kw = calls[0]
|
|
self.assertEqual(kw["cwd"], "~/projects/other") # the select's pick runs first
|
|
self.assertEqual(kw["stamp"]["projects"], ["other", "film"])
|
|
self.assertEqual(kw["scope"]["writeDirs"],
|
|
["~/projects/other", "~/projects/film", "/tmp"])
|
|
self.assertIn("project `film`", kw["scope"]["systemPrompt"])
|
|
self.assertEqual(g.post("/api/spawn", json={
|
|
"prompt": "x", "projects": []}).status_code, 400)
|
|
|
|
def test_guest_spawn_refusals(self):
|
|
self._capture()
|
|
g = self.guest()
|
|
self.assertEqual(g.post("/api/spawn", json={
|
|
"prompt": "x", "projects": ["other"]}).status_code, 403)
|
|
self.assertEqual(g.post("/api/spawn", json={ # an extra needs chat too
|
|
"prompt": "x", "projects": ["film", "other"]}).status_code, 403)
|
|
self.assertEqual(g.post("/api/spawn", json={
|
|
"prompt": "x", "projects": ["film"], "model": "opus"}).status_code, 403)
|
|
claire = self.guest(DEPS.auth.open_session(self.claire["id"]))
|
|
self.assertEqual(claire.post("/api/spawn", json={ # view only
|
|
"prompt": "x", "projects": ["film"]}).status_code, 403)
|
|
|
|
def test_resume_needs_access_and_tags_the_author(self):
|
|
calls = self._capture()
|
|
sid = "11111111-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(sid, {"owner": self.claire["id"], "projects": ["film"]})
|
|
g = self.guest()
|
|
self.assertEqual(g.post("/api/resume", json={
|
|
"sessionId": sid, "prompt": "hi"}).status_code, 404)
|
|
DEPS.meta_store.update(sid, {"participants": [self.dom["id"]]})
|
|
self.assertEqual(g.post("/api/resume", json={
|
|
"sessionId": sid, "prompt": "hi", "cwd": "/"}).status_code, 200)
|
|
prompt, kw = calls[-1]
|
|
self.assertIn('name="Dom"', prompt)
|
|
self.assertNotIn("cwd", kw) # never the client's
|
|
# …and the owner's own turn in that shared thread is tagged too
|
|
self.c.post("/api/resume", json={"sessionId": sid, "prompt": "me"},
|
|
headers=OWNER_H)
|
|
self.assertTrue(calls[-1][0].startswith('<ai-agent-from user="owner"'))
|
|
|
|
def test_share_by_conversation_owner(self):
|
|
sid = "22222222-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(sid, {"owner": self.dom["id"], "projects": ["film"]})
|
|
r = self.guest().put("/api/conversation-share",
|
|
json={"id": sid, "participants": [self.claire["id"]]})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertEqual(DEPS.meta_store.peek(sid)["participants"], ["claire"])
|
|
self.assertTrue(access.visible(DEPS, guest_from_record(self.claire), sid))
|
|
|
|
# ── per-user notifications ───────────────────────────────────────────────
|
|
def test_audience(self):
|
|
self.assertEqual(access.audience({}), ["owner"])
|
|
self.assertEqual(access.audience({"owner": "dom"}), ["dom"])
|
|
self.assertEqual(access.audience({"owner": "dom", "participants": ["claire"]}),
|
|
["dom", "claire"])
|
|
self.assertEqual(access.audience({"participants": ["dom"]}), ["owner", "dom"])
|
|
# the owner follows a guest's conversation; a guest mutes one
|
|
self.assertEqual(access.audience({"owner": "dom", "follow": {"owner": True}}),
|
|
["dom", "owner"])
|
|
self.assertEqual(access.audience({"owner": "dom", "participants": ["claire"],
|
|
"follow": {"dom": False}}), ["claire"])
|
|
# a guest's "follow" can't add them to a conversation they don't see
|
|
self.assertEqual(access.audience({"follow": {"claire": True}}), ["owner"])
|
|
|
|
def _forwarded(self):
|
|
"""Capture what goes to the owner's webhooks (the phone)."""
|
|
import notifications.store as notify_store
|
|
sent = []
|
|
orig = notify_store.forward
|
|
notify_store.forward = lambda targets, payload: (
|
|
sent.append(payload) or [{"webhook": "ha", "ok": True, "status": 200}])
|
|
self.addCleanup(setattr, notify_store, "forward", orig)
|
|
return sent
|
|
|
|
def test_pushes_reach_the_conversation_audience(self):
|
|
sent = self._forwarded()
|
|
sid = "33333333-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(sid, {"owner": self.dom["id"], "projects": ["film"],
|
|
"participants": [self.claire["id"]]})
|
|
r = self.c.post("/api/notify", headers=OWNER_H, json={
|
|
"title": "done", "message": "m", "sessionId": sid}).json()
|
|
self.assertEqual(sent, []) # the owner doesn't follow it
|
|
self.assertEqual({d["user"] for d in r["delivered"]}, {"dom", "claire"})
|
|
self.assertTrue(r["ok"])
|
|
# the owner writes in it → now follows it → their phone hears the next one
|
|
self._capture()
|
|
self.c.post("/api/resume", json={"sessionId": sid, "prompt": "me"}, headers=OWNER_H)
|
|
r = self.c.post("/api/notify", headers=OWNER_H, json={
|
|
"title": "again", "message": "m", "sessionId": sid}).json()
|
|
self.assertEqual(len(sent), 1)
|
|
self.assertEqual({d["user"] for d in r["delivered"]}, {"dom", "claire", "owner"})
|
|
# the owner's own conversation: the phone only
|
|
sent.clear()
|
|
r = self.c.post("/api/notify", headers=OWNER_H, json={
|
|
"title": "mine", "message": "m", "sessionId": "own-session"}).json()
|
|
self.assertEqual(len(sent), 1)
|
|
self.assertEqual({d["user"] for d in r["delivered"]}, {"owner"})
|
|
|
|
def test_asks_reach_the_conversation_audience(self):
|
|
sent = self._forwarded()
|
|
sid = "44444444-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(sid, {"participants": [self.dom["id"]]})
|
|
r = self.c.post("/api/ask", headers=OWNER_H, json={
|
|
"question": "go?", "options": ["yes", "no"], "sessionId": sid}).json()
|
|
self.assertEqual(len(sent), 1)
|
|
self.assertEqual({d["user"] for d in r["delivered"]}, {"owner", "dom"})
|
|
self.assertEqual(DEPS.notify_store.ask_status(r["id"])["to"], ["owner", "dom"])
|
|
|
|
def test_feed_and_log_are_per_user(self):
|
|
self._forwarded()
|
|
mine, doms = "own-feed-session", "55555555-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(doms, {"owner": self.dom["id"], "projects": ["film"]})
|
|
for sid in (mine, doms):
|
|
self.c.post("/api/notify", headers=OWNER_H, json={
|
|
"title": f"from {sid}", "message": "m", "sessionId": sid})
|
|
orig = DEPS.store.all_summaries
|
|
DEPS.store.all_summaries = lambda: [
|
|
(f"/x/{s}.jsonl", {"sessionId": s, "messages": 1}) for s in (mine, doms)]
|
|
self.addCleanup(setattr, DEPS.store, "all_summaries", orig)
|
|
|
|
def seen(client, headers=None):
|
|
feed = client.get("/api/notifications", headers=headers).json()
|
|
log = client.get("/api/notify-log", headers=headers).json()
|
|
return ({c["id"] for c in feed["conversations"]},
|
|
{e["sessionId"] for e in log["notifications"]
|
|
if e["sessionId"] in (mine, doms)})
|
|
|
|
self.assertEqual(seen(self.c, OWNER_H), ({f"{mine}.jsonl"}, {mine}))
|
|
self.assertEqual(seen(self.guest()), ({f"{doms}.jsonl"}, {doms}))
|
|
flat = {n["convId"] for n in _flat_notifications(DEPS)}
|
|
self.assertNotIn(f"{doms}.jsonl", flat) # the desk phone's unread
|
|
# the owner follows Dom's conversation from its bell
|
|
r = self.c.put("/api/conversation-follow", headers=OWNER_H,
|
|
json={"id": doms, "on": True})
|
|
self.assertTrue(r.json()["notifying"], r.text)
|
|
self.assertEqual(seen(self.c, OWNER_H)[0], {f"{mine}.jsonl", f"{doms}.jsonl"})
|
|
# Dom mutes it: still visible to him, no longer in his feed
|
|
r = self.guest().put("/api/conversation-follow", json={"id": doms, "on": False})
|
|
self.assertFalse(r.json()["notifying"])
|
|
self.assertEqual(seen(self.guest()), (set(), set()))
|
|
self.assertEqual(self.guest().get(f"/api/conversation-share?id={doms}").status_code, 200)
|
|
|
|
def test_guest_links_point_at_the_guest_host(self):
|
|
from notifications.service import VIEW_URL
|
|
from users import deliver
|
|
if not deliver.GUEST_PUBLIC_URL:
|
|
deliver.GUEST_PUBLIC_URL = "https://claude.example.test"
|
|
self.addCleanup(setattr, deliver, "GUEST_PUBLIC_URL", "")
|
|
url = f"{VIEW_URL}/conversation/x.jsonl"
|
|
guest_url = deliver.url_for(guest_from_record(self.dom), url, VIEW_URL)
|
|
self.assertEqual(guest_url, "https://claude.example.test/conversation/x.jsonl")
|
|
self.assertEqual(deliver.url_for(OWNER, guest_url, VIEW_URL), url)
|
|
|
|
def test_forms_say_who_they_are_for(self):
|
|
sid = "66666666-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(sid, {"owner": self.dom["id"], "projects": ["film"]})
|
|
self.c.post("/api/forms", headers=OWNER_H, json={
|
|
"title": "pick", "fields": [{"key": "a", "type": "text", "label": "A"}],
|
|
"sessionId": sid})
|
|
mine = [f for f in self.c.get("/api/forms", headers=OWNER_H).json()["forms"]
|
|
if f["sessionId"] == sid]
|
|
doms = [f for f in self.guest().get("/api/forms").json()["forms"]
|
|
if f["sessionId"] == sid]
|
|
self.assertEqual([f["forYou"] for f in mine], [False]) # still listed
|
|
self.assertEqual([f["forYou"] for f in doms], [True])
|
|
|
|
# ── user admin: spend + impersonation ────────────────────────────────────
|
|
def test_users_list_carries_spend(self):
|
|
cards = [_card("a" * 8, owner=self.dom["id"]),
|
|
_card("b" * 8, owner=self.dom["id"]),
|
|
{**_card("c" * 8, owner=self.dom["id"]), "endedAt": "2020-01-01T00:00:00Z"},
|
|
_card("d" * 8)]
|
|
import datetime
|
|
now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT00:00:00Z")
|
|
for c in cards[:2]:
|
|
c["endedAt"] = now
|
|
import conversations.catalog as catalog
|
|
orig = catalog._conversation_cards
|
|
catalog._conversation_cards = lambda deps, full: cards
|
|
self.addCleanup(setattr, catalog, "_conversation_cards", orig)
|
|
users = {u["id"]: u for u in
|
|
self.c.get("/api/users", headers=OWNER_H).json()["users"]}
|
|
self.assertEqual((users["dom"]["spentUsd"], users["dom"]["totalSpentUsd"],
|
|
users["dom"]["conversations"]), (2.0, 3.0, 3))
|
|
self.assertEqual(users["claire"]["spentUsd"], 0)
|
|
|
|
def test_impersonation(self):
|
|
before = len(DEPS.auth.sessions_of(self.dom["id"]))
|
|
c = TestClient(main.app) # the owner's browser on the admin host
|
|
r = c.post(f"/api/users/{self.dom['id']}/impersonate", headers=OWNER_H)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertTrue(c.cookies.get(SESSION_COOKIE))
|
|
# the owner's credentials — and yet: Dom
|
|
me = c.get("/api/me", headers=OWNER_H).json()
|
|
self.assertEqual((me["id"], me["role"], me["impersonating"]), ("dom", "guest", True))
|
|
self.assertEqual(c.get("/api/users", headers=OWNER_H).status_code, 403)
|
|
# not one of Dom's own sessions, and no "Dom started…" ping on a spawn
|
|
self.assertEqual(len(DEPS.auth.sessions_of(self.dom["id"])), before)
|
|
self.assertIsNone(DEPS.users.get("dom")["lastLoginAt"])
|
|
calls, pings = self._capture(), []
|
|
orig = run_routes.deliver.owner_ping_start
|
|
run_routes.deliver.owner_ping_start = lambda *a: pings.append(a)
|
|
self.addCleanup(setattr, run_routes.deliver, "owner_ping_start", orig)
|
|
c.post("/api/spawn", json={"prompt": "x", "projects": ["film"]}, headers=OWNER_H)
|
|
self.assertEqual((len(calls), pings), (1, []))
|
|
# back to admin
|
|
c.post("/api/auth/logout", headers=OWNER_H)
|
|
self.assertEqual(c.get("/api/me", headers=OWNER_H).json()["role"], "owner")
|
|
# guests can't, and a disabled guest can't be impersonated
|
|
self.assertEqual(self.guest().post(
|
|
f"/api/users/{self.claire['id']}/impersonate").status_code, 403)
|
|
DEPS.users.update(self.claire["id"], {"disabled": True})
|
|
self.assertEqual(self.c.post(f"/api/users/{self.claire['id']}/impersonate",
|
|
headers=OWNER_H).status_code, 404)
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────
|
|
def test_author_tag_round_trip(self):
|
|
dom = guest_from_record(self.dom)
|
|
author, text = _split_author(f"{author_tag(dom)}\nfix the trailer")
|
|
self.assertEqual(author, {"id": "dom", "name": "Dom", "role": "guest"})
|
|
self.assertEqual(text, "fix the trailer")
|
|
self.assertEqual(_split_author("plain")[0], None)
|
|
|
|
def test_model_families(self):
|
|
dom = guest_from_record(self.dom)
|
|
self.assertEqual(check_model(dom, None), "sonnet")
|
|
self.assertEqual(check_model(dom, "claude-haiku-4-5-20251001"),
|
|
"claude-haiku-4-5-20251001")
|
|
self.assertTrue(OWNER.can("anything", "ship"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|