state-hub/tests/test_consistency_sweep.py

from __future__ import annotations

import json
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from api.schemas.consistency_sweep import (
    ConsistencySweepIssueSummary,
    ConsistencySweepRemoteAllRun,
    ConsistencySweepRepoResult,
)
from api.services import consistency_sweep as sweep_service


@pytest.mark.asyncio
async def test_remote_all_sweep_records_progress_and_parses_stderr(client, monkeypatch):
    async def fake_run_remote_all_sweep(session, *, max_seconds: int):
        return ConsistencySweepRemoteAllRun(
            max_seconds=max_seconds,
            exit_code=0,
            lock_skipped=False,
            repos_processed=[
                ConsistencySweepRepoResult(
                    repo_slug="state-hub",
                    repo_path="/home/worsch/state-hub",
                    result="pass",
                    summary=ConsistencySweepIssueSummary(info=1),
                    fixes_applied=["pull: already up to date"],
                )
            ],
            skipped_clean=["demo-service"],
            skipped_missing=["remote-only"],
            skipped_budget=[],
            progress_event_id=None,
            started_at=datetime(2026, 6, 21, 12, 0, tzinfo=UTC),
            completed_at=datetime(2026, 6, 21, 12, 1, tzinfo=UTC),
        )

    monkeypatch.setattr(
        "api.routers.consistency_sweep.run_remote_all_sweep",
        AsyncMock(side_effect=fake_run_remote_all_sweep),
    )

    response = await client.post("/consistency/sweep/remote-all", json={"max_seconds": 300})

    assert response.status_code == 201, response.text
    body = response.json()
    assert body["exit_code"] == 0
    assert body["lock_skipped"] is False
    assert body["repos_processed"][0]["repo_slug"] == "state-hub"
    assert body["skipped_clean"] == ["demo-service"]
    assert body["skipped_missing"] == ["remote-only"]


@pytest.mark.asyncio
async def test_remote_all_sweep_lock_skip_is_idempotent(client, monkeypatch):
    async def fake_run_remote_all_sweep(session, *, max_seconds: int):
        return ConsistencySweepRemoteAllRun(
            max_seconds=max_seconds,
            exit_code=0,
            lock_skipped=True,
            repos_processed=[],
            skipped_clean=[],
            skipped_missing=[],
            skipped_budget=[],
            progress_event_id=None,
            started_at=datetime(2026, 6, 21, 12, 0, tzinfo=UTC),
            completed_at=datetime(2026, 6, 21, 12, 0, 1, tzinfo=UTC),
        )

    monkeypatch.setattr(
        "api.routers.consistency_sweep.run_remote_all_sweep",
        AsyncMock(side_effect=fake_run_remote_all_sweep),
    )

    response = await client.post("/consistency/sweep/remote-all", json={})

    assert response.status_code == 201, response.text
    assert response.json()["lock_skipped"] is True
    assert response.json()["repos_processed"] == []


def test_parse_stderr_extracts_skip_lists():
    stderr = (
        "  CLEAN (skipped): alpha, beta\n"
        "  NOT ON THIS HOST (skipped): gamma\n"
        "  BUDGET EXHAUSTED after 300s (skipped): delta, epsilon\n"
    )
    parsed = sweep_service._parse_stderr(stderr)
    assert parsed == {
        "skipped_clean": ["alpha", "beta"],
        "skipped_missing": ["gamma"],
        "skipped_budget": ["delta", "epsilon"],
    }


def test_extract_json_payload_skips_human_readable_prefix_lines():
    stdout = (
        "  CLEAN (skipped): quiet-repo\n"
        "  BUDGET EXHAUSTED after 30s (skipped): other-repo\n"
        '{\n  "repo_slug": "state-hub",\n  "repo_path": "/home/worsch/state-hub",\n'
        '  "result": "pass",\n  "summary": {"fail": 0, "warn": 0, "info": 0},\n'
        '  "fixes_applied": []\n}\n'
    )
    payload = sweep_service._extract_json_payload(stdout)
    assert payload["repo_slug"] == "state-hub"


def test_parse_stdout_handles_single_and_batch_payloads():
    single = json.dumps(
        {
            "repo_slug": "state-hub",
            "repo_path": "/home/worsch/state-hub",
            "result": "warn",
            "summary": {"fail": 0, "warn": 1, "info": 0},
            "fixes_applied": [],
        }
    )
    batch = json.dumps(
        [
            {
                "repo_slug": "alpha",
                "repo_path": "/tmp/alpha",
                "result": "pass",
                "summary": {"fail": 0, "warn": 0, "info": 0},
                "fixes_applied": [],
            },
            {
                "repo_slug": "beta",
                "repo_path": "/tmp/beta",
                "result": "fail",
                "summary": {"fail": 1, "warn": 0, "info": 0},
                "fixes_applied": [],
            },
        ]
    )

    single_result = sweep_service._parse_stdout(single)
    batch_result = sweep_service._parse_stdout(batch)

    assert len(single_result) == 1
    assert single_result[0].result == "warn"
    assert [repo.repo_slug for repo in batch_result] == ["alpha", "beta"]
    assert batch_result[1].result == "fail"


@pytest.mark.asyncio
async def test_run_remote_all_sweep_invokes_script_and_logs_progress(client, monkeypatch):
    captured: dict[str, object] = {}

    def fake_run(cmd, capture_output, text):
        captured["cmd"] = cmd
        return SimpleNamespace(
            stdout=json.dumps(
                [
                    {
                        "repo_slug": "state-hub",
                        "repo_path": "/home/worsch/state-hub",
                        "result": "pass",
                        "summary": {"fail": 0, "warn": 0, "info": 0},
                        "fixes_applied": [],
                    }
                ]
            ),
            stderr="  CLEAN (skipped): quiet-repo\n",
            returncode=0,
        )

    async def fake_to_thread(fn, *args, **kwargs):
        return fn(*args, **kwargs)

    monkeypatch.setattr(sweep_service.asyncio, "to_thread", fake_to_thread)
    monkeypatch.setattr(sweep_service.subprocess, "run", fake_run)

    response = await client.post("/consistency/sweep/remote-all", json={"max_seconds": 120})

    assert response.status_code == 201, response.text
    body = response.json()
    assert "--remote" in captured["cmd"]
    assert "--all" in captured["cmd"]
    assert captured["cmd"][captured["cmd"].index("--max-seconds") + 1] == "120"
    assert body["exit_code"] == 0
    assert body["skipped_clean"] == ["quiet-repo"]
    assert body["progress_event_id"] is not None

    events = await client.get(
        "/progress/",
        params={"event_type": "consistency_sweep_remote_all"},
    )
    assert events.status_code == 200, events.text
    assert events.json()[0]["detail"]["skipped_clean"] == ["quiet-repo"]