tegwick
71027f0a67
Add /ops REST inventory, status, runs, and fail-closed operator-token mutations (trigger, enable/disable, pause/unpause) with audit trail. Ship thin HTML UI at /ops/ui, runbook/k8s access docs, and contract tests.
58 lines
2 KiB
Python
58 lines
2 KiB
Python
"""Unit tests for operator token auth (ACTIVITY-WP-0024-T02/T06)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from activity_core.ops_auth import (
|
|
extract_operator_token,
|
|
operator_token_configured,
|
|
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"
|
|
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
|