activity-core/tests/test_ops_auth.py

"""Unit tests for operator token + SSO auth (ACTIVITY-WP-0024/0025)."""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest
from fastapi import HTTPException

from activity_core.ops_auth import (
    extract_operator_token,
    extract_sso_principal,
    operator_token_configured,
    require_operator,
    verify_operator_token,
)


def test_extract_operator_token_prefers_header() -> None:
    assert (
        extract_operator_token(
            x_operator_token="abc",
            authorization="Bearer other",
        )
        == "abc"
    )


def test_extract_operator_token_bearer() -> None:
    assert extract_operator_token(authorization="Bearer secret-token") == "secret-token"


def test_verify_requires_config_when_fail_closed(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False)
    with pytest.raises(HTTPException) as exc:
        verify_operator_token(None)
    assert exc.value.status_code == 403


def test_verify_allows_anonymous_dev(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
    monkeypatch.setenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", "1")
    assert verify_operator_token(None) == "anonymous-dev"


def test_verify_token_match(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "correct-horse")
    assert verify_operator_token("correct-horse") == "operator-token"
    with pytest.raises(HTTPException) as exc:
        verify_operator_token("wrong")
    assert exc.value.status_code == 401
    with pytest.raises(HTTPException) as exc2:
        verify_operator_token(None)
    assert exc2.value.status_code == 401


def test_operator_token_configured(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
    assert operator_token_configured() is False
    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "x")
    assert operator_token_configured() is True


def test_extract_sso_principal_remote_user() -> None:
    req = MagicMock()
    req.headers = {"Remote-User": "bernd", "Remote-Email": "bernd@example.com"}
    # MagicMock headers.get needs side_effect
    headers = {"Remote-User": "bernd", "Remote-Email": "bernd@example.com"}

    class H(dict):
        def get(self, key, default=None):  # type: ignore[no-untyped-def]
            for k, v in self.items():
                if k.lower() == str(key).lower():
                    return v
            return default

    req.headers = H(headers)
    assert extract_sso_principal(req) == "bernd"


@pytest.mark.asyncio
async def test_require_operator_prefers_sso(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "tok")

    class H(dict):
        def get(self, key, default=None):  # type: ignore[no-untyped-def]
            for k, v in self.items():
                if k.lower() == str(key).lower():
                    return v
            return default

    req = MagicMock()
    req.headers = H({"Remote-User": "alice"})
    principal = await require_operator(req, x_operator_token=None, authorization=None)
    assert principal == "sso:alice"