mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-29 19:21:40 +00:00
Add auth and per-user separation test suites
24 new tests: setup/login/session lifecycle, invites, resets (with and without SMTP), admin guards and the last-admin lockout guard, resource reassignment on user deletion, cross-user 404s on projects/skills/ connectors/plugins/models/schedules/commands/memory, private model backends rejected at spawn/schedule time, and claude_gen materializing only shared + owner rows at launch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019ws7xj5Ej623hh4GXQCYYR
This commit is contained in:
Claude
@@ -0,0 +1,281 @@
|
||||
"""Per-user separation: users see shared resources plus their own — never each
|
||||
other's — across projects, agents, schedules, memory, activity, and the Claude page
|
||||
resources; the control layer applies only the owner's rows at launch."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def users(client):
|
||||
"""Three actors: an admin, and two regular users (alice, bob)."""
|
||||
r = client.post("/auth/setup", json={"email": "admin@x.co", "password": "admin-pass-1"})
|
||||
admin = {"Authorization": f"Bearer {r.json()['token']}"}
|
||||
out = {"admin": admin, "admin_user": r.json()["user"]}
|
||||
for name in ("alice", "bob"):
|
||||
invite = client.post(
|
||||
"/auth/users", json={"email": f"{name}@x.co"}, headers=admin
|
||||
).json()
|
||||
token = invite["invite_url"].split("token=")[1]
|
||||
r = client.post("/auth/reset", json={"token": token, "password": f"{name}-pass-1"})
|
||||
out[name] = {"Authorization": f"Bearer {r.json()['token']}"}
|
||||
out[f"{name}_user"] = r.json()["user"]
|
||||
return out
|
||||
|
||||
|
||||
def _mkproject(client, headers, pid):
|
||||
r = client.post("/projects", json={"id": pid, "root_dir": f"/tmp/{pid}"}, headers=headers)
|
||||
assert r.status_code == 201
|
||||
return r.json()
|
||||
|
||||
|
||||
# ---- projects --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_projects_are_invisible_across_users(client, users):
|
||||
_mkproject(client, users["alice"], "alices")
|
||||
_mkproject(client, users["bob"], "bobs")
|
||||
|
||||
assert [p["id"] for p in client.get("/projects", headers=users["alice"]).json()] == ["alices"]
|
||||
assert [p["id"] for p in client.get("/projects", headers=users["bob"]).json()] == ["bobs"]
|
||||
# Existence is not leaked: someone else's project 404s, as do its nested routes.
|
||||
assert client.get("/projects/bobs", headers=users["alice"]).status_code == 404
|
||||
assert client.get("/projects/bobs/agents", headers=users["alice"]).status_code == 404
|
||||
assert client.delete("/projects/bobs", headers=users["alice"]).status_code == 404
|
||||
# Admin sees and can manage everything.
|
||||
assert {p["id"] for p in client.get("/projects", headers=users["admin"]).json()} == {
|
||||
"alices", "bobs",
|
||||
}
|
||||
|
||||
|
||||
def test_shared_projects_visible_but_admin_managed(client, users, env):
|
||||
token_headers = {"Authorization": f"Bearer {env['token']}"}
|
||||
_mkproject(client, token_headers, "sharedproj") # env token => shared (owner NULL)
|
||||
|
||||
assert client.get("/projects/sharedproj", headers=users["alice"]).status_code == 200
|
||||
# Viewing yes; mutating no — shared rows are admin-managed.
|
||||
r = client.patch(
|
||||
"/projects/sharedproj", json={"root_dir": "/tmp/x"}, headers=users["alice"]
|
||||
)
|
||||
assert r.status_code == 403
|
||||
assert client.patch(
|
||||
"/projects/sharedproj", json={"root_dir": "/tmp/x"}, headers=users["admin"]
|
||||
).status_code == 200
|
||||
|
||||
|
||||
def test_owner_operates_own_project_without_admin(client, users, fake_launch, tmp_path):
|
||||
"""A regular user drives the full lifecycle on their own project (spawn needs a
|
||||
worker; here we only prove the API-side gates: enqueue allowed, 404 for others)."""
|
||||
_mkproject(client, users["alice"], "alices")
|
||||
r = client.post(
|
||||
"/projects/alices/agents/spawn",
|
||||
json={"name": "a1", "task": "do the thing"},
|
||||
headers=users["alice"],
|
||||
)
|
||||
assert r.status_code == 202
|
||||
assert r.json()["requested_by"].startswith("user:")
|
||||
# Bob can't even see the project, let alone spawn into it.
|
||||
r = client.post(
|
||||
"/projects/alices/agents/spawn",
|
||||
json={"name": "b1", "task": "sneaky"},
|
||||
headers=users["bob"],
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_admin_can_reassign_owner(client, users):
|
||||
_mkproject(client, users["alice"], "alices")
|
||||
bob_id = users["bob_user"]["id"]
|
||||
r = client.patch(
|
||||
"/projects/alices", json={"owner_user_id": bob_id}, headers=users["admin"]
|
||||
)
|
||||
assert r.status_code == 200 and r.json()["owner_user_id"] == bob_id
|
||||
assert client.get("/projects/alices", headers=users["alice"]).status_code == 404
|
||||
assert client.get("/projects/alices", headers=users["bob"]).status_code == 200
|
||||
# Owners themselves cannot hand projects around.
|
||||
r = client.patch(
|
||||
"/projects/alices", json={"owner_user_id": None}, headers=users["bob"]
|
||||
)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
# ---- claude page resources (skills / connectors / plugins / models) --------------------
|
||||
|
||||
|
||||
def test_skills_are_separated_and_shared_rows_common(client, users, env):
|
||||
token_headers = {"Authorization": f"Bearer {env['token']}"}
|
||||
mk = lambda h, name: client.post( # noqa: E731
|
||||
"/claude/skills", json={"name": name, "content": "# x"}, headers=h
|
||||
)
|
||||
assert mk(token_headers, "shared-skill").status_code == 201
|
||||
alice_skill = mk(users["alice"], "alice-skill")
|
||||
assert alice_skill.status_code == 201
|
||||
assert alice_skill.json()["owner_user_id"] == users["alice_user"]["id"]
|
||||
assert mk(users["bob"], "bob-skill").status_code == 201
|
||||
|
||||
names = lambda h: {s["name"] for s in client.get("/claude/skills", headers=h).json()} # noqa: E731
|
||||
assert names(users["alice"]) == {"shared-skill", "alice-skill"}
|
||||
assert names(users["bob"]) == {"shared-skill", "bob-skill"}
|
||||
assert names(users["admin"]) == {"shared-skill", "alice-skill", "bob-skill"}
|
||||
|
||||
# Cross-user mutation 404s (invisible); shared mutation 403s for non-admins.
|
||||
alice_id = alice_skill.json()["id"]
|
||||
assert client.delete(f"/claude/skills/{alice_id}", headers=users["bob"]).status_code == 404
|
||||
shared_id = next(
|
||||
s["id"] for s in client.get("/claude/skills", headers=users["admin"]).json()
|
||||
if s["name"] == "shared-skill"
|
||||
)
|
||||
assert client.patch(
|
||||
f"/claude/skills/{shared_id}", json={"enabled": False}, headers=users["alice"]
|
||||
).status_code == 403
|
||||
assert client.delete(f"/claude/skills/{alice_id}", headers=users["alice"]).status_code == 200
|
||||
|
||||
|
||||
def test_connectors_plugins_models_follow_same_rules(client, users):
|
||||
a, b = users["alice"], users["bob"]
|
||||
r = client.post(
|
||||
"/claude/connectors",
|
||||
json={"name": "alice-mcp", "transport": "http", "url": "https://a.example/mcp"},
|
||||
headers=a,
|
||||
)
|
||||
assert r.status_code == 201
|
||||
r = client.post(
|
||||
"/claude/plugins",
|
||||
json={"name": "alice-plug", "marketplace": "mp", "marketplace_repo": "o/r"},
|
||||
headers=a,
|
||||
)
|
||||
assert r.status_code == 201
|
||||
r = client.post(
|
||||
"/claude/models",
|
||||
json={"name": "alice-model", "base_url": "http://localhost:4000", "model": "m"},
|
||||
headers=a,
|
||||
)
|
||||
assert r.status_code == 201
|
||||
model_id = r.json()["id"]
|
||||
|
||||
assert client.get("/claude/connectors", headers=b).json() == []
|
||||
assert client.get("/claude/plugins", headers=b).json() == []
|
||||
assert client.get("/claude/models", headers=b).json() == []
|
||||
|
||||
# Bob can't spawn or schedule onto Alice's private model backend.
|
||||
_mkproject(client, b, "bobs")
|
||||
r = client.post(
|
||||
"/projects/bobs/agents/spawn",
|
||||
json={"name": "b1", "task": "t", "model_id": model_id},
|
||||
headers=b,
|
||||
)
|
||||
assert r.status_code == 400 and "not found" in r.json()["detail"]
|
||||
r = client.post(
|
||||
"/projects/bobs/schedules",
|
||||
json={"name_prefix": "s", "task": "t", "interval_seconds": 3600, "model_id": model_id},
|
||||
headers=b,
|
||||
)
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
# ---- schedules, activity, memory -------------------------------------------------------
|
||||
|
||||
|
||||
def test_schedules_follow_project_visibility(client, users):
|
||||
_mkproject(client, users["alice"], "alices")
|
||||
r = client.post(
|
||||
"/projects/alices/schedules",
|
||||
json={"name_prefix": "nightly", "task": "t", "interval_seconds": 3600},
|
||||
headers=users["alice"],
|
||||
)
|
||||
assert r.status_code == 201
|
||||
sid = r.json()["id"]
|
||||
|
||||
assert client.get("/schedules", headers=users["bob"]).json() == []
|
||||
assert client.get("/projects/alices/schedules", headers=users["bob"]).status_code == 404
|
||||
assert client.patch(
|
||||
f"/schedules/{sid}", json={"enabled": False}, headers=users["bob"]
|
||||
).status_code == 404
|
||||
assert len(client.get("/schedules", headers=users["admin"]).json()) == 1
|
||||
assert client.delete(f"/schedules/{sid}", headers=users["alice"]).status_code == 200
|
||||
|
||||
|
||||
def test_activity_feed_is_scoped(client, users):
|
||||
_mkproject(client, users["alice"], "alices")
|
||||
r = client.post(
|
||||
"/projects/alices/agents/spawn",
|
||||
json={"name": "a1", "task": "t"},
|
||||
headers=users["alice"],
|
||||
)
|
||||
cmd_id = r.json()["id"]
|
||||
|
||||
assert client.get("/commands", headers=users["bob"]).json() == []
|
||||
assert client.get(f"/commands/{cmd_id}", headers=users["bob"]).status_code == 404
|
||||
assert client.get(f"/commands/{cmd_id}", headers=users["alice"]).status_code == 200
|
||||
assert len(client.get("/commands", headers=users["admin"]).json()) == 1
|
||||
|
||||
|
||||
def test_memory_notes_follow_project_visibility(client, users):
|
||||
_mkproject(client, users["alice"], "alices")
|
||||
r = client.post(
|
||||
"/memory/notes",
|
||||
json={"title": "alice fact", "body": "b", "kind": "fact", "project_id": "alices"},
|
||||
headers=users["alice"],
|
||||
)
|
||||
assert r.status_code == 201
|
||||
note_id = r.json()["id"]
|
||||
|
||||
# Global notes reach every user's agents, so only admins write them.
|
||||
r = client.post(
|
||||
"/memory/notes", json={"title": "global", "body": "b", "kind": "fact"},
|
||||
headers=users["alice"],
|
||||
)
|
||||
assert r.status_code == 403
|
||||
assert client.post(
|
||||
"/memory/notes", json={"title": "global", "body": "b", "kind": "fact"},
|
||||
headers=users["admin"],
|
||||
).status_code == 201
|
||||
|
||||
bob_titles = {n["title"] for n in client.get("/memory/notes", headers=users["bob"]).json()}
|
||||
assert bob_titles == {"global"} # global visible, alice's project note not
|
||||
assert client.get(f"/memory/notes/{note_id}", headers=users["bob"]).status_code == 404
|
||||
graph = client.get("/memory/graph", headers=users["bob"]).json()
|
||||
assert {n["title"] for n in graph["notes"]} == {"global"}
|
||||
|
||||
|
||||
# ---- control layer: what a launch materializes -----------------------------------------
|
||||
|
||||
|
||||
def test_launch_applies_only_owner_and_shared_rows(client, users, conn, tmp_path):
|
||||
"""claude_gen.apply for a project owned by alice syncs shared + alice's skills and
|
||||
connectors — never bob's."""
|
||||
from handler.control import claude_gen
|
||||
from handler.db import repository as repo
|
||||
|
||||
alice_id = users["alice_user"]["id"]
|
||||
bob_id = users["bob_user"]["id"]
|
||||
repo.create_claude_skill(conn, "shared-skill", "# s")
|
||||
repo.create_claude_skill(conn, "alice-skill", "# a", owner_user_id=alice_id)
|
||||
repo.create_claude_skill(conn, "bob-skill", "# b", owner_user_id=bob_id)
|
||||
repo.create_claude_connector(
|
||||
conn, "alice-mcp", "http", url="https://a.example/mcp", owner_user_id=alice_id
|
||||
)
|
||||
repo.create_claude_connector(
|
||||
conn, "bob-mcp", "http", url="https://b.example/mcp", owner_user_id=bob_id
|
||||
)
|
||||
|
||||
workdir = tmp_path / "wd"
|
||||
workdir.mkdir()
|
||||
summary = claude_gen.apply(str(workdir), conn=conn, visible_to=alice_id)
|
||||
assert summary["skills_written"] == 2 # shared + alice's
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
mcp = json.load(open(claude_gen.mcp_config_path(str(workdir))))
|
||||
assert "alice-mcp" in mcp["mcpServers"] and "bob-mcp" not in mcp["mcpServers"]
|
||||
skills_root = os.path.expanduser("~/.claude/skills")
|
||||
synced = set(os.listdir(skills_root))
|
||||
assert {"shared-skill", "alice-skill"} <= synced and "bob-skill" not in synced
|
||||
|
||||
# A shared/legacy project (owner None) gets shared rows only.
|
||||
summary = claude_gen.apply(str(workdir), conn=conn, visible_to=None)
|
||||
assert summary["skills_written"] == 1
|
||||
synced = set(os.listdir(skills_root))
|
||||
assert "alice-skill" not in synced and "shared-skill" in synced
|
||||
@@ -0,0 +1,250 @@
|
||||
"""User accounts: first-run setup, sign-in, sessions, resets/invites, admin management.
|
||||
|
||||
The email flows run with SMTP unconfigured (the default test env), which is itself a
|
||||
supported mode: links are returned to the admin instead of mailed. Delivery is covered
|
||||
by faking ``emailer.send`` where it matters.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_session(client):
|
||||
"""Complete first-run setup; returns (headers, user) for the created admin."""
|
||||
r = client.post(
|
||||
"/auth/setup", json={"email": "admin@example.com", "password": "admin-pass-1"}
|
||||
)
|
||||
assert r.status_code == 201
|
||||
body = r.json()
|
||||
assert body["user"]["is_admin"] is True
|
||||
return {"Authorization": f"Bearer {body['token']}"}, body["user"]
|
||||
|
||||
|
||||
def _invite(client, admin_headers, email, is_admin=False):
|
||||
r = client.post(
|
||||
"/auth/users", json={"email": email, "is_admin": is_admin}, headers=admin_headers
|
||||
)
|
||||
assert r.status_code == 201
|
||||
return r.json()
|
||||
|
||||
|
||||
def _accept(client, invite, password):
|
||||
token = invite["invite_url"].split("token=")[1]
|
||||
r = client.post("/auth/reset", json={"token": token, "password": password})
|
||||
assert r.status_code == 200
|
||||
return {"Authorization": f"Bearer {r.json()['token']}"}, r.json()["user"]
|
||||
|
||||
|
||||
# ---- first-run setup -------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_status_flips_after_setup(client):
|
||||
assert client.get("/auth/status").json()["initialized"] is False
|
||||
client.post("/auth/setup", json={"email": "a@b.co", "password": "password-1"})
|
||||
assert client.get("/auth/status").json()["initialized"] is True
|
||||
|
||||
|
||||
def test_first_user_is_admin_and_second_setup_refused(client, admin_session):
|
||||
headers, user = admin_session
|
||||
assert user["is_admin"] is True
|
||||
r = client.post("/auth/setup", json={"email": "x@y.co", "password": "password-1"})
|
||||
assert r.status_code == 409
|
||||
|
||||
|
||||
def test_setup_rejects_bad_email_and_short_password(client):
|
||||
bad_email = client.post("/auth/setup", json={"email": "nope", "password": "password-1"})
|
||||
assert bad_email.status_code == 422
|
||||
short = client.post("/auth/setup", json={"email": "a@b.co", "password": "short"})
|
||||
assert short.status_code == 422
|
||||
|
||||
|
||||
# ---- sign-in / session lifecycle -------------------------------------------------------
|
||||
|
||||
|
||||
def test_login_logout_me(client, admin_session):
|
||||
r = client.post("/auth/login", json={"email": "Admin@Example.COM", "password": "admin-pass-1"})
|
||||
assert r.status_code == 200 # email matching is case-insensitive
|
||||
headers = {"Authorization": f"Bearer {r.json()['token']}"}
|
||||
me = client.get("/auth/me", headers=headers).json()
|
||||
assert me == {
|
||||
"kind": "user", "user_id": r.json()["user"]["id"],
|
||||
"email": "admin@example.com", "is_admin": True,
|
||||
}
|
||||
assert client.post("/auth/logout", headers=headers).status_code == 200
|
||||
assert client.get("/auth/me", headers=headers).status_code == 401
|
||||
|
||||
|
||||
def test_login_rejects_wrong_password_and_unknown_email(client, admin_session):
|
||||
assert client.post(
|
||||
"/auth/login", json={"email": "admin@example.com", "password": "wrong-pass"}
|
||||
).status_code == 401
|
||||
assert client.post(
|
||||
"/auth/login", json={"email": "ghost@example.com", "password": "whatever-1"}
|
||||
).status_code == 401
|
||||
|
||||
|
||||
def test_disabled_user_cannot_login_and_live_session_dies(client, admin_session):
|
||||
admin_headers, _ = admin_session
|
||||
invite = _invite(client, admin_headers, "dev@example.com")
|
||||
dev_headers, dev = _accept(client, invite, "dev-password-1")
|
||||
|
||||
r = client.patch(f"/auth/users/{dev['id']}", json={"disabled": True}, headers=admin_headers)
|
||||
assert r.status_code == 200 and r.json()["disabled"] is True
|
||||
assert client.post(
|
||||
"/auth/login", json={"email": "dev@example.com", "password": "dev-password-1"}
|
||||
).status_code == 403
|
||||
# The existing session stops resolving too — disable means locked out now.
|
||||
assert client.get("/auth/me", headers=dev_headers).status_code == 401
|
||||
|
||||
|
||||
def test_change_password_revokes_other_sessions(client, admin_session):
|
||||
headers, user = admin_session
|
||||
other = client.post(
|
||||
"/auth/login", json={"email": "admin@example.com", "password": "admin-pass-1"}
|
||||
)
|
||||
other_headers = {"Authorization": f"Bearer {other.json()['token']}"}
|
||||
|
||||
r = client.post(
|
||||
"/auth/change-password",
|
||||
json={"current_password": "admin-pass-1", "new_password": "admin-pass-2"},
|
||||
headers=headers,
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert client.get("/auth/me", headers=headers).status_code == 200 # this session lives
|
||||
assert client.get("/auth/me", headers=other_headers).status_code == 401 # others die
|
||||
assert client.post(
|
||||
"/auth/login", json={"email": "admin@example.com", "password": "admin-pass-2"}
|
||||
).status_code == 200
|
||||
|
||||
wrong = client.post(
|
||||
"/auth/change-password",
|
||||
json={"current_password": "nope-nope-1", "new_password": "admin-pass-3"},
|
||||
headers=headers,
|
||||
)
|
||||
assert wrong.status_code == 403
|
||||
|
||||
|
||||
# ---- invites & resets ------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_invite_flow_creates_usable_account(client, admin_session):
|
||||
admin_headers, _ = admin_session
|
||||
invite = _invite(client, admin_headers, "Dev@Example.com")
|
||||
assert invite["emailed"] is False # SMTP unconfigured -> link only
|
||||
assert invite["user"]["has_password"] is False
|
||||
|
||||
dev_headers, dev = _accept(client, invite, "dev-password-1")
|
||||
assert dev["email"] == "dev@example.com" and dev["is_admin"] is False
|
||||
assert client.get("/auth/me", headers=dev_headers).json()["email"] == "dev@example.com"
|
||||
# The invite link is one-shot.
|
||||
token = invite["invite_url"].split("token=")[1]
|
||||
assert client.post(
|
||||
"/auth/reset", json={"token": token, "password": "again-password-1"}
|
||||
).status_code == 400
|
||||
|
||||
|
||||
def test_invite_duplicate_email_conflicts(client, admin_session):
|
||||
admin_headers, _ = admin_session
|
||||
_invite(client, admin_headers, "dev@example.com")
|
||||
r = client.post("/auth/users", json={"email": "DEV@example.com"}, headers=admin_headers)
|
||||
assert r.status_code == 409
|
||||
|
||||
|
||||
def test_admin_reset_link_and_forgot(client, admin_session, monkeypatch):
|
||||
admin_headers, admin = admin_session
|
||||
invite = _invite(client, admin_headers, "dev@example.com")
|
||||
dev_headers, dev = _accept(client, invite, "dev-password-1")
|
||||
|
||||
# Admin-minted reset link works and revokes the old session on use.
|
||||
r = client.post(f"/auth/users/{dev['id']}/reset-link", headers=admin_headers)
|
||||
assert r.status_code == 200
|
||||
token = r.json()["reset_url"].split("token=")[1]
|
||||
reset = client.post("/auth/reset", json={"token": token, "password": "dev-password-2"})
|
||||
assert reset.status_code == 200
|
||||
assert client.get("/auth/me", headers=dev_headers).status_code == 401
|
||||
|
||||
# Self-serve forgot: without SMTP it reports emailed=False and mints nothing.
|
||||
r = client.post("/auth/forgot", json={"email": "dev@example.com"})
|
||||
assert r.json() == {"ok": True, "emailed": False}
|
||||
|
||||
# With (faked) SMTP configured, the link lands in an email — capture and use it.
|
||||
sent = []
|
||||
from handler import emailer
|
||||
|
||||
monkeypatch.setattr(emailer, "configured", lambda settings=None: True)
|
||||
monkeypatch.setattr(
|
||||
emailer, "send", lambda to, subject, body, settings=None: sent.append((to, subject, body))
|
||||
)
|
||||
r = client.post("/auth/forgot", json={"email": "dev@example.com"})
|
||||
assert r.json() == {"ok": True, "emailed": True}
|
||||
assert sent and sent[0][0] == "dev@example.com"
|
||||
emailed_token = sent[0][2].split("token=")[1].split()[0]
|
||||
assert client.post(
|
||||
"/auth/reset", json={"token": emailed_token, "password": "dev-password-3"}
|
||||
).status_code == 200
|
||||
# Unknown addresses get the same answer and no email.
|
||||
sent.clear()
|
||||
assert client.post("/auth/forgot", json={"email": "ghost@example.com"}).json()["ok"] is True
|
||||
assert sent == []
|
||||
|
||||
|
||||
# ---- admin management guards -----------------------------------------------------------
|
||||
|
||||
|
||||
def test_user_management_is_admin_only(client, admin_session):
|
||||
admin_headers, _ = admin_session
|
||||
invite = _invite(client, admin_headers, "dev@example.com")
|
||||
dev_headers, dev = _accept(client, invite, "dev-password-1")
|
||||
|
||||
assert client.get("/auth/users", headers=dev_headers).status_code == 403
|
||||
assert client.post(
|
||||
"/auth/users", json={"email": "x@y.co"}, headers=dev_headers
|
||||
).status_code == 403
|
||||
assert client.patch(
|
||||
f"/auth/users/{dev['id']}", json={"is_admin": True}, headers=dev_headers
|
||||
).status_code == 403
|
||||
|
||||
listed = client.get("/auth/users", headers=admin_headers).json()
|
||||
assert {u["email"] for u in listed} == {"admin@example.com", "dev@example.com"}
|
||||
|
||||
|
||||
def test_last_admin_cannot_be_demoted_disabled_or_deleted(client, admin_session):
|
||||
admin_headers, admin = admin_session
|
||||
for body in ({"is_admin": False}, {"disabled": True}):
|
||||
r = client.patch(f"/auth/users/{admin['id']}", json=body, headers=admin_headers)
|
||||
assert r.status_code == 400, body
|
||||
assert client.delete(f"/auth/users/{admin['id']}", headers=admin_headers).status_code == 400
|
||||
|
||||
# With a second active admin the original may step down.
|
||||
invite = _invite(client, admin_headers, "admin2@example.com", is_admin=True)
|
||||
_accept(client, invite, "admin2-pass-1")
|
||||
r = client.patch(f"/auth/users/{admin['id']}", json={"is_admin": False}, headers=admin_headers)
|
||||
assert r.status_code == 200 and r.json()["is_admin"] is False
|
||||
|
||||
|
||||
def test_deleting_a_user_shares_their_resources(client, admin_session, conn):
|
||||
admin_headers, _ = admin_session
|
||||
invite = _invite(client, admin_headers, "dev@example.com")
|
||||
dev_headers, dev = _accept(client, invite, "dev-password-1")
|
||||
|
||||
r = client.post(
|
||||
"/projects", json={"id": "devproj", "root_dir": "/tmp/devproj"}, headers=dev_headers
|
||||
)
|
||||
assert r.status_code == 201 and r.json()["owner_user_id"] == dev["id"]
|
||||
|
||||
r = client.delete(f"/auth/users/{dev['id']}", headers=admin_headers)
|
||||
assert r.status_code == 200
|
||||
project = client.get("/projects/devproj", headers=admin_headers).json()
|
||||
assert project["owner_user_id"] is None # reassigned to shared, not orphaned
|
||||
assert client.get("/auth/me", headers=dev_headers).status_code == 401
|
||||
|
||||
|
||||
def test_legacy_env_tokens_keep_working(client, admin_session, env):
|
||||
token_headers = {"Authorization": f"Bearer {env['token']}"}
|
||||
me = client.get("/auth/me", headers=token_headers).json()
|
||||
assert me["kind"] == "token" and me["user_id"] is None
|
||||
# ADMIN_TOKEN unset falls back to AUTH_TOKEN, so the env token passes admin gates.
|
||||
assert client.get("/auth/users", headers=token_headers).status_code == 200
|
||||
assert client.get("/projects", headers=token_headers).status_code == 200
|
||||
Reference in New Issue
Block a user