103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from src.api.v1.platform import operator_runs
|
|
from src.core.awooop_operator_auth import (
|
|
AwoooPOperatorPrincipal,
|
|
authenticate_awooop_operator_headers,
|
|
)
|
|
|
|
|
|
def test_operator_auth_rejects_missing_key_in_prod() -> None:
|
|
with pytest.raises(HTTPException) as exc:
|
|
authenticate_awooop_operator_headers(
|
|
operator_id="ops@example.com",
|
|
operator_key=None,
|
|
configured_key="",
|
|
environment="prod",
|
|
)
|
|
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_operator_auth_rejects_invalid_key() -> None:
|
|
with pytest.raises(HTTPException) as exc:
|
|
authenticate_awooop_operator_headers(
|
|
operator_id="ops@example.com",
|
|
operator_key="wrong",
|
|
configured_key="expected",
|
|
environment="prod",
|
|
)
|
|
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_operator_auth_returns_principal_for_valid_key() -> None:
|
|
principal = authenticate_awooop_operator_headers(
|
|
operator_id="telegram:123456",
|
|
operator_key="expected",
|
|
configured_key="expected",
|
|
environment="prod",
|
|
)
|
|
|
|
assert principal == AwoooPOperatorPrincipal(
|
|
operator_id="telegram:123456",
|
|
auth_method="operator_api_key",
|
|
)
|
|
|
|
|
|
def test_operator_auth_rejects_malformed_identity() -> None:
|
|
with pytest.raises(HTTPException) as exc:
|
|
authenticate_awooop_operator_headers(
|
|
operator_id="../operator",
|
|
operator_key="expected",
|
|
configured_key="expected",
|
|
environment="prod",
|
|
)
|
|
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decide_approval_uses_trusted_principal_not_body(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
async def fake_decide_approval_svc(**kwargs: object) -> dict[str, object]:
|
|
captured.update(kwargs)
|
|
return {
|
|
"run_id": kwargs["run_id"],
|
|
"decision": kwargs["decision"],
|
|
"new_state": "running",
|
|
"approval_token_jti": "jti-test",
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
operator_runs,
|
|
"decide_approval_svc",
|
|
fake_decide_approval_svc,
|
|
)
|
|
|
|
body = operator_runs.DecideApprovalRequest(
|
|
project_id="awoooi",
|
|
decision="approve",
|
|
approver_id="spoofed-browser-value",
|
|
)
|
|
principal = AwoooPOperatorPrincipal(
|
|
operator_id="telegram:123456",
|
|
auth_method="operator_api_key",
|
|
)
|
|
|
|
response = await operator_runs.decide_approval(
|
|
"018f2d04-4c37-7a18-b764-df0df0cbe111",
|
|
body,
|
|
principal,
|
|
)
|
|
|
|
assert response["approval_token_jti"] == "jti-test"
|
|
assert captured["approver_id"] == "telegram:123456"
|
|
assert captured["project_id"] == "awoooi"
|