activity-core/tests/test_repo_manager_freshness_definition.py

"""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 = {"receipts": [
            {"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["receipt_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_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: {"receipts": [], "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