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" def test_gitea_playbook_honors_custom_readback_paths(tmp_path: Path) -> None: gitea_files = _write_common_gitea_files(tmp_path) host_file = tmp_path / "host.prom" host_file.write_text( "\n".join( [ 'awoooi_host_gitea_actions_active_container_count{host="188"} 0', 'awoooi_host_gitea_actions_active_process_group_count{host="188"} 0', 'awoooi_host_gitea_actions_active_process_cpu_percent{host="188"} 0', 'awoooi_host_process_family_cpu_percent{host="188",family="gitea_service"} 0', ] ), encoding="utf-8", ) docker_file = tmp_path / "docker.prom" docker_file.write_text( 'docker_container_cpu_cores{host="188",container_name="momo-scheduler"} 0.7\n', encoding="utf-8", ) script_dir = tmp_path / "deployed-scripts" result = subprocess.run( [ sys.executable, str(PLAYBOOK_PATH), "--host", "188", "--metrics-file", str(host_file), "--docker-stats-file", str(docker_file), "--script-dir", str(script_dir), "--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, ) assert result.returncode == 0 payload = json.loads(result.stdout) assert payload["classification"] == "observing_gitea_pressure_below_threshold" assert payload["readback"]["metrics_file"] == str(host_file) assert payload["readback"]["docker_stats_file"] == str(docker_file) assert payload["readback"]["script_dir"] == str(script_dir) assert str(script_dir / "gitea-queue-hook-backlog-playbook.py") in payload["commands"]["check_mode"] assert str(script_dir / "host-sustained-load-controller.py") in payload["commands"]["post_apply_verifier"] assert str(host_file) in payload["commands"]["check_mode"] assert str(docker_file) in payload["commands"]["post_apply_verifier"] assert "/home/wooo/node_exporter_textfiles" not in payload["commands"]["check_mode"] assert "/home/wooo/node_exporter_textfiles" not in payload["commands"]["post_apply_verifier"]