tegwick
21dc228cb1
Cron schedules always passed trigger_key="scheduled", so ops_run
idempotency collapsed every weekday into one key. After the first fire,
create_ops_run was a silent no-op, the claim loop starved, and dual-clock
host timers produced empty FI briefs.
Map scheduled fires to run_id (or scheduled:{iso}) via
emit_triggering_event_id; log duplicate skips; document the contract.
276 lines
8.5 KiB
Python
276 lines
8.5 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,
|
|
emit_triggering_event_id,
|
|
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_emit_triggering_event_id_scheduled_uses_run_id() -> None:
|
|
"""Bare 'scheduled' must not be the ops_run key for every weekday fire."""
|
|
day1 = emit_triggering_event_id("scheduled", "run-day-1")
|
|
day2 = emit_triggering_event_id("scheduled", "run-day-2")
|
|
assert day1 == "run-day-1"
|
|
assert day2 == "run-day-2"
|
|
assert day1 != day2
|
|
|
|
|
|
def test_emit_triggering_event_id_prefers_scheduled_for() -> None:
|
|
got = emit_triggering_event_id(
|
|
"scheduled",
|
|
"run-fallback",
|
|
scheduled_for="2026-08-05T05:30:00+00:00",
|
|
)
|
|
assert got == "scheduled:2026-08-05T05:30:00+00:00"
|
|
|
|
|
|
def test_emit_triggering_event_id_passes_through_events_and_manual() -> None:
|
|
assert emit_triggering_event_id("evt-abc", "run-x") == "evt-abc"
|
|
assert (
|
|
emit_triggering_event_id("manual-3da4cf06", "run-y") == "manual-3da4cf06"
|
|
)
|
|
|
|
|
|
def test_scheduled_fires_produce_distinct_ops_keys() -> None:
|
|
"""Regression: two cron days must not share one ops_run idempotency key."""
|
|
def_id = "3169ab1f-882b-59c7-9763-d014dc96f4fc"
|
|
source = "emit-fi-daily-brief-task"
|
|
key_aug4 = build_idempotency_key(
|
|
TaskSpec(
|
|
title="FI",
|
|
activity_definition_id=def_id,
|
|
source_id=source,
|
|
triggering_event_id=emit_triggering_event_id("scheduled", "run-aug4"),
|
|
)
|
|
)
|
|
key_aug5 = build_idempotency_key(
|
|
TaskSpec(
|
|
title="FI",
|
|
activity_definition_id=def_id,
|
|
source_id=source,
|
|
triggering_event_id=emit_triggering_event_id("scheduled", "run-aug5"),
|
|
)
|
|
)
|
|
# Old bug: both would be "...:scheduled"
|
|
legacy = f"{def_id}:{source}:scheduled"
|
|
assert key_aug4 != key_aug5
|
|
assert key_aug4 != legacy
|
|
assert key_aug5 != legacy
|
|
|
|
|
|
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
|