tegwick
15eb3a2066
Add durable claimable ops_runs table, emit dual-write on TaskSpec, REST claim/lease/complete/fail API, ops status visibility, and consumer docs aligned with ACT-ADR-005. T07 railiance rollout remains deploy-side.
223 lines
6.7 KiB
Python
223 lines
6.7 KiB
Python
"""Tests for ops_run claim queue (ACTIVITY-WP-0026)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from activity_core.ops_run_queue import (
|
|
build_idempotency_key,
|
|
max_attempts,
|
|
ops_run_queue_enabled,
|
|
ops_run_to_dict,
|
|
)
|
|
from activity_core.rules.models import TaskSpec
|
|
|
|
|
|
def test_ops_run_queue_enabled_default(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("OPS_RUN_QUEUE_ENABLED", raising=False)
|
|
assert ops_run_queue_enabled() is True
|
|
monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "false")
|
|
assert ops_run_queue_enabled() is False
|
|
monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "1")
|
|
assert ops_run_queue_enabled() is True
|
|
|
|
|
|
def test_build_idempotency_key() -> None:
|
|
spec = TaskSpec(
|
|
title="t",
|
|
activity_definition_id="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
|
|
source_id="emit-fi",
|
|
triggering_event_id="wf-1",
|
|
)
|
|
assert build_idempotency_key(spec) == (
|
|
"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee:emit-fi:wf-1"
|
|
)
|
|
|
|
|
|
def test_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("OPS_RUN_MAX_ATTEMPTS", raising=False)
|
|
assert max_attempts() == 3
|
|
monkeypatch.setenv("OPS_RUN_MAX_ATTEMPTS", "5")
|
|
assert max_attempts() == 5
|
|
|
|
|
|
def test_ops_run_to_dict_shape() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
row = MagicMock()
|
|
row.id = uuid.uuid4()
|
|
row.activity_definition_id = uuid.uuid4()
|
|
row.idempotency_key = "k"
|
|
row.target_repo = "freedom-intelligence"
|
|
row.title = "FI brief"
|
|
row.description = "d"
|
|
row.labels = ["automated", "research-brief"]
|
|
row.priority = "medium"
|
|
row.state = "open"
|
|
row.claim_owner = None
|
|
row.lease_until = None
|
|
row.attempt = 0
|
|
row.source_type = "rule"
|
|
row.source_id = "emit"
|
|
row.triggering_event_id = "e1"
|
|
row.approach_hint = None
|
|
row.result = {}
|
|
row.created_at = now
|
|
row.updated_at = now
|
|
d = ops_run_to_dict(row)
|
|
assert d["target_repo"] == "freedom-intelligence"
|
|
assert d["state"] == "open"
|
|
assert "research-brief" in d["labels"]
|
|
assert d["created_at"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ops_run_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from activity_core.ops_run_queue import create_ops_run_from_spec
|
|
|
|
monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "false")
|
|
session = AsyncMock()
|
|
spec = TaskSpec(
|
|
title="t",
|
|
activity_definition_id=str(uuid.uuid4()),
|
|
source_id="s",
|
|
triggering_event_id="e",
|
|
)
|
|
assert await create_ops_run_from_spec(session, spec) is None
|
|
session.execute.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_ops_run_inserts(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from activity_core.ops_run_queue import create_ops_run_from_spec
|
|
|
|
monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "true")
|
|
run_id = uuid.uuid4()
|
|
result = MagicMock()
|
|
result.scalar_one_or_none.return_value = run_id
|
|
session = AsyncMock()
|
|
session.execute = AsyncMock(return_value=result)
|
|
spec = TaskSpec(
|
|
title="FI daily",
|
|
target_repo="freedom-intelligence",
|
|
labels=["automated", "research-brief"],
|
|
activity_definition_id=str(uuid.uuid4()),
|
|
source_id="emit-fi-daily-brief-task",
|
|
triggering_event_id="manual-1",
|
|
)
|
|
got = await create_ops_run_from_spec(session, spec)
|
|
assert got == run_id
|
|
session.execute.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_and_complete_roundtrip(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""In-memory style: claim filters labels and complete transitions state."""
|
|
from activity_core import ops_run_queue as oq
|
|
|
|
now = datetime.now(timezone.utc)
|
|
open_run = MagicMock()
|
|
open_run.id = uuid.uuid4()
|
|
open_run.state = "open"
|
|
open_run.labels = ["automated", "research-brief"]
|
|
open_run.attempt = 0
|
|
open_run.claim_owner = None
|
|
open_run.lease_until = None
|
|
open_run.result = {}
|
|
|
|
session = AsyncMock()
|
|
|
|
reopen_result = MagicMock()
|
|
reopen_result.rowcount = 0
|
|
|
|
id_result = MagicMock()
|
|
id_result.scalars.return_value.all.return_value = [open_run.id]
|
|
|
|
async def execute_side_effect(stmt, *args, **kwargs):
|
|
# reopen_stale_claims uses Update; claim select uses Select (may contain FOR UPDATE)
|
|
name = type(stmt).__name__.lower()
|
|
if name == "update":
|
|
return reopen_result
|
|
return id_result
|
|
|
|
session.execute = AsyncMock(side_effect=execute_side_effect)
|
|
session.get = AsyncMock(return_value=open_run)
|
|
|
|
claimed = await oq.claim_ops_runs(
|
|
session,
|
|
worker_id="worker-1",
|
|
labels=["research-brief"],
|
|
limit=1,
|
|
lease_seconds=60,
|
|
)
|
|
assert len(claimed) == 1
|
|
assert open_run.state == "claimed"
|
|
assert open_run.claim_owner == "worker-1"
|
|
assert open_run.attempt == 1
|
|
assert open_run.lease_until is not None
|
|
assert open_run.lease_until > now
|
|
|
|
done = await oq.complete_ops_run(
|
|
session,
|
|
open_run.id,
|
|
worker_id="worker-1",
|
|
result={"path": "briefs/x.md"},
|
|
)
|
|
assert done is not None
|
|
assert open_run.state == "succeeded"
|
|
assert open_run.result["path"] == "briefs/x.md"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fail_reopen_under_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from activity_core import ops_run_queue as oq
|
|
|
|
monkeypatch.setenv("OPS_RUN_MAX_ATTEMPTS", "3")
|
|
row = MagicMock()
|
|
row.id = uuid.uuid4()
|
|
row.state = "claimed"
|
|
row.claim_owner = "w1"
|
|
row.attempt = 1
|
|
row.result = {}
|
|
session = AsyncMock()
|
|
session.get = AsyncMock(return_value=row)
|
|
|
|
out = await oq.fail_ops_run(
|
|
session, row.id, worker_id="w1", error="timeout", reopen=True
|
|
)
|
|
assert out is not None
|
|
assert row.state == "open"
|
|
assert row.claim_owner is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fail_permanent_at_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from activity_core import ops_run_queue as oq
|
|
|
|
monkeypatch.setenv("OPS_RUN_MAX_ATTEMPTS", "3")
|
|
row = MagicMock()
|
|
row.id = uuid.uuid4()
|
|
row.state = "claimed"
|
|
row.claim_owner = "w1"
|
|
row.attempt = 3
|
|
row.result = {}
|
|
session = AsyncMock()
|
|
session.get = AsyncMock(return_value=row)
|
|
|
|
out = await oq.fail_ops_run(
|
|
session, row.id, worker_id="w1", error="timeout", reopen=True
|
|
)
|
|
assert out is not None
|
|
assert row.state == "failed"
|
|
|
|
|
|
def test_label_filter_any_vs_all() -> None:
|
|
"""Document labels_mode semantics used by claim_ops_runs."""
|
|
row_labels = {"automated", "research-brief"}
|
|
want = {"research-brief", "missing"}
|
|
assert bool(want & row_labels) # any
|
|
assert not want.issubset(row_labels) # all
|