generated from coulomb/repo-seed
127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from activity_core import activities
|
|
from activity_core.issue_sink import IssueCoreRestSink
|
|
from activity_core.rules.models import TaskRef, TaskSpec
|
|
|
|
|
|
class DummyResponse:
|
|
def __init__(self, payload: dict[str, Any]) -> None:
|
|
self.payload = payload
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict[str, Any]:
|
|
return self.payload
|
|
|
|
|
|
def test_issue_core_rest_sink_posts_task_contract(monkeypatch) -> None:
|
|
posts: list[dict[str, Any]] = []
|
|
|
|
def fake_post(url: str, **kwargs: Any) -> DummyResponse:
|
|
posts.append({"url": url, **kwargs})
|
|
return DummyResponse({
|
|
"issue_id": "issue-123",
|
|
"issue_url": "http://issue-core.test/issues/issue-123",
|
|
"backend": "issue-core",
|
|
})
|
|
|
|
monkeypatch.setattr(httpx, "post", fake_post)
|
|
|
|
ref = IssueCoreRestSink("http://issue-core.test/").emit(TaskSpec(
|
|
title="Run SBOM rescan for activity-core",
|
|
description="SBOM is older than 30 days.",
|
|
target_repo="activity-core",
|
|
priority="medium",
|
|
labels=["sbom", "security", "automated"],
|
|
due_in_days=7,
|
|
source_type="rule",
|
|
source_id="flag-stale-sbom",
|
|
triggering_event_id="scheduled",
|
|
activity_definition_id="activity-1",
|
|
))
|
|
|
|
assert ref == TaskRef(
|
|
external_id="issue-123",
|
|
backend_url="http://issue-core.test/issues/issue-123",
|
|
backend="issue-core",
|
|
)
|
|
assert posts == [
|
|
{
|
|
"url": "http://issue-core.test/issues/",
|
|
"json": {
|
|
"title": "Run SBOM rescan for activity-core",
|
|
"description": "SBOM is older than 30 days.",
|
|
"target_repo": "activity-core",
|
|
"priority": "medium",
|
|
"labels": ["sbom", "security", "automated"],
|
|
"due_in_days": 7,
|
|
"source_type": "rule",
|
|
"source_id": "flag-stale-sbom",
|
|
"triggering_event_id": "scheduled",
|
|
"activity_definition_id": "activity-1",
|
|
},
|
|
"timeout": 10.0,
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emit_tasks_raises_when_sink_fails(monkeypatch) -> None:
|
|
class FailingSink:
|
|
def emit(self, task_spec: TaskSpec) -> TaskRef:
|
|
raise RuntimeError(f"boom for {task_spec.title}")
|
|
|
|
class FakeTransaction:
|
|
async def __aenter__(self) -> None:
|
|
return None
|
|
|
|
async def __aexit__(self, *exc_info: object) -> bool:
|
|
return False
|
|
|
|
class FakeSession:
|
|
def begin(self) -> FakeTransaction:
|
|
return FakeTransaction()
|
|
|
|
async def __aenter__(self) -> "FakeSession":
|
|
return self
|
|
|
|
async def __aexit__(self, *exc_info: object) -> bool:
|
|
return False
|
|
|
|
def add(self, row: object) -> None:
|
|
raise AssertionError("failed emissions should not write spawn logs")
|
|
|
|
class FakeSessionFactory:
|
|
def __call__(self) -> FakeSession:
|
|
return FakeSession()
|
|
|
|
monkeypatch.setattr(activities, "get_issue_sink", lambda: FailingSink())
|
|
monkeypatch.setattr(activities, "_get_session_factory", lambda: FakeSessionFactory())
|
|
|
|
with pytest.raises(RuntimeError, match="task emission sink failure"):
|
|
await activities.emit_tasks({
|
|
"activity_id": "00000000-0000-0000-0000-000000000001",
|
|
"triggering_event_id": "scheduled",
|
|
"run_id": "00000000-0000-0000-0000-000000000002",
|
|
"task_specs": [
|
|
{
|
|
"title": "Run SBOM rescan for activity-core",
|
|
"description": "",
|
|
"target_repo": "activity-core",
|
|
"priority": "medium",
|
|
"labels": ["sbom"],
|
|
"due_in_days": None,
|
|
"source_type": "rule",
|
|
"source_id": "flag-stale-sbom",
|
|
"condition": "context.repo.sbom_age_days > 30",
|
|
}
|
|
],
|
|
})
|