praxis/tests/test_auth.py

"""Auth unit tests (TASK-03-06) — mocked PgStore, no real Postgres.

Covers: password hash/verify/rehash, cookie config (secure flag, missing
secret), rate limiter threshold, current_operator dependency (401/503
cases, active/inactive), login/logout/me route handlers.
"""

from __future__ import annotations

import os
import types
from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from starlette.middleware.sessions import SessionMiddleware

from server.auth.cookies import get_session_middleware_kwargs
from server.auth.dependencies import current_operator
from server.auth.models import Operator
from server.auth.passwords import hash_password, needs_rehash, verify_password
from server.auth.rate_limit import limiter, rate_limit_login, reset_login_rate_limit
from server.auth.routes import router


@pytest.fixture(autouse=True)
def _reset_limiter():
    reset_login_rate_limit()
    yield
    reset_login_rate_limit()


# ── Passwords ───────────────────────────────────────────────────────────────


def test_password_hash_verify_roundtrip():
    h = hash_password("correct horse battery staple")
    assert h.startswith("$argon2id$")
    assert verify_password(h, "correct horse battery staple") is True


def test_password_verify_wrong_returns_false():
    h = hash_password("secret-1")
    assert verify_password(h, "secret-2") is False
    # no exception raised — uniform 401 path
    assert verify_password(h, "") is False


def test_needs_rehash_false_for_current_defaults():
    h = hash_password("x")
    assert needs_rehash(h) is False


def test_needs_rehash_true_for_weak_hash():
    # A hash produced with weaker params triggers rehash.
    from argon2 import PasswordHasher
    weak = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1).hash("x")
    assert needs_rehash(weak) is True


# ── Cookie config ───────────────────────────────────────────────────────────


def test_cookie_kwargs_defaults(monkeypatch):
    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "x" * 48)
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "true")
    kw = get_session_middleware_kwargs()
    assert kw["session_cookie"] == "praxis_op"
    assert kw["max_age"] == 28800
    assert kw["httponly"] is True
    assert kw["samesite"] == "strict"
    assert kw["secure"] is True
    assert kw["path"] == "/"


def test_cookie_secure_false(monkeypatch):
    monkeypatch.setenv("PRAXIS_COOKIE_SECRET", "x" * 48)
    monkeypatch.setenv("PRAXIS_COOKIE_SECURE", "false")
    kw = get_session_middleware_kwargs()
    assert kw["secure"] is False


def test_cookie_secret_unset_generates_random(monkeypatch):
    monkeypatch.delenv("PRAXIS_COOKIE_SECRET", raising=False)
    kw = get_session_middleware_kwargs()
    assert kw["secret_key"]
    assert len(kw["secret_key"]) >= 32


# ── current_operator dependency ─────────────────────────────────────────────


def _make_app_with_store(store) -> FastAPI:
    app = FastAPI()
    app.state.pg_store = store
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
    app.include_router(router)
    return app


def _mock_store(operator_row=None):
    store = MagicMock()
    store.get_operator_by_id = AsyncMock(return_value=operator_row)
    return store


def test_current_operator_no_cookie_401():
    app = _make_app_with_store(_mock_store(operator_row=None))
    with TestClient(app) as client:
        r = client.get("/api/operator/me")
        assert r.status_code == 401


def test_current_operator_no_postgres_503():
    app = FastAPI()
    app.state.pg_store = None
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
    app.include_router(router)
    with TestClient(app) as client:
        r = client.get("/api/operator/me")
        assert r.status_code == 503


def test_current_operator_inactive_401():
    op = {
        "id": "11111111-1111-1111-1111-111111111111",
        "username": "ghost",
        "display_name": "Ghost",
        "role": "operator",
        "is_active": False,
    }
    store = _mock_store(operator_row=op)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        # seed a session by hitting login would need a real store; instead
        # set the session directly via a cookie. Use TestClient's cookie jar.
        # Easiest: POST /login with a mocked store that returns the op.
        store.get_operator_by_username = AsyncMock(return_value=op)
        store.update_last_login = AsyncMock()
        store.pool = MagicMock()
        conn = MagicMock()
        conn.execute = AsyncMock()
        cm = MagicMock()
        cm.__aenter__ = AsyncMock(return_value=conn)
        cm.__aexit__ = AsyncMock(return_value=None)
        store.pool.acquire = MagicMock(return_value=cm)
        # hash the password so verify works
        op = dict(op)
        op["password_hash"] = hash_password("pw")
        store.get_operator_by_username = AsyncMock(return_value=op)
        r = client.post("/api/operator/login", json={"username": "ghost", "password": "pw"})
        # inactive operator → 401 even with correct password
        assert r.status_code == 401


def test_current_operator_valid_cookie_returns_operator():
    op = {
        "id": "22222222-2222-2222-2222-222222222222",
        "username": "alice",
        "display_name": "Alice",
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "alice", "password": "pw"})
        assert r.status_code == 200
        assert r.json()["operator"]["username"] == "alice"
        # cookie is now set; /me should work
        r2 = client.get("/api/operator/me")
        assert r2.status_code == 200
        assert r2.json()["operator"]["username"] == "alice"


# ── Login route ─────────────────────────────────────────────────────────────


def test_login_wrong_password_401_no_cookie():
    op = {
        "id": "33333333-3333-3333-3333-333333333333",
        "username": "bob",
        "display_name": None,
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("correct"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "bob", "password": "wrong"})
        assert r.status_code == 401
        # no auth cookie set on failure
        cookies = client.cookies.get("praxis_op")
        assert not cookies


def test_login_unknown_user_401():
    store = _mock_store(operator_row=None)
    store.get_operator_by_username = AsyncMock(return_value=None)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "nobody", "password": "x"})
        assert r.status_code == 401


def test_login_no_postgres_503():
    app = FastAPI()
    app.state.pg_store = None
    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
    app.include_router(router)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "a", "password": "b"})
        assert r.status_code == 503


# ── Logout ──────────────────────────────────────────────────────────────────


def test_logout_clears_session():
    op = {
        "id": "44444444-4444-4444-4444-444444444444",
        "username": "carol",
        "display_name": "Carol",
        "role": "operator",
        "is_active": True,
        "password_hash": hash_password("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    conn = MagicMock()
    conn.execute = AsyncMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        client.post("/api/operator/login", json={"username": "carol", "password": "pw"})
        r = client.post("/api/operator/logout")
        assert r.status_code == 200
        assert r.json()["ok"] is True
        # /me now 401
        r2 = client.get("/api/operator/me")
        assert r2.status_code == 401


# ── Rehash on login ─────────────────────────────────────────────────────────


def test_login_rehash_when_needed():
    from argon2 import PasswordHasher
    weak_hasher = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1)
    op = {
        "id": "55555555-5555-5555-5555-555555555555",
        "username": "dave",
        "display_name": "Dave",
        "role": "operator",
        "is_active": True,
        "password_hash": weak_hasher.hash("pw"),
    }
    store = _mock_store(operator_row=op)
    store.get_operator_by_username = AsyncMock(return_value=op)
    store.update_last_login = AsyncMock()
    store.pool = MagicMock()
    executed = []
    conn = MagicMock()
    async def _exec(*a, **kw):
        executed.append(a)
    conn.execute = _exec
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)
    app = _make_app_with_store(store)
    with TestClient(app) as client:
        r = client.post("/api/operator/login", json={"username": "dave", "password": "pw"})
        assert r.status_code == 200
        assert executed, "rehash UPDATE should have run"
        # the second arg to execute is the new hash; verify it's argon2id
        assert executed[0][1].startswith("$argon2id$")


# ── Rate limiter ────────────────────────────────────────────────────────────


def test_rate_limit_login_decorator():
    # The decorator factory returns a decorator; applying it should not raise.
    deco = rate_limit_login()
    assert callable(deco)


def test_limiter_is_in_memory():
    assert getattr(limiter, "_storage_uri", "memory://") == "memory://" or limiter._storage is not None