70 lines
3.2 KiB
Python
70 lines
3.2 KiB
Python
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.api.v1.endpoints import operational_tasks as endpoints
|
|
from app.auth.project_dependencies import ProjectContext
|
|
from app.domain.schemas.operational_data import TaskDataQuery, TaskDataSnapshot
|
|
from app.domain.schemas.operational_task import EvidenceRecord
|
|
from app.main import app
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
def _service_context():
|
|
return ProjectContext(project_id=uuid4(), user_id=uuid4(), project_code="test",
|
|
project_role="member", system_role="service")
|
|
|
|
|
|
def test_dynamic_investigation_routes_are_in_the_public_contract():
|
|
paths = app.openapi()["paths"]
|
|
assert "post" in paths["/api/v1/operational-tasks/{task_id}/access-requests"]
|
|
assert "post" in paths["/api/v1/operational-tasks/{task_id}/access-requests/{access_request_id}/decision"]
|
|
assert "patch" in paths["/api/v1/operational-tasks/{task_id}/actions/{action_id}/draft"]
|
|
assert "post" in paths["/api/v1/operational-tasks/{task_id}/actions/{action_id}/confirm-dispatch"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_data_query_requires_task_grant_before_source_request(monkeypatch):
|
|
service = SimpleNamespace(can_read_source=AsyncMock(return_value=False))
|
|
source = AsyncMock()
|
|
monkeypatch.setattr(endpoints, "_service", lambda _ctx, _conn: service)
|
|
monkeypatch.setattr(endpoints, "read_task_data", source)
|
|
with pytest.raises(HTTPException) as error:
|
|
await endpoints.query_operational_task_data(
|
|
uuid4(), TaskDataQuery(source="scada_devices"), authorization="Bearer AGENT",
|
|
ctx=_service_context(), conn=object(),
|
|
)
|
|
assert error.value.status_code == 403
|
|
source.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_approved_task_source_uses_server_delegation(monkeypatch):
|
|
now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
|
|
snapshot = TaskDataSnapshot(source="scada_devices", source_ref="scada_devices:sha256:test",
|
|
queried_at=now, origin="system_data", data={"items": []}, truncated=False)
|
|
evidence = EvidenceRecord(evidence_id=snapshot.source_ref, kind="system_data",
|
|
summary="SCADA 设备数据快照", source_label="scada_devices",
|
|
source_ref=snapshot.source_ref, observed_at=now, recorded_at=now,
|
|
recorded_by="service", has_snapshot=True)
|
|
service = SimpleNamespace(can_read_source=AsyncMock(return_value=True),
|
|
record_data_snapshot=AsyncMock(return_value=(evidence, snapshot)))
|
|
source = AsyncMock(return_value=snapshot)
|
|
monkeypatch.setattr(endpoints, "_service", lambda _ctx, _conn: service)
|
|
monkeypatch.setattr(endpoints, "read_task_data", source)
|
|
result = await endpoints.query_operational_task_data(
|
|
uuid4(), TaskDataQuery(source="scada_devices"), authorization="Bearer AGENT",
|
|
ctx=_service_context(), conn=object(),
|
|
)
|
|
assert source.await_args.kwargs["delegated"] is True
|
|
assert result.evidence.evidence_id == snapshot.source_ref
|
|
service.record_data_snapshot.assert_awaited_once()
|