f5c2733cfb
LangGraph-based release automation agent with: - PR discovery (webhook + polling) - AI code review via Claude Code CLI (subscription-based) - Auto-create Jira tickets for PRs without ticket ID - Jira ticket lifecycle management (code review -> staging -> done) - CI/CD pipeline trigger, polling, and approval gates - Slack interactive messages with approval buttons - Per-repo semantic versioning - PostgreSQL persistence (threads, staging, releases) - FastAPI API (webhooks, approvals, status, manual triggers) - Docker Compose deployment 1069 tests, 95%+ coverage.
112 lines
4.1 KiB
Python
112 lines
4.1 KiB
Python
"""Tests for operator token authentication dependency.
|
|
|
|
Phase 5 - Step 3: require_operator_token FastAPI dependency.
|
|
Written FIRST (TDD RED phase).
|
|
"""
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, Depends, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from release_agent.api.dependencies import require_operator_token
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_app_with_token(operator_token: str = "") -> FastAPI:
|
|
"""Return a minimal app with a protected route and the given token config."""
|
|
app = FastAPI()
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.operator_token.get_secret_value.return_value = operator_token
|
|
app.state.settings = mock_settings
|
|
|
|
@app.get("/protected")
|
|
async def protected_route(_: None = Depends(require_operator_token)):
|
|
return {"ok": True}
|
|
|
|
return app
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# require_operator_token tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRequireOperatorToken:
|
|
def test_valid_token_allows_access(self) -> None:
|
|
app = _make_app_with_token("super-secret-token")
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/protected",
|
|
headers={"X-Operator-Token": "super-secret-token"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_missing_token_header_returns_401_when_token_configured(self) -> None:
|
|
app = _make_app_with_token("super-secret-token")
|
|
with TestClient(app) as client:
|
|
response = client.get("/protected")
|
|
assert response.status_code == 401
|
|
|
|
def test_wrong_token_returns_401(self) -> None:
|
|
app = _make_app_with_token("super-secret-token")
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/protected",
|
|
headers={"X-Operator-Token": "wrong-token"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
def test_empty_operator_token_config_skips_auth(self) -> None:
|
|
"""When operator_token is not configured (empty), all requests pass."""
|
|
app = _make_app_with_token("")
|
|
with TestClient(app) as client:
|
|
response = client.get("/protected")
|
|
assert response.status_code == 200
|
|
|
|
def test_empty_operator_token_config_passes_even_without_header(self) -> None:
|
|
app = _make_app_with_token("")
|
|
with TestClient(app) as client:
|
|
response = client.get("/protected", headers={})
|
|
assert response.status_code == 200
|
|
|
|
def test_token_comparison_is_constant_time(self) -> None:
|
|
"""Verify hmac.compare_digest is used (not == operator) — tested by checking
|
|
that the function still works correctly, not timing (which we can't test here)."""
|
|
app = _make_app_with_token("my-token")
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/protected",
|
|
headers={"X-Operator-Token": "my-token"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_empty_string_token_header_rejected_when_token_configured(self) -> None:
|
|
app = _make_app_with_token("configured-token")
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/protected",
|
|
headers={"X-Operator-Token": ""},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
def test_401_response_has_detail_field(self) -> None:
|
|
app = _make_app_with_token("secret")
|
|
with TestClient(app) as client:
|
|
response = client.get("/protected")
|
|
data = response.json()
|
|
assert "detail" in data
|
|
|
|
def test_valid_token_returns_correct_response_body(self) -> None:
|
|
app = _make_app_with_token("token123")
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/protected",
|
|
headers={"X-Operator-Token": "token123"},
|
|
)
|
|
assert response.json() == {"ok": True}
|