File size: 4,512 Bytes
c794b6b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """Authentication contract tests."""
import os
import sys
from unittest.mock import MagicMock
os.environ.setdefault("CEPHEUS_CLOUD", "1")
os.environ.setdefault("CEPHEUS_API_KEY", "test-key")
os.environ.setdefault("CEPHEUS_AUTH_DEV_MODE", "1")
for mod in (
"google.adk",
"google.adk.agents",
"google.adk.runners",
"google.adk.sessions",
):
sys.modules.setdefault(mod, MagicMock())
BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if BACKEND_DIR not in sys.path:
sys.path.insert(0, BACKEND_DIR)
from fastapi.testclient import TestClient # noqa: E402
import main # noqa: E402
client = TestClient(main.app)
API_HEADERS = {"X-API-Key": "test-key"}
def test_auth_status_dev_mode():
r = client.get("/auth/status")
assert r.status_code == 200
body = r.json()
assert body["auth_enabled"] is True
assert body["mode"] == "dev"
def test_login_and_me_jwt():
r = client.post("/auth/login", json={"username": "admin", "password": "admin"})
assert r.status_code == 200
tokens = r.json()
assert tokens.get("access_token")
bearer = {"Authorization": f"Bearer {tokens['access_token']}"}
me = client.get("/auth/me", headers=bearer)
assert me.status_code == 200
assert me.json()["role"] == "admin"
def test_login_invalid_credentials():
r = client.post("/auth/login", json={"username": "admin", "password": "wrong"})
assert r.status_code == 401
def test_refresh_token_rotation():
login = client.post("/auth/login", json={"username": "staff", "password": "staff"}).json()
refreshed = client.post("/auth/refresh", json={"refresh_token": login["refresh_token"]})
assert refreshed.status_code == 200
assert refreshed.json().get("access_token")
def test_staff_accept_and_complete_request():
staff_headers = {
"Authorization": f"Bearer {client.post('/auth/login', json={'username': 'staff', 'password': 'staff'}).json()['access_token']}"
}
admin_headers = {
"Authorization": f"Bearer {client.post('/auth/login', json={'username': 'admin', 'password': 'admin'}).json()['access_token']}"
}
issue = client.post(
"/issues",
headers=admin_headers,
json={"title": "Test corridor smoke", "desc": "Smoke test", "status": "ONGOING", "priority": "medium", "staff": 1},
).json()
issue_id = issue["id"]
alert = client.post(
"/alert",
headers=admin_headers,
json={
"type": "staff_request",
"location": "Corridor B",
"message": f"Staff needed for {issue['title']}",
"issue_id": issue_id,
},
)
assert alert.status_code == 200
pending = client.get("/staff/requests", headers=staff_headers).json()
req = next(r for r in pending if r.get("issue_id") == issue_id)
assert req["status"] in ("pending", "sent")
accepted = client.post(
f"/staff/requests/{req['id']}/accept",
headers=staff_headers,
json={"staff_name": "staff"},
)
assert accepted.status_code == 200
assert accepted.json()["progress"] == 25
completed = client.post(
f"/staff/requests/{req['id']}/complete",
headers=staff_headers,
json={"notes": "Area cleared"},
)
assert completed.status_code == 200
body = completed.json()
assert body["status"] == "resolved"
assert body["progress"] == 100
assert body["assignments"][0]["stage"] == "resolved"
logs = client.get("/incident/logs", headers=admin_headers).json()
assert any(
entry.get("event_type") == "staff_request_completed" and req["id"] in entry.get("msg", "")
for entry in logs
)
issues = client.get("/issues", headers=admin_headers).json()
updated_issue = next(i for i in issues if i["id"] == issue_id)
assert updated_issue["status"] == "RESOLVED"
assert updated_issue["progress"] == 100
def test_public_vision_staff_login_returns_jwt(monkeypatch):
"""HF public vision must still issue JWT for staff portal mutations."""
monkeypatch.setenv("ALLOW_PUBLIC_VISION", "1")
monkeypatch.setenv("CEPHEUS_JWT_SECRET", "test-jwt-secret-for-public-vision")
monkeypatch.setenv("CEPHEUS_AUTH_DEV_MODE", "0")
r = client.post("/auth/login", json={"username": "staff", "password": "staff"})
assert r.status_code == 200
body = r.json()
assert body.get("access_token")
assert body.get("token_type") == "bearer"
assert body.get("user", {}).get("role") == "staff"
|