Files
Unraid-AppOps-Gateway-Public/backend/tests/test_runner_scheduler_liveness_recovery.py

475 lines
19 KiB
Python

from datetime import UTC, datetime, timedelta
from threading import Event, Thread, current_thread
import pytest
from backend.app.connectors import gitea_actions
from backend.app.connectors.gitea_actions import GiteaActionsConnector
from backend.app.core.config import settings
def _configure(monkeypatch) -> GiteaActionsConnector:
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",
'{"linux-validation":"itworx-validation-runner-2",'
'"mcp-validation":"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)
return GiteaActionsConnector()
def _requests(created_at: str, state: dict[str, object] | None = None):
state = state if state is not None else {}
def fake_request(path, params=None):
params = params or {}
if state.get("fail"):
raise RuntimeError("fixture offline")
if path.endswith("/runners"):
runners = list(state.get("runners", [
{
"id": state.get("runner_id", 17),
"name": state.get("runner_name", "itworx-validation-runner-2"),
"status": state.get("runner_status", "online"),
"busy": state.get("runner_busy", False),
"disabled": state.get("runner_disabled", False),
"labels": state.get("runner_labels", ["linux-validation", "ubuntu-latest"]),
}
]))
if state.get("duplicate_runner") and runners:
runners.append(dict(runners[0]))
return {
"total_count": state.get("runner_total", len(runners)),
"runners": runners,
}
status = params.get("status")
if status == "queued":
jobs = [] if state.get("queued") is False else [
{
"id": state.get("job_id", 91),
"run_id": state.get("run_id", 81),
"run_number": state.get("run_number"),
"workflow_id": state.get("workflow_id"),
"attempt": state.get("attempt"),
"run_attempt": state.get("run_attempt", 1),
"name": "full",
"status": "queued",
"created_at": created_at,
"labels": state.get("job_labels", ["linux-validation"]),
}
]
if state.get("duplicate_job"):
jobs.append(dict(jobs[0]))
if state.get("queued_wrong_status") and jobs:
jobs[0]["status"] = "pending"
return {
"total_count": state.get("queued_total", len(jobs)),
"jobs": jobs,
}
if status == "pending":
jobs = []
if state.get("pending"):
jobs = [{"id": 91, "run_id": 81, "run_attempt": 1,
"status": state.get("pending_status", "requested"), "created_at": created_at,
"labels": ["linux-validation"]}]
return {"total_count": len(jobs), "jobs": jobs}
if status == "in_progress":
jobs = []
if state.get("active_runner_id") is not None:
jobs = [{"id": 92, "run_id": 82, "run_attempt": 1,
"status": "in_progress", "runner_id": state["active_runner_id"],
"created_at": created_at, "started_at": created_at,
"labels": ["linux-validation"]}]
if state.get("active_malformed"):
jobs = [None]
return {"total_count": len(jobs), "jobs": jobs}
raise AssertionError(f"Unexpected Gitea request: {path} {params}")
return fake_request
def test_old_created_job_first_seen_as_queued_is_not_immediately_a_recovery_candidate(monkeypatch):
connector = _configure(monkeypatch)
created_at = (datetime.now(UTC) - timedelta(minutes=12)).isoformat()
monkeypatch.setattr(connector, "_request", _requests(created_at))
result = connector.snapshot(refresh=True)
assert result["queue"]["items"][0]["reason"] == "awaiting_scheduler_assignment"
assert result["queue"]["staleAssignmentRunnerLabels"] == []
assert result["recommendedAction"] is None
def test_old_created_job_requires_ten_minutes_of_fresh_queued_and_idle_observations(monkeypatch):
connector = _configure(monkeypatch)
created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
now = {"seconds": 1_000.0}
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(connector, "_request", _requests(created_at))
first = connector.snapshot(refresh=True)
assert first["recommendedAction"] is None
assert first["queue"]["oldestQueuedAgeSeconds"] >= 7_200
for sample in range(1, 20):
now["seconds"] = 1_000.0 + sample * 30
result = connector.snapshot(refresh=True)
assert result["recommendedAction"] is None
now["seconds"] = 1_600.0
result = connector.snapshot(refresh=True)
assert result["queue"]["staleAssignmentRunnerLabels"] == ["linux-validation"]
assert result["recommendedAction"] == {
"intent": "runner.recover",
"target": "itworx-validation-runner-2",
"requiresApproval": False,
"autoExecute": False,
"reason": "Gitea Actions scheduler left a compatible idle runner unassigned for at least ten minutes.",
"evidence": {
"jobId": 91,
"label": "linux-validation",
"observedClaimableSeconds": 600,
},
}
def test_recent_unassigned_job_does_not_trigger_recovery(monkeypatch):
connector = _configure(monkeypatch)
created_at = (datetime.now(UTC) - timedelta(minutes=2)).isoformat()
monkeypatch.setattr(connector, "_request", _requests(created_at))
result = connector.snapshot(refresh=True)
assert result["queue"]["items"][0]["reason"] == "awaiting_scheduler_assignment"
assert result["queue"]["staleAssignmentRunnerLabels"] == []
assert result["recommendedAction"] is None
def test_busy_matching_runner_is_never_restarted_for_waiting_queue(monkeypatch):
connector = _configure(monkeypatch)
created_at = (datetime.now(UTC) - timedelta(minutes=12)).isoformat()
monkeypatch.setattr(connector, "_request", _requests(created_at, {"runner_busy": True}))
result = connector.snapshot(refresh=True)
assert result["queue"]["items"][0]["reason"] == "all_matching_runners_are_busy"
assert result["queue"]["staleAssignmentRunnerLabels"] == []
assert result["recommendedAction"] is None
def test_cache_hit_never_replays_auto_recovery_recommendation(monkeypatch):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(
connector,
"_request",
_requests((datetime.now(UTC) - timedelta(hours=2)).isoformat()),
)
for seconds in range(0, 601, 30):
now["seconds"] = 1_000.0 + seconds
fresh = connector.snapshot(refresh=True)
assert fresh["recommendedAction"]["intent"] == "runner.recover"
cached = connector.snapshot()
assert cached["cache"]["hit"] is True
assert cached["recommendedAction"] is None
assert cached["queue"]["staleAssignmentRunnerLabels"] == []
def test_error_clears_cached_auto_recovery_candidate(monkeypatch):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
state: dict[str, object] = {}
created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(connector, "_request", _requests(created_at, state))
for seconds in range(0, 601, 30):
now["seconds"] = 1_000.0 + seconds
result = connector.snapshot(refresh=True)
assert result["recommendedAction"]["intent"] == "runner.recover"
now["seconds"] = 1_601.0
state["fail"] = True
assert connector.snapshot(refresh=True)["status"] == "error"
state.clear()
now["seconds"] = 1_602.0
refreshed = connector.snapshot()
assert refreshed["cache"]["hit"] is False
assert refreshed["recommendedAction"] is None
assert refreshed["queue"]["staleAssignmentRunnerLabels"] == []
def test_gap_longer_than_poll_window_restarts_observation_period(monkeypatch):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(
connector,
"_request",
_requests((datetime.now(UTC) - timedelta(hours=2)).isoformat()),
)
assert connector.snapshot(refresh=True)["recommendedAction"] is None
now["seconds"] = 1_046.0
assert connector.snapshot(refresh=True)["recommendedAction"] is None
for sample in range(1, 21):
now["seconds"] = 1_046.0 + sample * 30
result = connector.snapshot(refresh=True)
assert result["recommendedAction"]["intent"] == "runner.recover"
@pytest.mark.parametrize(
"interruption",
[
{"runner_busy": True},
{"runner_status": "offline"},
{"runner_disabled": True},
{"runner_name": "renamed-but-not-mapped"},
{"active_runner_id": 17},
{"active_malformed": True},
{"queued": False},
{"queued": False, "pending": True},
{"queued_total": 2},
{"runner_total": 2},
{"duplicate_runner": True},
{"runner_total": True},
{"attempt": 2},
{"job_labels": ["linux-validation" + ("x" * 120)]},
{"job_labels": ["linux-validation" + (" " * 105)]},
{"job_labels": [{"name": "linux-validation" + (" " * 105)}]},
{"job_labels": [" linux-validation"]},
{"job_labels": ["linux-validation "]},
{"job_labels": [{"name": " linux-validation"}]},
{"runner_labels": [" linux-validation", "ubuntu-latest"]},
{"runner_labels": ["linux-validation ", "ubuntu-latest"]},
{"runner_labels": ["linux-validation", "x" * 121]},
{"duplicate_job": True},
{"queued_wrong_status": True},
{"runner_busy": "false"},
],
ids=[
"busy",
"offline",
"disabled",
"exact-target-missing",
"active-assignment",
"malformed-active-inventory",
"job-absent",
"pending-only",
"incomplete-queued-count",
"incomplete-runner-count",
"duplicate-runner-id",
"boolean-runner-count",
"conflicting-attempt-fields",
"overlong-job-label",
"raw-overlong-job-label-with-whitespace",
"raw-overlong-dict-job-label-with-whitespace",
"prefixed-job-label-does-not-match-runner-label",
"suffixed-job-label-does-not-match-runner-label",
"prefixed-dict-job-label-does-not-match-runner-label",
"prefixed-runner-label-does-not-match-job-label",
"suffixed-runner-label-does-not-match-job-label",
"overlong-runner-label",
"duplicate-job-id",
"wrong-status",
"malformed-runner-busy-type",
],
)
def test_nonclaimable_or_incomplete_sample_resets_dwell(monkeypatch, interruption):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
state: dict[str, object] = {}
created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(connector, "_request", _requests(created_at, state))
assert connector.snapshot(refresh=True)["recommendedAction"] is None
now["seconds"] = 1_015.0
state.update(interruption)
interrupted = connector.snapshot(refresh=True)
action = interrupted["recommendedAction"]
assert not (isinstance(action, dict) and action.get("intent") == "runner.recover")
assert interrupted["queue"]["staleAssignmentRunnerLabels"] == []
state.clear()
now["seconds"] = 1_030.0
assert connector.snapshot(refresh=True)["recommendedAction"] is None
for seconds in range(60, 601, 30):
now["seconds"] = 1_000.0 + seconds
result = connector.snapshot(refresh=True)
assert result["recommendedAction"] is None
now["seconds"] = 1_630.0
assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover"
def test_failed_refresh_resets_recovery_observation(monkeypatch):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
state: dict[str, object] = {}
created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(connector, "_request", _requests(created_at, state))
assert connector.snapshot(refresh=True)["recommendedAction"] is None
now["seconds"] = 1_015.0
state["fail"] = True
failed = connector.snapshot(refresh=True)
assert failed["status"] == "error"
assert failed["recommendedAction"] is None
state.clear()
now["seconds"] = 1_030.0
assert connector.snapshot(refresh=True)["recommendedAction"] is None
for seconds in range(60, 601, 30):
now["seconds"] = 1_000.0 + seconds
result = connector.snapshot(refresh=True)
assert result["recommendedAction"] is None
now["seconds"] = 1_630.0
assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover"
def test_missing_configuration_resets_recovery_observation(monkeypatch):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(connector, "_request", _requests(created_at))
assert connector.snapshot(refresh=True)["recommendedAction"] is None
now["seconds"] = 1_015.0
monkeypatch.setitem(settings.apis, "GITEA_ADMIN_TOKEN", None)
missing = connector.snapshot(refresh=True)
assert missing["status"] == "missing_config"
monkeypatch.setitem(settings.apis, "GITEA_ADMIN_TOKEN", "fixture-secret")
now["seconds"] = 1_030.0
assert connector.snapshot(refresh=True)["recommendedAction"] is None
for seconds in range(60, 601, 30):
now["seconds"] = 1_000.0 + seconds
result = connector.snapshot(refresh=True)
assert result["recommendedAction"] is None
now["seconds"] = 1_630.0
assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover"
def test_switching_gitea_origin_does_not_preserve_old_source_dwell(monkeypatch):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(
connector,
"_request",
_requests((datetime.now(UTC) - timedelta(hours=2)).isoformat()),
)
assert connector.snapshot(refresh=True)["recommendedAction"] is None
now["seconds"] = 1_015.0
monkeypatch.setitem(settings.apis, "GITEA_URL", "https://other-gitea.test")
assert connector.snapshot(refresh=True)["recommendedAction"] is None
now["seconds"] = 1_030.0
monkeypatch.setitem(settings.apis, "GITEA_URL", "https://gitea.test")
assert connector.snapshot(refresh=True)["recommendedAction"] is None
for seconds in range(60, 601, 30):
now["seconds"] = 1_000.0 + seconds
result = connector.snapshot(refresh=True)
assert result["recommendedAction"] is None
now["seconds"] = 1_630.0
assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover"
@pytest.mark.parametrize(
"changed_identity",
[
{"run_attempt": 2},
{"run_attempt": None, "attempt": 2},
{"run_id": 82},
{"run_number": 10},
{"workflow_id": 20},
{"runner_id": 18},
{"runner_name": "renamed-but-not-mapped"},
{"runner_labels": ["linux-validation", "ubuntu-latest", "new-label"]},
{"job_labels": ["linux-validation", "ubuntu-latest"]},
],
ids=[
"attempt",
"legacy-attempt",
"run-id",
"run-number",
"workflow-id",
"mapped-runner-id",
"exact-target-name",
"runner-signature",
"job-labels",
],
)
def test_changed_job_or_exact_runner_identity_restarts_dwell(monkeypatch, changed_identity):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
state: dict[str, object] = {}
created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(connector, "_request", _requests(created_at, state))
assert connector.snapshot(refresh=True)["recommendedAction"] is None
now["seconds"] = 1_015.0
state.update(changed_identity)
assert connector.snapshot(refresh=True)["recommendedAction"] is None
state.clear()
now["seconds"] = 1_030.0
assert connector.snapshot(refresh=True)["recommendedAction"] is None
for seconds in range(60, 601, 30):
now["seconds"] = 1_000.0 + seconds
result = connector.snapshot(refresh=True)
assert result["recommendedAction"] is None
now["seconds"] = 1_630.0
assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover"
def test_slower_older_refresh_cannot_advance_newer_observation(monkeypatch):
connector = _configure(monkeypatch)
now = {"seconds": 1_000.0}
entered_slow_request = Event()
release_slow_request = Event()
results = {}
created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
normal_request = _requests(created_at)
def delayed_request(path, params=None):
if current_thread().name == "slow-old" and not entered_slow_request.is_set():
entered_slow_request.set()
assert release_slow_request.wait(timeout=5)
return normal_request(path, params)
monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"])
monkeypatch.setattr(connector, "_request", delayed_request)
slow = Thread(
name="slow-old",
target=lambda: results.setdefault("old", connector.snapshot(refresh=True)),
)
slow.start()
assert entered_slow_request.wait(timeout=5)
now["seconds"] = 1_700.0
results["new"] = connector.snapshot(refresh=True)
release_slow_request.set()
slow.join(timeout=5)
assert not slow.is_alive()
assert results["old"]["recommendedAction"] is None
assert results["new"]["recommendedAction"] is None
now["seconds"] = 1_701.0
latest = connector.snapshot(refresh=True)
assert latest["recommendedAction"] is None