307 lines
12 KiB
Python
307 lines
12 KiB
Python
import threading
|
|
|
|
import pytest
|
|
from sqlalchemy import delete, func, select
|
|
|
|
from app.core.db import SessionLocal, engine
|
|
from app.models.audit import AuditEvent
|
|
from app.models.vehicle import Vehicle
|
|
from app.seed_loader import reset_and_seed
|
|
from app.services.audit import record_audit_event
|
|
|
|
|
|
def test_unauthenticated_dashboard_is_rejected(client):
|
|
response = client.get("/api/v1/dashboard")
|
|
assert response.status_code == 401
|
|
assert response.json()["error"]["code"] == "401"
|
|
|
|
|
|
def test_oidc_status_is_disabled_without_configuration(client):
|
|
assert client.get("/api/v1/auth/oidc/status").json() == {
|
|
"enabled": False,
|
|
"provider_name": None,
|
|
}
|
|
assert client.get("/api/v1/auth/oidc/login").status_code == 404
|
|
|
|
|
|
def test_oidc_callback_auto_provisions_and_logs_in(client, monkeypatch):
|
|
import app.api.routers.auth as auth_router
|
|
|
|
class FakeClient:
|
|
async def authorize_access_token(self, _request):
|
|
return {
|
|
"userinfo": {
|
|
"sub": "external-subject-1",
|
|
"email": "oidc.user@example.test",
|
|
"email_verified": True,
|
|
"name": "OIDC User",
|
|
}
|
|
}
|
|
|
|
class FakeOAuth:
|
|
def create_client(self, _name):
|
|
return FakeClient()
|
|
|
|
monkeypatch.setattr(auth_router.settings, "oidc_enabled", True)
|
|
monkeypatch.setattr(auth_router.settings, "oidc_issuer_url", "https://id.example.test")
|
|
monkeypatch.setattr(auth_router.settings, "oidc_client_id", "client")
|
|
monkeypatch.setattr(auth_router.settings, "oidc_client_secret", "secret")
|
|
monkeypatch.setattr(auth_router.settings, "oidc_allowed_email_domains", "example.test")
|
|
monkeypatch.setattr(auth_router, "oauth", FakeOAuth())
|
|
|
|
response = client.get("/api/v1/auth/oidc/callback", follow_redirects=False)
|
|
assert response.status_code == 307
|
|
assert response.headers["location"].endswith("/dashboard")
|
|
session = client.get("/api/v1/auth/session")
|
|
assert session.status_code == 200
|
|
assert session.json()["display_name"] == "OIDC User"
|
|
assert session.json()["role"] == "rental_employee"
|
|
|
|
|
|
def test_oidc_callback_rejects_missing_email_verification_claim(client, monkeypatch):
|
|
import app.api.routers.auth as auth_router
|
|
|
|
class FakeClient:
|
|
async def authorize_access_token(self, _request):
|
|
return {"userinfo": {"sub": "unverified-subject", "email": "new@example.test"}}
|
|
|
|
class FakeOAuth:
|
|
def create_client(self, _name):
|
|
return FakeClient()
|
|
|
|
monkeypatch.setattr(auth_router.settings, "oidc_enabled", True)
|
|
monkeypatch.setattr(auth_router.settings, "oidc_issuer_url", "https://id.example.test")
|
|
monkeypatch.setattr(auth_router.settings, "oidc_client_id", "client")
|
|
monkeypatch.setattr(auth_router.settings, "oidc_client_secret", "secret")
|
|
monkeypatch.setattr(auth_router.settings, "oidc_allowed_email_domains", "example.test")
|
|
monkeypatch.setattr(auth_router, "oauth", FakeOAuth())
|
|
|
|
response = client.get("/api/v1/auth/oidc/callback", follow_redirects=False)
|
|
assert response.status_code == 401
|
|
assert client.get("/api/v1/auth/session").status_code == 401
|
|
|
|
|
|
def test_demo_login_grants_access(ops_client):
|
|
response = ops_client.get("/api/v1/dashboard")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_rental_employee_cannot_reset_demo(employee_client):
|
|
response = employee_client.post("/api/v1/demo/reset")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_operations_manager_can_reset_demo(ops_client):
|
|
response = ops_client.post("/api/v1/demo/reset")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["counts"]["vehicles"] == 50
|
|
assert body["anchor_date"]
|
|
assert body["seeded_at"]
|
|
assert body["scenario_integrity"]["all_ready"] is True
|
|
assert body["scenario_integrity"]["not_ready"] == []
|
|
|
|
|
|
def test_demo_reset_rejects_concurrent_rebuild(ops_client):
|
|
import app.api.routers.demo as demo_router
|
|
|
|
assert demo_router._reset_guard.acquire(blocking=False)
|
|
try:
|
|
response = ops_client.post("/api/v1/demo/reset")
|
|
finally:
|
|
demo_router._reset_guard.release()
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_demo_reset_database_lock_blocks_another_api_replica(ops_client):
|
|
import app.api.routers.demo as demo_router
|
|
|
|
# Use a raw connection: an ordinary Session intentionally participates in the
|
|
# shared mutation barrier and would make an exclusive reset wait before it can test
|
|
# this separate non-blocking replica lock.
|
|
with engine.connect() as lock_connection:
|
|
lock_connection.scalar(select(func.pg_advisory_lock(demo_router._RESET_ADVISORY_LOCK_ID)))
|
|
try:
|
|
response = ops_client.post("/api/v1/demo/reset")
|
|
finally:
|
|
lock_connection.scalar(
|
|
select(func.pg_advisory_unlock(demo_router._RESET_ADVISORY_LOCK_ID))
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_demo_reset_waits_for_active_mutation_and_then_replaces_it_atomically():
|
|
started = threading.Event()
|
|
completed = threading.Event()
|
|
errors: list[BaseException] = []
|
|
|
|
def reset_in_other_session() -> None:
|
|
started.set()
|
|
try:
|
|
with SessionLocal() as reset_db:
|
|
reset_and_seed(reset_db)
|
|
except BaseException as exc: # pragma: no cover - assertion reports thread failures
|
|
errors.append(exc)
|
|
finally:
|
|
completed.set()
|
|
|
|
with SessionLocal() as mutation_db:
|
|
vehicle = mutation_db.scalar(
|
|
select(Vehicle).where(Vehicle.public_ref == "MO-001").with_for_update()
|
|
)
|
|
vehicle.location = "Concurrent mutation marker"
|
|
mutation_db.flush()
|
|
worker = threading.Thread(target=reset_in_other_session, daemon=True)
|
|
worker.start()
|
|
assert started.wait(timeout=1)
|
|
assert not completed.wait(timeout=0.2)
|
|
mutation_db.commit()
|
|
|
|
worker.join(timeout=10)
|
|
assert not worker.is_alive(), "reset deadlocked behind the active mutation"
|
|
assert errors == []
|
|
with SessionLocal() as db:
|
|
restored = db.scalar(select(Vehicle).where(Vehicle.public_ref == "MO-001"))
|
|
assert restored.location != "Concurrent mutation marker"
|
|
|
|
|
|
def test_demo_reset_cooldown_returns_retry_after(ops_client, monkeypatch):
|
|
import app.api.routers.demo as demo_router
|
|
|
|
monkeypatch.setattr(demo_router.settings, "demo_reset_cooldown_seconds", 60)
|
|
with SessionLocal() as db:
|
|
cooldown_event = record_audit_event(
|
|
db,
|
|
actor_type="user",
|
|
actor_label="Cooldown test",
|
|
action="demo_reset",
|
|
entity_type="system",
|
|
)
|
|
db.commit()
|
|
cooldown_event_id = cooldown_event.id
|
|
response = ops_client.post("/api/v1/demo/reset")
|
|
assert response.status_code == 429
|
|
assert int(response.headers["retry-after"]) >= 1
|
|
with SessionLocal() as db:
|
|
db.execute(delete(AuditEvent).where(AuditEvent.id == cooldown_event_id))
|
|
db.commit()
|
|
|
|
|
|
def test_demo_reset_rolls_back_every_change_when_integrity_check_fails(ops_client, monkeypatch):
|
|
import app.api.routers.demo as demo_router
|
|
|
|
monkeypatch.setattr(demo_router.settings, "demo_reset_cooldown_seconds", 0)
|
|
with SessionLocal() as db:
|
|
probe = record_audit_event(
|
|
db,
|
|
actor_type="system",
|
|
actor_label="Atomic reset test",
|
|
action="reset_atomicity_probe",
|
|
entity_type="system",
|
|
)
|
|
db.commit()
|
|
probe_id = probe.id
|
|
|
|
def fail_integrity_check(_db):
|
|
raise RuntimeError("Injected integrity-check failure")
|
|
|
|
monkeypatch.setattr(demo_router, "scenario_integrity_report", fail_integrity_check)
|
|
with pytest.raises(RuntimeError, match="Injected integrity-check failure"):
|
|
ops_client.post("/api/v1/demo/reset")
|
|
|
|
with SessionLocal() as db:
|
|
assert db.scalar(select(AuditEvent).where(AuditEvent.id == probe_id)) is not None
|
|
db.execute(delete(AuditEvent).where(AuditEvent.id == probe_id))
|
|
db.commit()
|
|
|
|
|
|
def test_reset_is_rejected_when_demo_allow_reset_is_disabled(ops_client, monkeypatch):
|
|
import app.api.routers.demo as demo_router
|
|
|
|
monkeypatch.setattr(demo_router.settings, "demo_allow_reset", False)
|
|
response = ops_client.post("/api/v1/demo/reset")
|
|
assert response.status_code == 403
|
|
|
|
# Restore real demo data: this test intentionally disabled reset, so a following test
|
|
# module must not inherit a database left mid-mutation by an earlier test.
|
|
monkeypatch.setattr(demo_router.settings, "demo_allow_reset", True)
|
|
assert ops_client.post("/api/v1/demo/reset").status_code == 200
|
|
|
|
|
|
def test_session_endpoint_requires_authentication(client):
|
|
response = client.get("/api/v1/demo/session")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_session_endpoint_confirms_logged_in_user(ops_client):
|
|
response = ops_client.get("/api/v1/demo/session")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["role"] == "operations_manager"
|
|
assert body["public_ref"] == "USR-OPS"
|
|
|
|
|
|
def test_logout_invalidates_session(ops_client):
|
|
confirmed = ops_client.get("/api/v1/demo/session")
|
|
assert confirmed.status_code == 200
|
|
|
|
logout = ops_client.post("/api/v1/demo/logout")
|
|
assert logout.status_code == 200
|
|
|
|
after = ops_client.get("/api/v1/demo/session")
|
|
assert after.status_code == 401
|
|
|
|
|
|
def test_logout_rejects_a_cookie_even_if_the_browser_retains_it(ops_client):
|
|
from app.core.config import get_settings
|
|
|
|
cookie_name = get_settings().session_cookie_name
|
|
stolen_token = ops_client.cookies.get(cookie_name)
|
|
assert stolen_token
|
|
|
|
logout = ops_client.post("/api/v1/auth/logout")
|
|
assert logout.status_code == 200
|
|
|
|
# Simulate the browser cookie race (or a copied cookie): server-side revocation is
|
|
# authoritative and must reject the original signed token independently of deletion.
|
|
ops_client.cookies.set(cookie_name, stolen_token)
|
|
assert ops_client.get("/api/v1/auth/session").status_code == 401
|
|
|
|
# Remove the deliberately injected hostless cookie before exercising a normal browser
|
|
# login. Otherwise httpx sends it alongside the real testserver cookie, which is not a
|
|
# state a browser can create for the same origin/path pair.
|
|
ops_client.cookies.clear()
|
|
|
|
# A fresh login in the same second receives a distinct signed token and remains valid.
|
|
fresh_login = ops_client.post("/api/v1/demo/login", json={"role": "operations_manager"})
|
|
assert fresh_login.status_code == 200
|
|
assert ops_client.get("/api/v1/auth/session").status_code == 200
|
|
|
|
|
|
def test_logout_without_a_session_is_safe(client):
|
|
response = client.post("/api/v1/demo/logout")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_demo_reset_preserves_integration_telemetry(ops_client, client):
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
probe = client.get(
|
|
"/api/v1/integrations/mcp/operations-summary",
|
|
headers={
|
|
"X-Service-Token": settings.mcp_hub_service_token,
|
|
"X-Client-Id": "itworx-mcp-hub:mobilityops:reset-probe",
|
|
},
|
|
)
|
|
assert probe.status_code == 200
|
|
assert ops_client.post("/api/v1/demo/reset").status_code == 200
|
|
|
|
assert client.post("/api/v1/demo/login", json={"role": "operations_manager"}).status_code == 200
|
|
events = client.get("/api/v1/audit", params={"action": "mcp_tool_request"}).json()
|
|
assert any(
|
|
event["actor_label"] == "itworx-mcp-hub"
|
|
and event["metadata"].get("reported_client_id", "").endswith(":reset-probe")
|
|
for event in events
|
|
)
|