awoooi/scripts/ops/tests/test_gitea_queue_hook_backlog_playbook.py

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path


SCRIPT_ROOT = Path(__file__).resolve().parents[1]
PLAYBOOK_PATH = SCRIPT_ROOT / "gitea-queue-hook-backlog-playbook.py"


def _write_common_gitea_files(tmp_path: Path) -> dict[str, Path]:
    metrics_file = tmp_path / "gitea.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'gitea_build_info{goarch="amd64",goos="linux",goversion="go1.25.8",version="1.25.5"} 1',
                "gitea_hooktasks 1217",
                "gitea_repositories 13",
                "gitea_webhooks 2",
                "go_goroutines 121",
                "go_sched_gomaxprocs_threads 2",
                "process_cpu_seconds_total 41040.25",
                "process_open_fds 56",
                "process_resident_memory_bytes 8.86448128e+08",
            ]
        ),
        encoding="utf-8",
    )
    health_file = tmp_path / "health.json"
    health_file.write_text(
        json.dumps(
            {
                "status": "pass",
                "checks": {
                    "cache:ping": [{"status": "pass"}],
                    "database:ping": [{"status": "pass"}],
                },
            }
        ),
        encoding="utf-8",
    )
    version_file = tmp_path / "version.json"
    version_file.write_text('{"version":"1.25.5"}', encoding="utf-8")
    return {
        "gitea_metrics": metrics_file,
        "health": health_file,
        "version": version_file,
    }


def _run_playbook(
    tmp_path: Path,
    *,
    host_metrics: list[str],
    docker_metrics: list[str],
) -> subprocess.CompletedProcess[str]:
    gitea_files = _write_common_gitea_files(tmp_path)
    host_file = tmp_path / "host.prom"
    host_file.write_text("\n".join(host_metrics), encoding="utf-8")
    docker_file = tmp_path / "docker.prom"
    docker_file.write_text("\n".join(docker_metrics), encoding="utf-8")

    return subprocess.run(
        [
            sys.executable,
            str(PLAYBOOK_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(host_file),
            "--docker-stats-file",
            str(docker_file),
            "--gitea-metrics-file",
            str(gitea_files["gitea_metrics"]),
            "--gitea-health-file",
            str(gitea_files["health"]),
            "--gitea-version-file",
            str(gitea_files["version"]),
            "--json",
        ],
        capture_output=True,
        text=True,
    )


def test_gitea_playbook_classifies_hooktask_backlog_without_secret_reads(tmp_path: Path) -> None:
    result = _run_playbook(
        tmp_path,
        host_metrics=[
            'awoooi_host_gitea_actions_active_container_count{host="110"} 0',
            'awoooi_host_gitea_actions_active_process_group_count{host="110"} 0',
            'awoooi_host_gitea_actions_active_process_cpu_percent{host="110"} 0',
            'awoooi_host_process_family_cpu_percent{host="110",family="gitea_service"} 53.1',
            'awoooi_host_process_family_process_count{host="110",family="gitea_service"} 2',
        ],
        docker_metrics=['docker_container_cpu_cores{host="110",container_name="gitea"} 1.7052'],
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert payload["classification"] == "blocked_gitea_hooktask_backlog_check_required"
    assert payload["controlled_apply_allowed"] is False
    assert payload["readback"]["gitea_health_http"]["status"] == "pass"
    assert payload["readback"]["gitea_version_http"]["version"] == "1.25.5"
    assert payload["readback"]["selected_gitea_metrics"]["gitea_hooktasks"] == 1217
    assert payload["readback"]["gitea_container_cpu_cores"] == 1.7052
    assert payload["operation_boundaries"]["database_query_performed"] is False
    assert payload["operation_boundaries"]["secret_value_read"] is False
    assert "/home/wooo/gitea/app.ini" not in result.stdout
    assert "Authorization" not in result.stdout


def test_gitea_playbook_routes_active_actions_to_runner_queue_packet(tmp_path: Path) -> None:
    result = _run_playbook(
        tmp_path,
        host_metrics=[
            'awoooi_host_gitea_actions_active_container_count{host="110"} 1',
            'awoooi_host_gitea_actions_active_process_group_count{host="110"} 1',
            'awoooi_host_gitea_actions_active_process_cpu_percent{host="110"} 220.5',
            'awoooi_host_process_family_cpu_percent{host="110",family="gitea_service"} 20',
        ],
        docker_metrics=['docker_container_cpu_cores{host="110",container_name="gitea"} 1.4'],
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert payload["classification"] == "blocked_gitea_actions_pressure_requires_runner_queue_packet"
    assert payload["next_action"] == "run_runner_queue_readback_and_keep_110_pressure_gate_fail_closed"
    assert payload["readback"]["active_actions"]["container_count"] == 1
    assert payload["readback"]["active_actions"]["process_group_count"] == 1


def test_gitea_playbook_rejects_stale_docker_attribution(tmp_path: Path) -> None:
    result = _run_playbook(
        tmp_path,
        host_metrics=[
            'node_textfile_mtime_seconds{file="/host/home/wooo/node_exporter_textfiles/docker_stats.prom"} 1000',
            "node_time_seconds 5000",
            'awoooi_host_gitea_actions_active_container_count{host="110"} 0',
            'awoooi_host_gitea_actions_active_process_group_count{host="110"} 0',
        ],
        docker_metrics=['docker_container_cpu_cores{host="110",container_name="gitea"} 1.7052'],
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert payload["classification"] == "blocked_gitea_pressure_attribution_stale_requires_textfile_recovery"
    assert payload["readback"]["docker_stats"]["fresh"] is False
    assert payload["readback"]["top_containers"] == []
    assert payload["readback"]["top_containers_untrusted"][0]["container_name"] == "gitea"