"""Draft repo-manager-bounded-freshness definition (ACTIVITY-WP-0044 / RMGR-WP-0020-T03).""" from __future__ import annotations import json import subprocess from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from sqlalchemy.dialects import postgresql from activity_core import activities, sync_activity_definitions, sync_schedules from activity_core.bounded_operations import ( BOUNDED_OPERATION_REGISTRY, normalize_operation_result, operation_sources, validate_bounded_operations, ) from activity_core.context_resolvers import repo_manager from activity_core.definition_parser import parse_file DEF_PATH = ( Path(__file__).parent.parent / "activity-definitions" / "repo-manager-bounded-freshness.md" ) def _source(**params): base = {"limit": 6, "max_seconds": 240, "apply": True} base.update(params) return { "type": "repo-manager", "query": "freshness_run", "operation": "repo_manager_freshness", "required": True, "params": base, "bind_to": "context.freshness", } def test_definition_parses_disabled_with_reviewed_shape(): d = parse_file(DEF_PATH) assert d.id == "repo-manager-bounded-freshness" assert d.enabled is False tc = d.trigger_config assert tc["trigger_type"] == "cron" assert tc["cron_expression"] == "40 */2 * * *" assert tc["timezone"] == "UTC" assert tc["misfire_policy"] == "skip" [(source, spec)] = operation_sources(d.context_sources) assert spec.operation_id == "repo_manager_freshness" assert source["params"] == {"limit": 6, "max_seconds": 240, "apply": True} assert all(rule["condition"] == "False" for rule in d.rules) def test_operation_is_registered_with_single_attempt_and_timeout(): spec = BOUNDED_OPERATION_REGISTRY["repo_manager_freshness"] assert spec.temporal_max_attempts == 1 assert spec.resolve_before_execute is False assert spec.max_timeout_seconds >= 240 + repo_manager._TIMEOUT_GRACE_SECONDS @pytest.mark.parametrize( "params,error", [ ({"limit": 0}, "limit"), ({"limit": 13}, "limit"), ({"limit": True}, "limit"), ({"max_seconds": 271}, "max_seconds"), ({"max_seconds": "240"}, "max_seconds"), ({"apply": "yes"}, "apply"), ({"command": "rm -rf /"}, "code-fixed"), ({"rmgr_bin": "/tmp/x"}, "code-fixed"), ], ) def test_admission_fails_closed(params, error): with pytest.raises(ValueError, match=error): validate_bounded_operations([_source(**params)], []) def test_argv_is_code_fixed(monkeypatch): monkeypatch.delenv("ACTIVITY_CORE_RMGR_BIN", raising=False) assert repo_manager.build_freshness_argv({"limit": 6, "max_seconds": 240, "apply": True}) == [ "rmgr", "freshness", "run", "--limit", "6", "--max-seconds", "240", ] assert repo_manager.build_freshness_argv( {"limit": 3, "max_seconds": 60, "apply": False} )[-1] == "--dry-run" def test_invocation_projects_counts_only(monkeypatch): calls = [] def fake_run(argv, **kwargs): calls.append((argv, kwargs["timeout"])) out = {"schema": "repo-manager.freshness-run.v1", "results": [ {"repo": "a", "outcome": "applied", "detail": "must not persist"}, {"repo": "b", "outcome": "noop"}, {"repo": "c", "outcome": "refused"}, {"repo": "d", "outcome": "bogus"}, ]} return subprocess.CompletedProcess(argv, 0, json.dumps(out), "") monkeypatch.setattr(repo_manager.subprocess, "run", fake_run) raw = repo_manager.repo_manager_freshness({"limit": 6, "max_seconds": 240, "apply": True}) projected = normalize_operation_result("repo_manager_freshness", raw) assert calls[0][1] == 240 + repo_manager._TIMEOUT_GRACE_SECONDS assert projected["result_count"] == 4 assert projected["outcomes"]["applied"] == 1 assert projected["outcomes"]["noop"] == 1 assert projected["outcomes"]["refused"] == 1 assert projected["apply"] is True and projected["exit_code"] == 0 assert "must not persist" not in json.dumps(projected) def test_dry_run_rows_without_outcome_count_nothing(): raw = {"schema": "repo-manager.freshness-run.v1", "dry_run": True, "results": [{"slug": "a", "action": "reconcile"}, {"slug": "b", "action": "skip_noop"}], "apply": False, "limit": 2, "max_seconds": 60, "exit_code": 0} projected = normalize_operation_result("repo_manager_freshness", raw) assert projected["result_count"] == 2 assert sum(projected["outcomes"].values()) == 0 def test_invocation_fails_on_non_json(monkeypatch): monkeypatch.setattr( repo_manager.subprocess, "run", lambda argv, **kw: subprocess.CompletedProcess(argv, 2, "", "boom"), ) with pytest.raises(RuntimeError, match="exit 2"): repo_manager.repo_manager_freshness({"limit": 1, "max_seconds": 10, "apply": False}) @pytest.mark.asyncio async def test_execute_bounded_operation_dispatches_to_repo_manager(monkeypatch): monkeypatch.setattr( repo_manager, "repo_manager_freshness", lambda params: {"results": [], "apply": False, "limit": 1, "max_seconds": 10}, ) patch = await activities.execute_bounded_operation({"source": _source(apply=False)}) assert patch["freshness"]["kind"] == "repo_manager_freshness" @pytest.mark.asyncio async def test_sync_upserts_disabled_row_and_paused_schedule(monkeypatch): d = parse_file(DEF_PATH) monkeypatch.setattr(sync_activity_definitions, "scan_and_parse", lambda: [d]) statements = [] session = MagicMock() session.execute = AsyncMock(side_effect=lambda stmt: statements.append(stmt)) begin = MagicMock() begin.__aenter__ = AsyncMock(return_value=None) begin.__aexit__ = AsyncMock(return_value=False) session.begin = MagicMock(return_value=begin) ctx = MagicMock() ctx.__aenter__ = AsyncMock(return_value=session) ctx.__aexit__ = AsyncMock(return_value=False) assert await sync_activity_definitions.sync(lambda: ctx) == 1 params = statements[0].compile(dialect=postgresql.dialect()).params assert params["enabled"] is False assert params["id"] == sync_activity_definitions._definition_uuid(d.id) upserted = [] async def fake_upsert(client, defn): upserted.append(defn) monkeypatch.setattr(sync_schedules, "upsert_schedule", fake_upsert) monkeypatch.setattr(sync_schedules, "list_schedules", AsyncMock(return_value=[])) row = SimpleNamespace( id=params["id"], name=d.name, enabled=params["enabled"], trigger_config=d.trigger_config, context_sources=d.context_sources, task_templates=[], dedupe_key_strategy="skip", version=1, ) result = await sync_schedules.sync_schedule_rows(object(), [row]) assert result.upserted == 0 and result.paused == 1 assert upserted[0].enabled is False