1import httpx2import pytest345async def test_login_me_tokens_rbac(app_client):6 c = app_client7 r = await c.get("/api/v1/auth/me"); assert r.status_code == 200 and r.json()["role"] == "admin"8 r = await c.get("/api/v1/overview", headers=c.auth_headers); assert r.status_code == 2009 anon = httpx.AsyncClient(transport=c._transport, base_url="http://test")10 assert (await anon.get("/api/v1/overview")).status_code == 40111 assert (await anon.get("/api/v1/overview", headers={"Authorization": "Bearer akos_bogus"})).status_code == 40112 assert (await anon.post("/api/v1/auth/login", json={"email": "admin@test.local", "password": "wrong"})).status_code == 40113 # viewer cannot create credentials (operator role) nor create users (admin)14 r = await c.post("/api/v1/auth/users", json={"email": "viewer@test.local", "password": "viewer-password-1", "role": "viewer"}, headers=c.auth_headers); assert r.status_code == 20015 v = httpx.AsyncClient(transport=c._transport, base_url="http://test")16 assert (await v.post("/api/v1/auth/login", json={"email": "viewer@test.local", "password": "viewer-password-1"})).status_code == 20017 assert (await v.post("/api/v1/credentials", json={"name": "x", "scheme": "bearer", "value": "abc"})).status_code == 40318 assert (await v.post("/api/v1/auth/users", json={"email": "z@test.local", "password": "zzzzzzzzzzzz", "role": "viewer"})).status_code == 40319 assert (await v.get("/api/v1/apis")).status_code == 20020 # token scopes: read-only token cannot execute21 r = await c.post("/api/v1/auth/tokens", json={"name": "ro", "scopes": ["read"]}); ro = r.json()["token"]22 r = await c.post("/api/v1/execute", json={"project_id": "default", "url": "http://127.0.0.1:1/"}, headers={"Authorization": f"Bearer {ro}"}); assert r.status_code == 40323 # revocation24 tid = (await c.get("/api/v1/auth/tokens")).json()["tokens"][0]["id"]25 assert (await c.delete(f"/api/v1/auth/tokens/{tid}")).status_code == 20026 ev = (await c.get("/api/v1/audit", params={"action": "token"}, headers=c.auth_headers)).json()["events"]27 assert any(e["action"] == "token.revoke" for e in ev)282930async def test_tenant_separation(app_client, pool, mock_api):31 c = app_client32 for slug in ("tenant-a", "tenant-b"):33 r = await c.post("/api/v1/projects", json={"slug": slug, "name": slug, "policy_mode": "discovery"}, headers=c.auth_headers); assert r.status_code in (200, 409)34 ra = await c.post("/api/v1/auth/tokens", json={"name": "a", "scopes": ["read", "execute"], "project_id": "tenant-a"}); ta = {"Authorization": f"Bearer {ra.json()['token']}"}35 rb = await c.post("/api/v1/auth/tokens", json={"name": "b", "scopes": ["read", "execute"], "project_id": "tenant-b"}); tb = {"Authorization": f"Bearer {rb.json()['token']}"}36 r = await c.post("/api/v1/execute", json={"project_id": "tenant-a", "url": mock_api["base"] + "/items", "purpose": "tenant test"}, headers=ta); assert r.status_code == 200, r.text37 ex = r.json()["id"]38 assert (await c.get(f"/api/v1/executions/{ex}", headers=ta)).status_code == 20039 assert (await c.get(f"/api/v1/executions/{ex}", headers=tb)).status_code == 40340 assert (await c.get(f"/api/v1/executions/{ex}/body", headers=tb)).status_code == 40341 r = await c.post("/api/v1/execute", json={"project_id": "tenant-a", "url": mock_api["base"] + "/items"}, headers=tb); assert r.status_code == 40342 lst = (await c.get("/api/v1/executions", headers=tb)).json()["executions"]43 assert all(e["project_slug"] == "tenant-b" for e in lst)44 projs = (await c.get("/api/v1/projects", headers=tb)).json()["projects"]45 assert [p["slug"] for p in projs] == ["tenant-b"]46