tegwick
91353df7d0
Mark T03–T05 and T07–T08 done after live Authelia/TLS verification, prefer SSO principal in ops UI copy and audits, and document break-glass port-forward. Leave T06 waiting on net-kingdom LLDAP/Authelia group rules.
299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""Contract tests for /ops console API (ACTIVITY-WP-0024)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from activity_core import ops_api
|
|
from activity_core.ops_console import clear_audit_buffer_for_tests, recent_audits, record_ops_audit
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_audit() -> None:
|
|
clear_audit_buffer_for_tests()
|
|
yield
|
|
clear_audit_buffer_for_tests()
|
|
|
|
|
|
@pytest.fixture
|
|
def ops_app(monkeypatch: pytest.MonkeyPatch) -> FastAPI:
|
|
monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "test-token")
|
|
monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False)
|
|
monkeypatch.setenv("ACTCORE_DB_URL", "postgresql+asyncpg://unused/unused")
|
|
|
|
app = FastAPI()
|
|
app.include_router(ops_api.router)
|
|
|
|
session_factory = MagicMock(name="session_factory")
|
|
temporal = MagicMock(name="temporal")
|
|
ops_api.bind_ops_deps(lambda: session_factory, lambda: temporal)
|
|
app.state.session_factory = session_factory
|
|
app.state.temporal = temporal
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_status(ops_app: FastAPI) -> None:
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.get("/ops/auth/status")
|
|
assert res.status_code == 200
|
|
body = res.json()
|
|
assert body["operator_token_configured"] is True
|
|
assert body["mutation_header"] == "X-Operator-Token"
|
|
assert "temporal_ui_url" in body
|
|
assert body["temporal_ui_url"].startswith("http")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_requires_token(ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def_id = uuid.uuid4()
|
|
|
|
async def fake_status(*_a: Any, **_k: Any) -> dict[str, Any]:
|
|
return {"mode": "automation-status", "activities": []}
|
|
|
|
monkeypatch.setattr(ops_api, "ops_status", fake_status)
|
|
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.post(f"/ops/automations/{def_id}/trigger")
|
|
assert res.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_with_sso_principal(
|
|
ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""SSO Remote-User is preferred; audit must record sso:<user>, never the token."""
|
|
def_id = uuid.uuid4()
|
|
row = MagicMock()
|
|
row.name = "Weekly SBOM"
|
|
row.context_sources = []
|
|
row.task_templates = []
|
|
row.trigger_config = {"trigger_type": "cron", "cron_expression": "0 9 * * 1"}
|
|
|
|
session = AsyncMock()
|
|
session.get = AsyncMock(return_value=row)
|
|
session.__aenter__ = AsyncMock(return_value=session)
|
|
session.__aexit__ = AsyncMock(return_value=None)
|
|
ops_app.state.session_factory.return_value = session
|
|
|
|
handle = MagicMock()
|
|
handle.id = f"activity-{def_id}:manual-sso"
|
|
ops_app.state.temporal.start_workflow = AsyncMock(return_value=handle)
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def capture_audit(**kwargs: Any) -> dict[str, Any]:
|
|
captured.update(kwargs)
|
|
return {
|
|
"action": kwargs["action"],
|
|
"audit_id": "sso-a1",
|
|
"principal": kwargs["principal"],
|
|
}
|
|
|
|
monkeypatch.setattr("activity_core.ops_api.record_ops_audit", capture_audit)
|
|
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.post(
|
|
f"/ops/automations/{def_id}/trigger",
|
|
headers={"Remote-User": "alice.operator"},
|
|
json={},
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json()["audit"]["principal"] == "sso:alice.operator"
|
|
assert captured["principal"] == "sso:alice.operator"
|
|
assert "token" not in str(res.json()).lower() or "test-token" not in str(res.json())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_with_token(ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def_id = uuid.uuid4()
|
|
row = MagicMock()
|
|
row.name = "Weekly SBOM"
|
|
row.context_sources = []
|
|
row.task_templates = []
|
|
row.trigger_config = {"trigger_type": "cron", "cron_expression": "0 9 * * 1"}
|
|
|
|
session = AsyncMock()
|
|
session.get = AsyncMock(return_value=row)
|
|
session.__aenter__ = AsyncMock(return_value=session)
|
|
session.__aexit__ = AsyncMock(return_value=None)
|
|
ops_app.state.session_factory.return_value = session
|
|
|
|
handle = MagicMock()
|
|
handle.id = f"activity-{def_id}:manual-test"
|
|
ops_app.state.temporal.start_workflow = AsyncMock(return_value=handle)
|
|
|
|
monkeypatch.setattr(
|
|
"activity_core.ops_api.record_ops_audit",
|
|
AsyncMock(
|
|
return_value={
|
|
"action": "trigger",
|
|
"audit_id": "a1",
|
|
"principal": "operator",
|
|
}
|
|
),
|
|
)
|
|
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.post(
|
|
f"/ops/automations/{def_id}/trigger",
|
|
headers={"X-Operator-Token": "test-token"},
|
|
json={},
|
|
)
|
|
assert res.status_code == 200
|
|
body = res.json()
|
|
assert body["workflow_id"] == handle.id
|
|
assert body["name"] == "Weekly SBOM"
|
|
assert "token" not in str(body).lower() or "test-token" not in str(body)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_side_effect_requires_confirm(
|
|
ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def_id = uuid.uuid4()
|
|
row = MagicMock()
|
|
row.name = "Weekly Forgejo Package Prune"
|
|
row.context_sources = [{"query": "forgejo_package_prune", "apply": True}]
|
|
row.task_templates = []
|
|
row.trigger_config = {}
|
|
|
|
session = AsyncMock()
|
|
session.get = AsyncMock(return_value=row)
|
|
session.__aenter__ = AsyncMock(return_value=session)
|
|
session.__aexit__ = AsyncMock(return_value=None)
|
|
ops_app.state.session_factory.return_value = session
|
|
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.post(
|
|
f"/ops/automations/{def_id}/trigger",
|
|
headers={"X-Operator-Token": "test-token"},
|
|
json={},
|
|
)
|
|
assert res.status_code == 400
|
|
assert "side effect" in res.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_endpoint_wraps_report(
|
|
ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
async def fake_status(**kwargs: Any) -> dict[str, Any]:
|
|
return {
|
|
"mode": "automation-status",
|
|
"window": {"since": kwargs.get("since")},
|
|
"summary": {"total": 0},
|
|
"activities": [],
|
|
"warnings": [],
|
|
"exit_code": 0,
|
|
}
|
|
|
|
monkeypatch.setattr(ops_api, "ops_status", fake_status)
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.get("/ops/automations/status", params={"since": "sunday"})
|
|
assert res.status_code == 200
|
|
assert res.json()["mode"] == "automation-status"
|
|
assert res.json()["window"]["since"] == "sunday"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inventory_endpoint(
|
|
ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
async def fake_inv(**_k: Any) -> dict[str, Any]:
|
|
return {
|
|
"mode": "automation-inventory",
|
|
"automations": [
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"name": "Daily Triage",
|
|
"enabled": True,
|
|
"trigger_type": "cron",
|
|
"cron_expression": "20 7 * * *",
|
|
}
|
|
],
|
|
"summary": {"total": 1},
|
|
"warnings": [],
|
|
}
|
|
|
|
monkeypatch.setattr(ops_api, "ops_inventory", fake_inv)
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.get("/ops/automations")
|
|
assert res.status_code == 200
|
|
assert res.json()["summary"]["total"] == 1
|
|
assert res.json()["automations"][0]["name"] == "Daily Triage"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ui_index_renders(ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
async def fake_inv(**_k: Any) -> dict[str, Any]:
|
|
return {
|
|
"automations": [
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"name": "Daily Triage",
|
|
"enabled": True,
|
|
"trigger_type": "cron",
|
|
"cron_expression": "0 0 * * *",
|
|
"timezone": "Europe/Berlin",
|
|
"temporal": {"paused": False},
|
|
}
|
|
],
|
|
"summary": {"total": 1},
|
|
"warnings": [],
|
|
}
|
|
|
|
monkeypatch.setattr(ops_api, "ops_inventory", fake_inv)
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.get("/ops/ui/")
|
|
assert res.status_code == 200
|
|
assert "Daily Triage" in res.text
|
|
assert "SSO" in res.text or "Break-glass" in res.text
|
|
assert "Temporal UI" in res.text
|
|
assert "temporal.coulomb.social" in res.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_ops_audit_no_secret_leak(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("STATE_HUB_URL", raising=False)
|
|
event = await record_ops_audit(
|
|
action="trigger",
|
|
activity_id=str(uuid.uuid4()),
|
|
activity_name="X",
|
|
principal="operator",
|
|
detail={"workflow_id": "wf-1"},
|
|
)
|
|
assert event["action"] == "trigger"
|
|
assert "token" not in event
|
|
audits = recent_audits()
|
|
assert audits[0]["audit_id"] == event["audit_id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runs_404(ops_app: FastAPI) -> None:
|
|
def_id = uuid.uuid4()
|
|
session = AsyncMock()
|
|
session.get = AsyncMock(return_value=None)
|
|
session.__aenter__ = AsyncMock(return_value=session)
|
|
session.__aexit__ = AsyncMock(return_value=None)
|
|
ops_app.state.session_factory.return_value = session
|
|
|
|
transport = ASGITransport(app=ops_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.get(f"/ops/automations/{def_id}/runs")
|
|
assert res.status_code == 404
|