tests/test_auth.py

import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.auth.deps import COOKIE_NAME, require_role
from app.auth.jwt import create_token
from app.claimguard import audit
from app.config import Settings
from app.models import AuditEvent, Role, User
from tests.conftest import DEMO_PASSWORD, login


@pytest.fixture()
def agent_only_route(app: FastAPI) -> None:
    @app.get("/api/_test/agent-only")
    def agent_only(
        user: User = Depends(require_role(Role.INSURANCE_AGENT)),
    ) -> dict[str, int]:
        return {"user_id": user.id}


def test_login_success_sets_session_cookie_and_me(
    client: TestClient, users: dict[str, User]
) -> None:
    resp = client.post(
        "/api/auth/login", json={"email": "claimant@demo.ca", "password": DEMO_PASSWORD}
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["role"] == "claimant"
    assert body["email"] == "claimant@demo.ca"
    assert body["id"] == users["claimant"].id

    set_cookie = resp.headers["set-cookie"].lower()
    assert set_cookie.startswith(f"{COOKIE_NAME}=")
    assert "httponly" in set_cookie
    assert "samesite=lax" in set_cookie
    assert "path=/" in set_cookie
    assert COOKIE_NAME in client.cookies

    me = client.get("/api/auth/me")
    assert me.status_code == 200
    assert me.json() == body


def test_login_failures_identical_detail_and_audited(
    client: TestClient, users: dict[str, User], session: Session
) -> None:
    wrong_password = client.post(
        "/api/auth/login", json={"email": "claimant@demo.ca", "password": "wrong-password"}
    )
    unknown_email = client.post(
        "/api/auth/login", json={"email": "nobody@demo.ca", "password": DEMO_PASSWORD}
    )
    assert wrong_password.status_code == 401
    assert unknown_email.status_code == 401
    assert wrong_password.json()["detail"] == unknown_email.json()["detail"]
    assert COOKIE_NAME not in client.cookies

    events = session.scalars(
        select(AuditEvent).where(AuditEvent.event_type == "auth.failed").order_by(AuditEvent.id)
    ).all()
    assert len(events) == 2
    assert all(e.actor_user_id is None for e in events)
    assert "claimant@demo.ca" in events[0].payload_json
    assert "nobody@demo.ca" in events[1].payload_json

    valid, checked = audit.verify_chain(session)
    assert valid is True
    assert checked == 2


def test_login_success_records_auth_login_with_actor(
    client: TestClient, users: dict[str, User], session: Session
) -> None:
    login(client, "agent@demo.ca")

    event = session.scalar(select(AuditEvent).where(AuditEvent.event_type == "auth.login"))
    assert event is not None
    assert event.actor_user_id == users["insurance_agent"].id
    assert event.actor_role == "insurance_agent"

    valid, checked = audit.verify_chain(session)
    assert valid is True
    assert checked == 1


def test_me_without_cookie_is_401(client: TestClient) -> None:
    resp = client.get("/api/auth/me")
    assert resp.status_code == 401
    assert resp.json()["detail"] == "Not authenticated"


def test_expired_token_is_401(
    client: TestClient, users: dict[str, User], settings: Settings
) -> None:
    token = create_token(users["claimant"], settings, expires_hours=-1)
    client.cookies.set(COOKIE_NAME, token)
    resp = client.get("/api/auth/me")
    assert resp.status_code == 401
    assert resp.json()["detail"] == "Invalid or expired session"


def test_tampered_token_is_401(
    client: TestClient, users: dict[str, User], settings: Settings
) -> None:
    forged_settings = settings.model_copy(update={"jwt_secret": "attacker-secret"})
    token = create_token(users["claimant"], forged_settings)
    client.cookies.set(COOKIE_NAME, token)
    resp = client.get("/api/auth/me")
    assert resp.status_code == 401
    assert resp.json()["detail"] == "Invalid or expired session"


def test_logout_clears_cookie(as_claimant: TestClient) -> None:
    assert as_claimant.get("/api/auth/me").status_code == 200

    resp = as_claimant.post("/api/auth/logout")
    assert resp.status_code == 200
    assert resp.json() == {"status": "logged_out"}
    set_cookie = resp.headers["set-cookie"].lower()
    assert set_cookie.startswith(f"{COOKIE_NAME}=")
    assert "max-age=0" in set_cookie or "expires=" in set_cookie
    assert COOKIE_NAME not in as_claimant.cookies

    assert as_claimant.get("/api/auth/me").status_code == 401


def test_login_rejects_cross_origin(client: TestClient, users: dict[str, User]) -> None:
    resp = client.post(
        "/api/auth/login",
        json={"email": "claimant@demo.ca", "password": DEMO_PASSWORD},
        headers={"Origin": "http://evil.example"},
    )
    assert resp.status_code == 403
    assert COOKIE_NAME not in client.cookies


def test_login_allows_matching_origin(
    client: TestClient, users: dict[str, User], settings: Settings
) -> None:
    resp = client.post(
        "/api/auth/login",
        json={"email": "claimant@demo.ca", "password": DEMO_PASSWORD},
        headers={"Origin": settings.app_origin},
    )
    assert resp.status_code == 200


def test_login_allows_missing_origin(client: TestClient, users: dict[str, User]) -> None:
    resp = client.post(
        "/api/auth/login", json={"email": "claimant@demo.ca", "password": DEMO_PASSWORD}
    )
    assert resp.status_code == 200


def test_require_role_allows_agent(
    agent_only_route: None, as_agent: TestClient, users: dict[str, User]
) -> None:
    resp = as_agent.get("/api/_test/agent-only")
    assert resp.status_code == 200
    assert resp.json() == {"user_id": users["insurance_agent"].id}


def test_require_role_rejects_claimant(agent_only_route: None, as_claimant: TestClient) -> None:
    resp = as_claimant.get("/api/_test/agent-only")
    assert resp.status_code == 403
    assert resp.json()["detail"] == "Insufficient role"


def test_require_role_rejects_anonymous(agent_only_route: None, client: TestClient) -> None:
    resp = client.get("/api/_test/agent-only")
    assert resp.status_code == 401