core-api/tests/test_callers.py

"""
Tests for require_caller (T-1).

Builds a tiny standalone FastAPI app rather than reusing src.main's real
routes, so these are pure caller-resolution tests with no database and no
coupling to a particular domain's routing.
"""
import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from unittest.mock import patch

from src.domains.auth.callers import Caller, require_caller
from src.domains.auth.oidc import oidc_config
from src.shared.config import get_settings

settings = get_settings()


def _reset_service_keys():
    settings.office_service_key = ""
    settings.tatlock_service_key = ""
    settings.scheduler_service_key = ""


@pytest.fixture(autouse=True)
def clean_caller_state():
    """Service keys and OIDC state are process-global; isolate every test."""
    original_keys = (
        settings.office_service_key,
        settings.tatlock_service_key,
        settings.scheduler_service_key,
    )
    original_oidc = (oidc_config.enabled, oidc_config.issuers, oidc_config.audiences)
    _reset_service_keys()
    yield
    (
        settings.office_service_key,
        settings.tatlock_service_key,
        settings.scheduler_service_key,
    ) = original_keys
    oidc_config.enabled, oidc_config.issuers, oidc_config.audiences = original_oidc


# Named with a leading underscore so pytest does not try to collect it as
# a test module attribute (it previously warned: "cannot collect 'test_app'
# because it is not a function").
_caller_app = FastAPI()


@_caller_app.get("/only-office")
async def only_office(caller: Caller = Depends(require_caller("office"))):
    return {"kind": caller.kind, "identity": caller.identity, "client": caller.client}


@_caller_app.get("/office-tatlock-user")
async def office_tatlock_user(
    caller: Caller = Depends(require_caller("office", "tatlock", "user")),
):
    return {"kind": caller.kind, "identity": caller.identity}


@_caller_app.post("/office-user-write")
async def office_user_write(
    caller: Caller = Depends(require_caller("office", "user", identity_required=True)),
):
    return {"kind": caller.kind, "identity": caller.identity}


@pytest.fixture
def client():
    return TestClient(_caller_app)


class TestNoAuth:
    def test_no_token_defaults_to_local_user_when_oidc_disabled(self, client):
        """Matches existing housekeeping/dashboard tests: OIDC off -> local user."""
        response = client.get("/office-tatlock-user")
        assert response.status_code == 200
        assert response.json() == {"kind": "user", "identity": "local"}

    def test_no_token_401_when_oidc_enabled(self, client):
        oidc_config.configure(enabled=True, issuers=[], audiences=[])
        response = client.get("/office-tatlock-user")
        assert response.status_code == 401


class TestOfficeKey:
    def test_office_key_with_identity_resolves(self, client):
        settings.office_service_key = "office-secret"
        response = client.get(
            "/office-tatlock-user",
            headers={"Authorization": "Bearer office-secret", "X-Office-Identity": "jeroen"},
        )
        assert response.status_code == 200
        assert response.json() == {"kind": "office", "identity": "jeroen"}

    def test_office_key_client_header_is_informational(self, client):
        settings.office_service_key = "office-secret"
        response = client.get(
            "/only-office",
            headers={
                "Authorization": "Bearer office-secret",
                "X-Office-Identity": "jeroen",
                "X-Office-Client": "claude-desktop",
            },
        )
        assert response.status_code == 200
        assert response.json()["client"] == "claude-desktop"

    def test_office_key_missing_identity_on_read_defaults_empty(self, client):
        settings.office_service_key = "office-secret"
        response = client.get(
            "/office-tatlock-user", headers={"Authorization": "Bearer office-secret"}
        )
        assert response.status_code == 200
        assert response.json() == {"kind": "office", "identity": ""}

    def test_office_key_missing_identity_on_write_is_400(self, client):
        settings.office_service_key = "office-secret"
        response = client.post(
            "/office-user-write", headers={"Authorization": "Bearer office-secret"}
        )
        assert response.status_code == 400
        assert response.json()["detail"]["code"] == "identity_required"

    @pytest.mark.parametrize("identity", ["Jeroen", "group-home", "tatlock", "office", "book", "scheduler", "-abc", "a_b"])
    def test_office_key_malformed_or_reserved_identity_is_400(self, client, identity):
        settings.office_service_key = "office-secret"
        response = client.post(
            "/office-user-write",
            headers={"Authorization": "Bearer office-secret", "X-Office-Identity": identity},
        )
        assert response.status_code == 400
        assert response.json()["detail"]["code"] == "identity_invalid"

    def test_office_key_valid_identity_on_write_succeeds(self, client):
        settings.office_service_key = "office-secret"
        response = client.post(
            "/office-user-write",
            headers={"Authorization": "Bearer office-secret", "X-Office-Identity": "jeroen"},
        )
        assert response.status_code == 200
        assert response.json() == {"kind": "office", "identity": "jeroen"}


class TestTatlockAndSchedulerKeys:
    def test_tatlock_key_resolves_and_ignores_identity_header(self, client):
        settings.tatlock_service_key = "tatlock-secret"
        response = client.get(
            "/office-tatlock-user",
            headers={"Authorization": "Bearer tatlock-secret", "X-Office-Identity": "jeroen"},
        )
        assert response.status_code == 200
        assert response.json() == {"kind": "tatlock", "identity": "tatlock"}

    def test_scheduler_key_forbidden_on_a_route_that_excludes_it(self, client):
        settings.scheduler_service_key = "scheduler-secret"
        response = client.get(
            "/office-tatlock-user", headers={"Authorization": "Bearer scheduler-secret"}
        )
        assert response.status_code == 403
        assert response.json()["detail"]["code"] == "forbidden"


class TestServiceKeyBypassesJWT:
    def test_matched_service_key_never_calls_get_current_user(self, client):
        """Static keys are checked before any JWT parsing (dossier T-1 §1)."""
        settings.office_service_key = "office-secret"
        with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
            mock_get_current_user.side_effect = AssertionError(
                "get_current_user must not run when a service key matched"
            )
            response = client.get(
                "/office-tatlock-user",
                headers={"Authorization": "Bearer office-secret", "X-Office-Identity": "jeroen"},
            )
            assert response.status_code == 200
            mock_get_current_user.assert_not_called()

    def test_unmatched_token_falls_through_to_get_current_user(self, client):
        """A wrong-looking token with no matching key must still reach the JWT path."""
        settings.office_service_key = "office-secret"
        response = client.get(
            "/office-tatlock-user", headers={"Authorization": "Bearer not-the-office-key"}
        )
        # OIDC disabled -> get_current_user ignores the bad bearer and returns the local user.
        assert response.status_code == 200
        assert response.json() == {"kind": "user", "identity": "local"}