from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from fastapi.testclient import TestClient from sqlmodel import SQLModel, create_engine from backend.app.connectors import gitea_actions from backend.app.connectors.gitea_actions import GiteaActionsConnector from backend.app.core.config import settings @pytest.fixture def api_client(tmp_path, monkeypatch): db_path = tmp_path / "appops_test.db" monkeypatch.setenv("APPOPS_DB_PATH", str(db_path)) from backend.app.db import session from backend.app.db.init_db import init_db from backend.app.main import app engine = create_engine( f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False} ) session.engine = engine SQLModel.metadata.drop_all(engine) SQLModel.metadata.create_all(engine) init_db() with TestClient(app) as client: yield client def _configure(monkeypatch): monkeypatch.setitem(settings.apis, "GITEA_URL", "https://gitea.test") monkeypatch.setitem(settings.apis, "GITEA_ADMIN_TOKEN", "fixture-secret") monkeypatch.setitem( settings.apis, "GITEA_RUNNER_TARGETS_JSON", '{"unraid-deploy":"itworx-security-runner"}', ) monkeypatch.setattr(gitea_actions, "_CACHE", None) monkeypatch.setattr(gitea_actions, "_RECOVERY_OBSERVATIONS", {}, raising=False) monkeypatch.setattr(gitea_actions, "_RECOVERY_SOURCE", None, raising=False) monkeypatch.setattr(gitea_actions, "_SNAPSHOT_GENERATION", 0, raising=False) monkeypatch.setattr(gitea_actions, "_COMMITTED_GENERATION", 0, raising=False) def test_runner_snapshot_recommends_only_approval_gated_restart_for_stalled_queue( monkeypatch, ): _configure(monkeypatch) connector = GiteaActionsConnector() created_at = (datetime.now(UTC) - timedelta(minutes=12)).isoformat() def fake_request(path, params=None): if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 7, "name": "deploy-runner", "status": "offline", "busy": False, "disabled": False, "labels": [{"name": "unraid-deploy"}], } ], } if params.get("status") == "queued": return { "total_count": 1, "jobs": [ { "id": 8, "status": "queued", "created_at": created_at, "labels": ["unraid-deploy"], } ], } return {"total_count": 0, "jobs": []} monkeypatch.setattr(connector, "_request", fake_request) result = connector.snapshot(refresh=True) assert result["severity"] == "critical" assert result["queue"]["queued"] == 1 assert result["queue"]["oldestQueuedAgeSeconds"] >= 600 assert result["queue"]["sampleTruncated"] is False assert result["recommendedAction"] == { "intent": "container.restart", "target": "itworx-security-runner", "requiresApproval": True, "autoExecute": False, "reason": "Gitea Actions queue is stalled without a matching online runner.", } assert "fixture-secret" not in repr(result) def test_pending_jobs_do_not_trigger_runnable_queue_diagnostics(monkeypatch): _configure(monkeypatch) monkeypatch.setitem( settings.apis, "GITEA_RUNNER_TARGETS_JSON", '{"itworx-ci-build":"itworx-security-runner"}', ) connector = GiteaActionsConnector() created_at = (datetime.now(UTC) - timedelta(minutes=15)).isoformat() requested_statuses = [] def fake_request(path, params=None): if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 1, "name": "itworx-security-runner", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation"], } ], } status = params["status"] requested_statuses.append(status) if status == "pending": return { "total_count": 1, "jobs": [ { "id": 171, "name": "conditional release full", "status": "requested", "created_at": created_at, "labels": ["itworx-ci-build"], } ], } return {"total_count": 0, "jobs": []} monkeypatch.setattr(connector, "_request", fake_request) result = connector.snapshot(refresh=True) assert requested_statuses == ["queued", "pending", "in_progress"] assert result["severity"] == "healthy" assert result["queue"]["queued"] == 0 assert result["queue"]["oldestQueuedAgeSeconds"] is None assert result["queue"]["sampleTruncated"] is False assert result["queue"]["blockedByRunnerLabels"] == [] assert result["queue"]["staleAssignmentRunnerLabels"] == [] assert result["queue"]["items"] == [] assert result["recommendedAction"] is None def test_runner_snapshot_is_cached_and_healthy_when_queue_is_clear(monkeypatch): _configure(monkeypatch) connector = GiteaActionsConnector() calls = 0 def fake_request(path, params=None): nonlocal calls calls += 1 if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 1, "name": "shared", "status": "online", "labels": ["ubuntu-latest"], } ], } return {"total_count": 0, "jobs": []} monkeypatch.setattr(connector, "_request", fake_request) first = connector.snapshot(refresh=True) second = connector.snapshot() assert first["severity"] == "healthy" assert first["recommendedAction"] is None assert second["cache"]["hit"] is True assert calls == 4 def test_long_running_shared_job_is_critical_without_ambiguous_restart(monkeypatch): _configure(monkeypatch) monkeypatch.setitem(settings.apis, "GITEA_RUNNER_TARGETS_JSON", "{}") connector = GiteaActionsConnector() created_at = (datetime.now(UTC) - timedelta(hours=3)).isoformat() def fake_request(path, params=None): if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 1, "name": "shared", "status": "online", "busy": True, "labels": ["ubuntu-latest"], } ], } if params.get("status") == "in_progress": return { "total_count": 1, "jobs": [ { "id": 9, "status": "in_progress", "created_at": created_at, "labels": ["ubuntu-latest"], } ], } return {"total_count": 0, "jobs": []} monkeypatch.setattr(connector, "_request", fake_request) result = connector.snapshot(refresh=True) assert result["severity"] == "critical" assert result["queue"]["oldestInProgressAgeSeconds"] >= 7_200 assert result["recommendedAction"] is None def test_recently_started_old_queued_job_is_not_reported_as_stale(monkeypatch): _configure(monkeypatch) connector = GiteaActionsConnector() created_at = (datetime.now(UTC) - timedelta(hours=3)).isoformat() started_at = (datetime.now(UTC) - timedelta(seconds=30)).isoformat() def fake_request(path, params=None): if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 1, "name": "shared", "status": "online", "busy": True, "labels": ["ubuntu-latest"], } ], } if params.get("status") == "in_progress": return { "total_count": 1, "jobs": [ { "id": 9, "status": "in_progress", "created_at": created_at, "started_at": started_at, "labels": ["ubuntu-latest"], } ], } return {"total_count": 0, "jobs": []} monkeypatch.setattr(connector, "_request", fake_request) result = connector.snapshot(refresh=True) assert result["severity"] == "healthy" assert 0 <= result["queue"]["oldestInProgressAgeSeconds"] < 120 assert result["queue"]["staleActiveRunnerLabels"] == [] def test_long_running_job_recommends_only_exact_mapped_runner(monkeypatch): _configure(monkeypatch) connector = GiteaActionsConnector() created_at = (datetime.now(UTC) - timedelta(hours=3)).isoformat() def fake_request(path, params=None): if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 7, "name": "deploy-runner", "status": "online", "busy": True, "labels": ["unraid-deploy"], } ], } if params.get("status") == "in_progress": return { "total_count": 1, "jobs": [ { "id": 10, "status": "in_progress", "created_at": created_at, "labels": ["unraid-deploy"], } ], } return {"total_count": 0, "jobs": []} monkeypatch.setattr(connector, "_request", fake_request) result = connector.snapshot(refresh=True) assert result["recommendedAction"] == { "intent": "container.restart", "target": "itworx-security-runner", "requiresApproval": True, "autoExecute": False, "reason": "A Gitea Actions job exceeded the two-hour guardrail on an explicitly mapped runner.", } def test_runner_status_endpoint_degrades_when_not_configured(api_client, monkeypatch): monkeypatch.setitem(settings.apis, "GITEA_URL", None) monkeypatch.setitem(settings.apis, "GITEA_ADMIN_TOKEN", None) monkeypatch.setattr(gitea_actions, "_CACHE", None) response = api_client.get("/api/gitea/actions/status") assert response.status_code == 200 assert response.json()["status"] == "missing_config" assert response.json()["recommendedAction"] is None def test_default_docker_target_falls_back_to_v1_capability(monkeypatch): _configure(monkeypatch) monkeypatch.setitem(settings.apis, "GITEA_RUNNER_TARGETS_JSON", "{}") target = GiteaActionsConnector._target_for_label( "docker-validation", [ { "name": "itworx-validation-runner-2", "labels": ["linux-validation", "docker-validation"], } ], GiteaActionsConnector._restart_targets(), ) assert target == "itworx-validation-runner-2" def test_default_docker_target_prefers_v2_specialized_runner(monkeypatch): _configure(monkeypatch) monkeypatch.setitem(settings.apis, "GITEA_RUNNER_TARGETS_JSON", "{}") target = GiteaActionsConnector._target_for_label( "docker-validation", [ { "name": "itworx-validation-runner-2", "labels": ["linux-validation"], }, { "name": "itworx-docker-runner", "labels": ["docker-validation"], }, ], GiteaActionsConnector._restart_targets(), ) assert target == "itworx-docker-runner" def test_explicit_runner_target_mapping_remains_authoritative(monkeypatch): _configure(monkeypatch) monkeypatch.setitem( settings.apis, "GITEA_RUNNER_TARGETS_JSON", '{"docker-validation":"operator-selected-container"}', ) target = GiteaActionsConnector._target_for_label( "docker-validation", [{"name": "some-registration", "labels": ["docker-validation"]}], GiteaActionsConnector._restart_targets(), ) assert target == "operator-selected-container"