2160 lines
71 KiB
Python
2160 lines
71 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
import types
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from threading import Event, Thread, current_thread
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
FLEET_PATH = ROOT / "ops" / "gitea-deploy-broker" / "runner_fleet.py"
|
|
|
|
|
|
@pytest.fixture
|
|
def fleet_module(monkeypatch):
|
|
if sys.platform == "win32" and "fcntl" not in sys.modules:
|
|
fcntl = types.ModuleType("fcntl")
|
|
fcntl.LOCK_EX = 2
|
|
fcntl.LOCK_UN = 8
|
|
fcntl.LOCK_NB = 4
|
|
fcntl.flock = lambda *_args, **_kwargs: None
|
|
monkeypatch.setitem(sys.modules, "fcntl", fcntl)
|
|
spec = importlib.util.spec_from_file_location("appops_runner_fleet_scheduler_test", FLEET_PATH)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
module._host_guard_state_unpatched = module._host_guard_state
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"runner_fleet_specialize",
|
|
types.SimpleNamespace(specialization_state=lambda: "legacy"),
|
|
)
|
|
monkeypatch.setattr(module, "_host_guard_state", lambda: ("clear", ""))
|
|
monkeypatch.setattr(
|
|
module.RunnerCapacityController,
|
|
"snapshot",
|
|
lambda _self: _healthy_capacity_snapshot(),
|
|
)
|
|
return module
|
|
|
|
|
|
def _healthy_capacity_snapshot():
|
|
return {
|
|
"status": "ok",
|
|
"decision": {"pressure": "headroom"},
|
|
"host": {
|
|
"swapCritical": False,
|
|
"loadRatio": 0.1,
|
|
"availableMemoryBytes": 12 * 1024**3,
|
|
"cacheFreeBytes": 100 * 1024**3,
|
|
"swapSampleAt": datetime.now(UTC).timestamp(),
|
|
},
|
|
}
|
|
|
|
|
|
def _healthy_snapshot():
|
|
runners = [
|
|
{
|
|
"name": name,
|
|
"running": True,
|
|
"health": "healthy",
|
|
"restarts": 0,
|
|
"startedAt": "2026-09-16T18:00:00Z",
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
}
|
|
for name in (
|
|
"itworx-security-runner",
|
|
"itworx-validation-runner-2",
|
|
)
|
|
]
|
|
return {
|
|
"status": "healthy",
|
|
"generatedAt": "2026-10-01T10:00:00Z",
|
|
"capacity": _healthy_capacity_snapshot(),
|
|
"networks": {"count": 0, "capacity": None, "cleanupCandidates": []},
|
|
"storage": {
|
|
"status": "healthy",
|
|
"freeBytes": 30 * 1024**3,
|
|
"totalBytes": 40 * 1024**3,
|
|
},
|
|
"scheduler": {
|
|
"status": "healthy",
|
|
"candidate": None,
|
|
"stallThresholdSeconds": 600,
|
|
},
|
|
"topology": {
|
|
"dind": {
|
|
"name": "gitea-runner-dind",
|
|
"running": True,
|
|
"health": "healthy",
|
|
"restarts": 0,
|
|
"startedAt": "2026-09-16T18:00:00Z",
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
},
|
|
"runners": runners,
|
|
"generalSlots": 2,
|
|
"runnerCount": len(runners),
|
|
"specializedRunners": [],
|
|
"burstRunner": None,
|
|
},
|
|
}
|
|
|
|
|
|
|
|
|
|
def _jobs_page(jobs=()):
|
|
rows = list(jobs)
|
|
return {"total_count": len(rows), "jobs": rows}
|
|
|
|
|
|
def _scheduler_api(state, error_type):
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
runners = state.get(
|
|
"runners",
|
|
[
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
],
|
|
)
|
|
return {"total_count": len(runners), "runners": runners}
|
|
status = params.get("status")
|
|
if status == state.get("error_status"):
|
|
raise error_type("runner_fleet_gitea_test_error")
|
|
rows = state.get(
|
|
{"queued": "queued", "pending": "pending", "in_progress": "active"}.get(
|
|
status, ""
|
|
),
|
|
[],
|
|
)
|
|
if status == state.get("incomplete_status"):
|
|
return {"total_count": len(rows) + 1, "jobs": rows}
|
|
return _jobs_page(rows)
|
|
|
|
return fake_gitea
|
|
|
|
|
|
def test_container_projection_includes_restart_policy(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
payload = [
|
|
{
|
|
"Config": {"Labels": {"com.docker.compose.project": "gitea-deploy-runners"}},
|
|
"HostConfig": {"RestartPolicy": {"Name": "unless-stopped", "MaximumRetryCount": 0}},
|
|
"State": {
|
|
"Running": True,
|
|
"StartedAt": "2026-09-21T16:00:00Z",
|
|
"OOMKilled": False,
|
|
"Health": {"Status": "healthy"},
|
|
},
|
|
"RestartCount": 2,
|
|
}
|
|
]
|
|
monkeypatch.setattr(service, "_run", lambda _args, timeout=20: __import__("json").dumps(payload))
|
|
|
|
result = service._container("gitea-runner-dind")
|
|
|
|
assert result["restartPolicy"] == "unless-stopped"
|
|
assert result["running"] is True
|
|
assert result["health"] == "healthy"
|
|
assert result["restarts"] == 2
|
|
|
|
|
|
def test_snapshot_keeps_exited_dind_actionable_without_querying_dind_networks(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
states = {
|
|
"gitea-runner-dind": {
|
|
"name": "gitea-runner-dind",
|
|
"running": False,
|
|
"health": "none",
|
|
"restarts": 0,
|
|
"startedAt": "",
|
|
"oomKilled": False,
|
|
"restartPolicy": "no",
|
|
},
|
|
"itworx-validation-runner-2": {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
"restarts": 0,
|
|
"startedAt": "2026-09-16T18:00:00Z",
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
},
|
|
"itworx-security-runner": {
|
|
"name": "itworx-security-runner",
|
|
"running": True,
|
|
"health": "healthy",
|
|
"restarts": 0,
|
|
"startedAt": "2026-09-16T18:00:00Z",
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
},
|
|
}
|
|
monkeypatch.setattr(service, "_container", lambda name: states[name])
|
|
monkeypatch.setattr(service, "_optional_container", lambda _name: None)
|
|
monkeypatch.setattr(service, "_storage", lambda: {"status": "healthy", "freeBytes": 20_000_000_000, "totalBytes": 40_000_000_000})
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_network_inventory",
|
|
lambda: pytest.fail("network inventory must not query an exited DinD daemon"),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"cleanup_candidates",
|
|
lambda: pytest.fail("cleanup must not query an exited DinD daemon"),
|
|
)
|
|
monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None)
|
|
|
|
result = service.snapshot()
|
|
|
|
assert result["status"] == "critical"
|
|
assert result["topology"]["dind"]["running"] is False
|
|
assert result["networks"] == {"count": 0, "capacity": None, "cleanupCandidates": []}
|
|
assert result["policy"]["hostDockerMutationTargets"][0] == "gitea-runner-dind"
|
|
|
|
|
|
def test_snapshot_does_not_query_dind_api_during_health_start_grace(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
now = 1_800_000_000
|
|
monkeypatch.setattr(fleet_module.time, "time", lambda: now)
|
|
states = {
|
|
"gitea-runner-dind": {
|
|
"name": "gitea-runner-dind",
|
|
"running": True,
|
|
"health": "starting",
|
|
"restarts": 0,
|
|
"startedAt": datetime.fromtimestamp(now - 10, UTC).isoformat(),
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
},
|
|
"itworx-validation-runner-2": {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
"restarts": 0,
|
|
"startedAt": datetime.fromtimestamp(now - 300, UTC).isoformat(),
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
},
|
|
"itworx-security-runner": {
|
|
"name": "itworx-security-runner",
|
|
"running": True,
|
|
"health": "healthy",
|
|
"restarts": 0,
|
|
"startedAt": datetime.fromtimestamp(now - 300, UTC).isoformat(),
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
},
|
|
}
|
|
monkeypatch.setattr(service, "_container", lambda name: states[name])
|
|
monkeypatch.setattr(service, "_optional_container", lambda _name: None)
|
|
monkeypatch.setattr(service, "_storage", lambda: {"status": "healthy", "freeBytes": 20_000_000_000, "totalBytes": 40_000_000_000})
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_network_inventory",
|
|
lambda: pytest.fail("DinD API must not be queried during health start grace"),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"cleanup_candidates",
|
|
lambda: pytest.fail("cleanup must not query DinD during health start grace"),
|
|
)
|
|
monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None)
|
|
|
|
result = service.snapshot()
|
|
|
|
assert result["status"] == "healthy"
|
|
assert result["topology"]["dind"]["health"] == "starting"
|
|
assert result["networks"] == {"count": 0, "capacity": None, "cleanupCandidates": []}
|
|
|
|
|
|
def test_automatic_cycle_defers_every_other_recovery_while_dind_is_starting(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["topology"]["dind"] = {
|
|
"name": "gitea-runner-dind",
|
|
"running": True,
|
|
"health": "starting",
|
|
"startedAt": "2026-09-21T16:00:00Z",
|
|
"restartPolicy": "unless-stopped",
|
|
}
|
|
commands: list[list[str]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert events == []
|
|
assert result["dindRecovery"] == {
|
|
"status": "starting",
|
|
"target": "gitea-runner-dind",
|
|
}
|
|
assert result["schedulerRecovery"] == {"status": "deferred_for_dind_recovery"}
|
|
|
|
|
|
def test_automatic_cycle_recovers_critical_exited_dind_with_safe_host_capacity(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["status"] = "critical"
|
|
snapshot["topology"]["dind"].update(
|
|
{
|
|
"running": False,
|
|
"health": "none",
|
|
"startedAt": "",
|
|
"restartPolicy": "no",
|
|
}
|
|
)
|
|
commands: list[tuple[list[str], int]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda target: target == "gitea-runner-dind")
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_run",
|
|
lambda args, timeout=20: commands.append((args, timeout)) or "",
|
|
)
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == [
|
|
(["docker", "update", "--restart", "unless-stopped", "gitea-runner-dind"], 30),
|
|
(["docker", "start", "gitea-runner-dind"], 30),
|
|
]
|
|
assert [event["action"] for event in events] == [
|
|
"restart-policy.update",
|
|
"start",
|
|
]
|
|
assert result["status"] == "ok"
|
|
assert result["dindRecovery"] == {
|
|
"status": "requested",
|
|
"restartPolicy": "requested",
|
|
"target": "gitea-runner-dind",
|
|
}
|
|
|
|
|
|
def test_automatic_cycle_repairs_fixed_dind_policy_when_healthy_and_guard_clear(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["topology"]["dind"]["restartPolicy"] = "no"
|
|
commands: list[tuple[list[str], int]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_run",
|
|
lambda args, timeout=20: commands.append((args, timeout)) or "",
|
|
)
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == [
|
|
(["docker", "update", "--restart", "unless-stopped", "gitea-runner-dind"], 30)
|
|
]
|
|
assert result["dindRecovery"] == {
|
|
"status": "not_needed",
|
|
"restartPolicy": "requested",
|
|
}
|
|
assert len(events) == 1
|
|
assert events[0]["action"] == "restart-policy.update"
|
|
|
|
|
|
def test_automatic_cycle_never_restarts_running_unhealthy_dind(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["status"] = "healthy"
|
|
snapshot["topology"]["dind"] = {
|
|
"name": "gitea-runner-dind",
|
|
"running": True,
|
|
"health": "unhealthy",
|
|
"startedAt": "2026-09-16T18:00:00Z",
|
|
"restartPolicy": "unless-stopped",
|
|
}
|
|
commands: list[list[str]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["dindRecovery"] == {
|
|
"status": "deferred",
|
|
"target": "gitea-runner-dind",
|
|
}
|
|
assert any(
|
|
event.get("action") == "restart.deferred"
|
|
and event.get("target") == "gitea-runner-dind"
|
|
for event in events
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("guard_state", "guard_reason"),
|
|
[
|
|
(("paused", "host_resource_guard_paused"), "host_resource_guard_paused"),
|
|
(("unavailable", "host_guard_state_untrusted"), "host_guard_state_untrusted"),
|
|
],
|
|
)
|
|
def test_automatic_cycle_never_mutates_when_host_guard_is_paused_or_untrusted(
|
|
fleet_module, monkeypatch, guard_state, guard_reason
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["topology"]["dind"].update(
|
|
{"running": False, "health": "none", "restartPolicy": "no"}
|
|
)
|
|
commands: list[list[str]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_dind", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event))
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: guard_state)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == guard_reason
|
|
assert commands == []
|
|
assert events == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"action",
|
|
[
|
|
"restart-policy-update",
|
|
"dind-start",
|
|
"scheduler-restart",
|
|
"network-remove",
|
|
"cache-prune",
|
|
"runner-start",
|
|
],
|
|
)
|
|
def test_guard_pause_appearing_after_cycle_entry_blocks_each_mutation(
|
|
fleet_module, monkeypatch, action
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
dind = snapshot["topology"]["dind"]
|
|
monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None)
|
|
if action == "restart-policy-update":
|
|
dind["restartPolicy"] = "no"
|
|
elif action == "dind-start":
|
|
dind.update({"running": False, "health": "none"})
|
|
elif action == "scheduler-restart":
|
|
snapshot["topology"]["runners"] = [
|
|
{
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
}
|
|
]
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22,
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda name: {
|
|
"name": name,
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
elif action == "network-remove":
|
|
snapshot["networks"]["cleanupCandidates"] = [
|
|
{
|
|
"id": "a" * 64,
|
|
"name": "GITEA-ACTIONS-TASK-22-WORKFLOW-ci-JOB-test-network",
|
|
"ageSeconds": 3600,
|
|
"reason": "empty_gitea_actions_task_network_in_dedicated_dind",
|
|
}
|
|
]
|
|
elif action == "cache-prune":
|
|
snapshot["storage"]["status"] = "critical"
|
|
elif action == "runner-start":
|
|
snapshot["topology"]["runners"] = [
|
|
{
|
|
"name": "itworx-validation-runner-2",
|
|
"running": False,
|
|
"health": "unhealthy",
|
|
}
|
|
]
|
|
|
|
guard_states = iter(
|
|
[
|
|
("clear", ""),
|
|
("paused", "host_resource_guard_paused"),
|
|
]
|
|
)
|
|
commands: list[list[str]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: next(guard_states))
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_dind", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == "host_resource_guard_paused"
|
|
assert commands == []
|
|
assert events == []
|
|
|
|
|
|
def test_critical_current_pressure_after_cycle_entry_blocks_mutation(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["topology"]["dind"]["restartPolicy"] = "no"
|
|
current_samples = iter(
|
|
[
|
|
{
|
|
"status": "ok",
|
|
"decision": {"pressure": "critical"},
|
|
"host": {"swapCritical": True},
|
|
}
|
|
]
|
|
)
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", ""))
|
|
monkeypatch.setattr(
|
|
fleet_module.RunnerCapacityController,
|
|
"snapshot",
|
|
lambda _self: next(current_samples),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_run",
|
|
lambda args, timeout=20: commands.append(args) or "",
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == "runner_capacity_pressure_critical"
|
|
assert commands == []
|
|
|
|
|
|
def test_stale_capacity_sample_blocks_automatic_mutation(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["topology"]["dind"]["restartPolicy"] = "no"
|
|
stale = _healthy_capacity_snapshot()
|
|
stale["host"]["swapSampleAt"] = (
|
|
datetime.now(UTC).timestamp()
|
|
- fleet_module.CAPACITY_SAMPLE_MAX_AGE_SECONDS
|
|
- 1
|
|
)
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", ""))
|
|
monkeypatch.setattr(
|
|
fleet_module.RunnerCapacityController,
|
|
"snapshot",
|
|
lambda _self: stale,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_run",
|
|
lambda args, timeout=20: commands.append(args) or "",
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == "runner_capacity_telemetry_unavailable"
|
|
assert commands == []
|
|
|
|
|
|
@pytest.mark.parametrize("capacity", [{"status": "unavailable"}, {"status": "not_configured"}])
|
|
def test_unavailable_or_missing_current_pressure_blocks_mutation(
|
|
fleet_module, monkeypatch, capacity
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["topology"]["dind"]["restartPolicy"] = "no"
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", ""))
|
|
monkeypatch.setattr(
|
|
fleet_module.RunnerCapacityController,
|
|
"snapshot",
|
|
lambda _self: capacity,
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_run",
|
|
lambda args, timeout=20: commands.append(args) or "",
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == "runner_capacity_telemetry_unavailable"
|
|
assert commands == []
|
|
|
|
|
|
def test_automatic_cycle_rejects_incomplete_critical_snapshot(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["status"] = "critical"
|
|
snapshot["topology"]["dind"].update(
|
|
{"running": False, "health": "none", "restartPolicy": "no"}
|
|
)
|
|
snapshot["topology"].pop("runnerCount")
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda *_args, **_kwargs: pytest.fail("critical fleet must not mutate")
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_record", lambda **_event: pytest.fail("critical fleet must not audit a mutation")
|
|
)
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", ""))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == "runner_fleet_snapshot_untrusted"
|
|
assert result["dindRecovery"]["status"] == "deferred"
|
|
|
|
|
|
def test_automatic_cycle_recovers_stalled_scheduler_in_critical_snapshot(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["status"] = "critical"
|
|
snapshot["scheduler"] = {
|
|
"status": "stalled",
|
|
"candidate": {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
"stallThresholdSeconds": 600,
|
|
}
|
|
commands: list[tuple[list[str], int]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: snapshot["scheduler"]["candidate"],
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda name: {
|
|
"name": name,
|
|
"running": True,
|
|
"health": "healthy",
|
|
"startedAt": "2026-09-16T18:00:00Z",
|
|
"restarts": 0,
|
|
"oomKilled": False,
|
|
"restartPolicy": "unless-stopped",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_run",
|
|
lambda args, timeout=20: commands.append((args, timeout)) or "",
|
|
)
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event) or True)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == [
|
|
(["docker", "restart", "--timeout", "30", "itworx-validation-runner-2"], 45)
|
|
]
|
|
assert result["status"] == "ok"
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "requested",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
}
|
|
assert [event["action"] for event in events] == ["restart", "restart.result"]
|
|
|
|
|
|
def test_automatic_cycle_allows_critical_storage_prune_with_safe_host_capacity(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["status"] = "critical"
|
|
snapshot["storage"] = {
|
|
"status": "critical",
|
|
"freeBytes": 8 * 1024**3,
|
|
"totalBytes": 40 * 1024**3,
|
|
}
|
|
commands: list[list[str]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None)
|
|
monkeypatch.setattr(service, "_dind", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == [
|
|
["image", "prune", "--force"],
|
|
["builder", "prune", "--force", "--filter", "until=168h"],
|
|
]
|
|
assert result["status"] == "ok"
|
|
assert events == [
|
|
{
|
|
"action": "cache.prune",
|
|
"target": "gitea-runner-dind",
|
|
"reason": "ci_storage_critical",
|
|
"result": "completed",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("capacity", "reason"),
|
|
[
|
|
({"status": "ok", "decision": {"pressure": "critical"}, "host": {"swapCritical": False}},
|
|
"runner_capacity_pressure_critical"),
|
|
({"status": "unavailable"}, "runner_capacity_telemetry_unavailable"),
|
|
({"status": "ok", "decision": {}, "host": {}}, "runner_capacity_telemetry_untrusted"),
|
|
],
|
|
)
|
|
def test_automatic_cycle_blocks_critical_or_untrusted_capacity_snapshot(
|
|
fleet_module, monkeypatch, capacity, reason
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
snapshot["status"] = "critical"
|
|
snapshot["capacity"] = capacity
|
|
snapshot["topology"]["dind"].update(
|
|
{"running": False, "health": "none", "restartPolicy": "no"}
|
|
)
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda *_args, **_kwargs: pytest.fail("unsafe capacity must not mutate")
|
|
)
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", ""))
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == reason
|
|
|
|
|
|
def test_host_guard_probe_fails_closed_when_path_is_not_posix_root(monkeypatch, fleet_module):
|
|
monkeypatch.setattr(fleet_module.os, "name", "nt")
|
|
|
|
assert fleet_module._host_guard_state_unpatched() == (
|
|
"unavailable",
|
|
"host_guard_state_untrusted",
|
|
)
|
|
|
|
|
|
def test_recovery_budget_counts_start_and_restart_attempts(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
now = 1_800_000_000
|
|
monkeypatch.setattr(fleet_module.time, "time", lambda: now)
|
|
recent = datetime.fromtimestamp(now - 60, UTC).isoformat()
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_events",
|
|
lambda: [
|
|
{"at": recent, "action": "start", "target": "gitea-runner-dind"},
|
|
{"at": recent, "action": "restart", "target": "gitea-runner-dind"},
|
|
{"at": recent, "action": "start", "target": "gitea-runner-dind"},
|
|
],
|
|
)
|
|
|
|
assert service._recovery_available("gitea-runner-dind") is False
|
|
|
|
|
|
def test_scheduler_candidate_uses_live_gitea_queue_and_fixed_label_target(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [1_000.0]
|
|
created_at = datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat()
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
return {
|
|
"total_count": 1,
|
|
"runners": [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation", "ubuntu-latest"],
|
|
}
|
|
]
|
|
}
|
|
if params.get("status") == "queued":
|
|
return _jobs_page(
|
|
[
|
|
{
|
|
"id": 22931,
|
|
"status": "queued",
|
|
"run_id": 3001,
|
|
"attempt": 1,
|
|
"created_at": created_at,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
]
|
|
)
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
# An old created_at is not evidence of an old runnable/claimable interval.
|
|
assert service._scheduler_recovery_candidate() is None
|
|
candidate = None
|
|
for _ in range(fleet_module.SCHEDULER_STALL_SECONDS // 60):
|
|
monotonic[0] += 60
|
|
candidate = service._scheduler_recovery_candidate()
|
|
|
|
assert candidate == {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"runnerId": 17,
|
|
"labels": ("linux-validation",),
|
|
"reason": "stale_scheduler_assignment",
|
|
"runIdentity": (("run_id", 3001), ("attempt", 1)),
|
|
"ageSeconds": fleet_module.SCHEDULER_STALL_SECONDS,
|
|
}
|
|
|
|
|
|
def test_pending_age_is_not_carried_into_newly_queued_claimability(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [1_000.0]
|
|
created_at = datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat()
|
|
pending_job = {
|
|
"id": 22931,
|
|
"status": "requested",
|
|
"created_at": created_at,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
queued_job = {**pending_job, "status": "queued"}
|
|
state = {"pending": [pending_job], "queued": [], "active": []}
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
state["pending"] = []
|
|
state["queued"] = [queued_job]
|
|
monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS
|
|
# The first queued observation starts the timer even though created_at is old.
|
|
assert service._scheduler_recovery_candidate() is None
|
|
for _ in range(fleet_module.SCHEDULER_STALL_SECONDS // 60):
|
|
monotonic[0] += 60
|
|
candidate = service._scheduler_recovery_candidate()
|
|
|
|
assert candidate is not None
|
|
assert candidate["jobId"] == 22931
|
|
assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS
|
|
|
|
|
|
@pytest.mark.parametrize("uncertainty", ["error", "truncated"])
|
|
def test_scheduler_claimability_resets_after_unknown_or_incomplete_observation(
|
|
fleet_module, monkeypatch, uncertainty
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [2_000.0]
|
|
old_job = {
|
|
"id": 22931,
|
|
"status": "queued",
|
|
"created_at": datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat(),
|
|
"labels": ["linux-validation"],
|
|
}
|
|
state = {"queued": [old_job], "pending": [], "active": []}
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
monotonic[0] += 60
|
|
if uncertainty == "error":
|
|
state["error_status"] = "pending"
|
|
else:
|
|
state["incomplete_status"] = "queued"
|
|
with pytest.raises(fleet_module.RunnerFleetError):
|
|
service._scheduler_recovery_candidate()
|
|
|
|
state.pop("error_status", None)
|
|
state.pop("incomplete_status", None)
|
|
monotonic[0] += 60
|
|
assert service._scheduler_recovery_candidate() is None
|
|
for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 2):
|
|
monotonic[0] += 60
|
|
candidate = service._scheduler_recovery_candidate()
|
|
# The pre-error timer would have matured at this instant if unknown time
|
|
# had been bridged instead of starting over at the first fresh sample.
|
|
assert candidate is None
|
|
for _ in range(2):
|
|
monotonic[0] += 60
|
|
candidate = service._scheduler_recovery_candidate()
|
|
assert candidate is not None
|
|
assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS
|
|
|
|
|
|
@pytest.mark.parametrize("loss", ["busy", "active"])
|
|
def test_scheduler_claimability_resets_when_target_becomes_unavailable(
|
|
fleet_module, monkeypatch, loss
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [3_000.0]
|
|
job = {
|
|
"id": 22931,
|
|
"status": "queued",
|
|
"labels": ["linux-validation"],
|
|
}
|
|
state = {"queued": [job], "pending": [], "active": []}
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
monotonic[0] += 60
|
|
if loss == "busy":
|
|
state["runners"] = [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": True,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
]
|
|
else:
|
|
state["active"] = [
|
|
{"id": 22932, "status": "in_progress", "runner_id": 17}
|
|
]
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
state.pop("runners", None)
|
|
state["active"] = []
|
|
monotonic[0] += 60
|
|
assert service._scheduler_recovery_candidate() is None
|
|
for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 1):
|
|
monotonic[0] += 60
|
|
candidate = service._scheduler_recovery_candidate()
|
|
assert candidate is None
|
|
monotonic[0] += 60
|
|
candidate = service._scheduler_recovery_candidate()
|
|
assert candidate is not None
|
|
assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS
|
|
|
|
|
|
def test_scheduler_claimability_gap_restarts_the_observation_window(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [4_000.0]
|
|
state = {
|
|
"queued": [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}],
|
|
"pending": [],
|
|
"active": [],
|
|
}
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
monotonic[0] += fleet_module.SCHEDULER_OBSERVATION_MAX_GAP_SECONDS + 1
|
|
assert service._scheduler_recovery_candidate() is None
|
|
monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS
|
|
# A second long gap resets again instead of treating elapsed wall time as observed.
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
def test_older_eligible_sample_cannot_repopulate_after_newer_busy_sample(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [1_000.0]
|
|
queued_job = {
|
|
"id": 22931,
|
|
"status": "queued",
|
|
"run_id": 3001,
|
|
"attempt": 1,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
idle_state = {"queued": [queued_job], "pending": [], "active": []}
|
|
live_state = {"queued": [queued_job], "pending": [], "active": []}
|
|
stale_request = _scheduler_api(idle_state, fleet_module.RunnerFleetError)
|
|
live_request = _scheduler_api(live_state, fleet_module.RunnerFleetError)
|
|
older_sample_entered = Event()
|
|
release_older_sample = Event()
|
|
older_result: dict[str, object] = {}
|
|
|
|
def fake_gitea(path, params=None):
|
|
if current_thread().name == "older-idle-sample":
|
|
if not older_sample_entered.is_set():
|
|
older_sample_entered.set()
|
|
assert release_older_sample.wait(timeout=5)
|
|
return stale_request(path, params)
|
|
return live_request(path, params)
|
|
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
monotonic[0] = 1_010.0
|
|
older_sample = Thread(
|
|
name="older-idle-sample",
|
|
target=lambda: older_result.setdefault(
|
|
"candidate", service._scheduler_recovery_candidate()
|
|
),
|
|
)
|
|
older_sample.start()
|
|
assert older_sample_entered.wait(timeout=5)
|
|
|
|
monotonic[0] = 1_020.0
|
|
live_state["runners"] = [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": True,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
]
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
monotonic[0] = 1_030.0
|
|
release_older_sample.set()
|
|
older_sample.join(timeout=5)
|
|
assert not older_sample.is_alive()
|
|
assert older_result["candidate"] is None
|
|
|
|
live_state.pop("runners")
|
|
monotonic[0] = 1_031.0
|
|
assert service._scheduler_recovery_candidate() is None
|
|
for _ in range(9):
|
|
monotonic[0] += 60
|
|
candidate = service._scheduler_recovery_candidate()
|
|
|
|
# The rejected stale sample must not have started the ten-minute dwell.
|
|
assert monotonic[0] == 1_571.0
|
|
assert candidate is None
|
|
monotonic[0] += 59
|
|
assert service._scheduler_recovery_candidate() is None
|
|
monotonic[0] += 1
|
|
candidate = service._scheduler_recovery_candidate()
|
|
assert candidate is not None
|
|
assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS
|
|
|
|
|
|
def test_scheduler_claimability_is_process_local_and_lost_on_restart(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [5_000.0]
|
|
state = {
|
|
"queued": [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}],
|
|
"pending": [],
|
|
"active": [],
|
|
}
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
fleet_module.RunnerFleetService._clear_scheduler_claimability()
|
|
monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
@pytest.mark.parametrize("identity_change", ["origin", "job", "run", "labels", "runner"])
|
|
def test_scheduler_claimability_restarts_when_job_or_runner_identity_changes(
|
|
fleet_module, monkeypatch, identity_change
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [6_000.0]
|
|
job = {
|
|
"id": 22931,
|
|
"run_id": 3001,
|
|
"attempt": 1,
|
|
"status": "queued",
|
|
"labels": ["linux-validation"],
|
|
}
|
|
state = {"queued": [job], "pending": [], "active": []}
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
assert service._scheduler_recovery_candidate() is None
|
|
for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 1):
|
|
monotonic[0] += 60
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
if identity_change == "origin":
|
|
monkeypatch.setattr(
|
|
fleet_module, "GITEA_API_BASE", "http://other-gitea:3000"
|
|
)
|
|
elif identity_change == "job":
|
|
job["id"] = 22932
|
|
elif identity_change == "run":
|
|
job["run_id"] = 3002
|
|
elif identity_change == "labels":
|
|
job["labels"] = ["ubuntu-latest"]
|
|
state["runners"] = [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation", "ubuntu-latest"],
|
|
}
|
|
]
|
|
else:
|
|
state["runners"] = [
|
|
{
|
|
"id": 18,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
]
|
|
monotonic[0] += 60
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
def test_scheduler_candidate_pages_jobs_at_gitea_fifty_row_limit(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
queued = [
|
|
{"id": job_id, "status": "queued", "labels": ["unmapped-label"]}
|
|
for job_id in range(1, 51)
|
|
]
|
|
queued.append(
|
|
{"id": 51, "status": "queued", "labels": ["linux-validation"]}
|
|
)
|
|
job_calls: list[dict[str, object]] = []
|
|
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
return {
|
|
"total_count": 1,
|
|
"runners": [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
],
|
|
}
|
|
status = params["status"]
|
|
if status == "queued":
|
|
job_calls.append(params)
|
|
start = (int(params["page"]) - 1) * 50
|
|
return {"total_count": 51, "jobs": queued[start : start + 50]}
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
assert [(call["page"], call["limit"]) for call in job_calls] == [
|
|
(1, fleet_module.GITEA_JOB_PAGE_LIMIT),
|
|
(2, fleet_module.GITEA_JOB_PAGE_LIMIT),
|
|
]
|
|
|
|
|
|
def test_scheduler_candidate_rejects_job_inventory_over_one_thousand_rows(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
|
|
def fake_gitea(path, params=None):
|
|
if path.endswith("/runners"):
|
|
return {"total_count": 0, "runners": []}
|
|
return {"total_count": 1_001, "jobs": []}
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
with pytest.raises(
|
|
fleet_module.RunnerFleetError,
|
|
match="runner_fleet_gitea_invalid_job_inventory",
|
|
):
|
|
service._scheduler_recovery_candidate()
|
|
|
|
|
|
def test_scheduler_candidate_fails_closed_when_job_appears_in_two_statuses(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monotonic = [7_000.0]
|
|
job = {"id": 22931, "labels": ["linux-validation"]}
|
|
state = {
|
|
"queued": [{**job, "status": "queued"}],
|
|
"pending": [{**job, "status": "requested"}],
|
|
"active": [],
|
|
}
|
|
monkeypatch.setattr(
|
|
fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0]
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
|
|
with pytest.raises(
|
|
fleet_module.RunnerFleetError,
|
|
match="runner_fleet_gitea_job_status_changed_during_read",
|
|
):
|
|
service._scheduler_recovery_candidate()
|
|
state["pending"] = []
|
|
monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
def test_scheduler_candidate_does_not_classify_offline_target_as_claimable(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
state = {
|
|
"queued": [
|
|
{
|
|
"id": 22931,
|
|
"status": "queued",
|
|
"created_at": datetime.fromtimestamp(
|
|
1_800_000_000 - 7_200, UTC
|
|
).isoformat(),
|
|
"labels": ["linux-validation"],
|
|
}
|
|
],
|
|
"pending": [],
|
|
"active": [],
|
|
"runners": [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "offline",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
],
|
|
}
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("label_source", "expected_error"),
|
|
[
|
|
("job", "runner_fleet_gitea_invalid_queued_job"),
|
|
("runner", "runner_fleet_gitea_invalid_runner_inventory"),
|
|
],
|
|
)
|
|
def test_scheduler_candidate_rejects_labels_that_would_change_when_trimmed(
|
|
fleet_module, monkeypatch, label_source, expected_error
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
state = {
|
|
"queued": [
|
|
{"id": 22931, "status": "queued", "labels": ["linux-validation"]}
|
|
],
|
|
"pending": [],
|
|
"active": [],
|
|
}
|
|
if label_source == "job":
|
|
state["queued"][0]["labels"] = [" linux-validation"]
|
|
else:
|
|
state["runners"] = [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation "],
|
|
}
|
|
]
|
|
monkeypatch.setattr(
|
|
service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError)
|
|
)
|
|
|
|
with pytest.raises(fleet_module.RunnerFleetError, match=expected_error):
|
|
service._scheduler_recovery_candidate()
|
|
|
|
|
|
def test_scheduler_candidate_never_targets_busy_runner(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
return {
|
|
"total_count": 1,
|
|
"runners": [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": True,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
]
|
|
}
|
|
if params.get("status") == "queued":
|
|
return _jobs_page(
|
|
[{"id": 22931, "status": "queued", "labels": ["linux-validation"]}]
|
|
)
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
def test_scheduler_candidate_never_restarts_runner_with_in_progress_job(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
return {
|
|
"total_count": 1,
|
|
"runners": [{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}],
|
|
}
|
|
if params.get("status") == "queued":
|
|
return _jobs_page(
|
|
[{"id": 22931, "status": "queued", "labels": ["linux-validation"]}]
|
|
)
|
|
if params.get("status") == "in_progress":
|
|
return _jobs_page(
|
|
[{"id": 22932, "runner_id": 17, "status": "in_progress"}]
|
|
)
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
def test_scheduler_candidate_never_hides_disabled_busy_runner(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
if params.get("disabled") == "false":
|
|
return {"total_count": 0, "runners": []}
|
|
return {
|
|
"total_count": 1,
|
|
"runners": [
|
|
{
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": True,
|
|
"disabled": True,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
]
|
|
}
|
|
if params.get("status") == "queued":
|
|
return _jobs_page(
|
|
[{"id": 22931, "status": "queued", "labels": ["linux-validation"]}]
|
|
)
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
def test_scheduler_candidate_reads_later_runner_inventory_pages(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
runner_calls: list[dict[str, object]] = []
|
|
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
runner_calls.append(params)
|
|
if params.get("page") == 2:
|
|
return {
|
|
"total_count": 2,
|
|
"runners": [
|
|
{
|
|
"id": 18,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": True,
|
|
"disabled": True,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
],
|
|
}
|
|
return {
|
|
"total_count": 2,
|
|
"runners": [
|
|
{
|
|
"id": 17,
|
|
"name": "unrelated-runner",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["unrelated"],
|
|
}
|
|
],
|
|
}
|
|
if params.get("status") == "queued":
|
|
return _jobs_page(
|
|
[{"id": 22931, "status": "queued", "labels": ["linux-validation"]}]
|
|
)
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
assert [call.get("page") for call in runner_calls] == [1, 2]
|
|
|
|
|
|
def test_scheduler_candidate_fails_closed_on_malformed_runner_inventory(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
|
|
def fake_gitea(path, params=None):
|
|
if path.endswith("/runners"):
|
|
return {"total_count": 1}
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
with pytest.raises(
|
|
fleet_module.RunnerFleetError,
|
|
match="runner_fleet_gitea_invalid_runner_inventory",
|
|
):
|
|
service._scheduler_recovery_candidate()
|
|
|
|
|
|
@pytest.mark.parametrize("unsafe_first", [True, False])
|
|
def test_scheduler_candidate_vetoes_any_unsafe_duplicate_fixed_runner(
|
|
fleet_module, monkeypatch, unsafe_first
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
unsafe = {
|
|
"id": 17,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": True,
|
|
"disabled": True,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
idle = {
|
|
"id": 18,
|
|
"name": "itworx-validation-runner-2",
|
|
"status": "online",
|
|
"busy": False,
|
|
"disabled": False,
|
|
"labels": ["linux-validation"],
|
|
}
|
|
|
|
def fake_gitea(path, params=None):
|
|
params = params or {}
|
|
if path.endswith("/runners"):
|
|
runners = [unsafe, idle] if unsafe_first else [idle, unsafe]
|
|
return {"total_count": 2, "runners": runners}
|
|
if params.get("status") == "queued":
|
|
return _jobs_page(
|
|
[{"id": 22931, "status": "queued", "labels": ["linux-validation"]}]
|
|
)
|
|
return _jobs_page()
|
|
|
|
monkeypatch.setattr(service, "_gitea_request", fake_gitea)
|
|
|
|
assert service._scheduler_recovery_candidate() is None
|
|
|
|
|
|
def test_automatic_cycle_rechecks_the_same_scheduler_candidate_before_restart(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
events: list[dict[str, object]] = []
|
|
candidate = {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"runnerId": 17,
|
|
"labels": ("linux-validation",),
|
|
"runIdentity": (("run_id", 3001), ("attempt", 1)),
|
|
"ageSeconds": 600,
|
|
"reason": "stale_scheduler_assignment",
|
|
}
|
|
changed = {**candidate, "runIdentity": (("run_id", 3002), ("attempt", 1))}
|
|
candidates = iter([candidate, changed])
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service, "_scheduler_recovery_candidate", lambda: next(candidates)
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda args, timeout=20: commands.append(args) or ""
|
|
)
|
|
monkeypatch.setattr(service, "_record", lambda **event: events.append(event) or True)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "ineligible",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"reason": "scheduler_claimability_changed",
|
|
}
|
|
assert any(
|
|
event.get("action") == "restart.result"
|
|
and event.get("result") == "cancelled"
|
|
for event in events
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("unsafe_source", "expected_reason"),
|
|
[
|
|
("host_guard", "host_resource_guard_paused"),
|
|
("pressure", "runner_capacity_pressure_critical"),
|
|
],
|
|
)
|
|
def test_automatic_cycle_rechecks_host_safety_after_final_scheduler_read(
|
|
fleet_module, monkeypatch, unsafe_source, expected_reason
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
events: list[dict[str, object]] = []
|
|
safety_changed = [False]
|
|
candidate = {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"runnerId": 17,
|
|
"labels": ("linux-validation",),
|
|
"runIdentity": (("run_id", 3001), ("attempt", 1)),
|
|
"ageSeconds": 600,
|
|
"reason": "stale_scheduler_assignment",
|
|
}
|
|
scheduler_reads = [0]
|
|
|
|
def read_candidate():
|
|
scheduler_reads[0] += 1
|
|
if scheduler_reads[0] == 2:
|
|
safety_changed[0] = True
|
|
return candidate
|
|
|
|
def read_host_guard():
|
|
if safety_changed[0] and unsafe_source == "host_guard":
|
|
return "paused", "host_resource_guard_paused"
|
|
return "clear", ""
|
|
|
|
def read_capacity(_self):
|
|
capacity = _healthy_capacity_snapshot()
|
|
if safety_changed[0] and unsafe_source == "pressure":
|
|
capacity["decision"]["pressure"] = "critical"
|
|
return capacity
|
|
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(service, "_scheduler_recovery_candidate", read_candidate)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
monkeypatch.setattr(fleet_module, "_host_guard_state", read_host_guard)
|
|
monkeypatch.setattr(
|
|
fleet_module.RunnerCapacityController, "snapshot", read_capacity
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda args, timeout=20: commands.append(args) or ""
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_record", lambda **event: events.append(event) or True
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["status"] == "deferred"
|
|
assert result["reason"] == expected_reason
|
|
assert commands == []
|
|
# The durable intent remains in the budget even though the final fresh
|
|
# safety check cancelled the Docker mutation.
|
|
assert any(
|
|
event.get("action") == "restart" and event.get("result") == "intent"
|
|
for event in events
|
|
)
|
|
assert not any(event.get("action") == "restart.result" for event in events)
|
|
|
|
|
|
def test_automatic_cycle_restarts_only_exact_known_idle_runner_for_stale_scheduler(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[tuple[list[str], int]] = []
|
|
events: list[dict[str, object]] = []
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_run",
|
|
lambda args, timeout=20: commands.append((args, timeout)) or "",
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_record", lambda **event: events.append(event) or True
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == [
|
|
(["docker", "restart", "--timeout", "30", "itworx-validation-runner-2"], 45)
|
|
]
|
|
assert any(
|
|
event.get("action") == "restart"
|
|
and event.get("target") == "itworx-validation-runner-2"
|
|
and event.get("reason") == "stale_scheduler_assignment"
|
|
for event in events
|
|
)
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "requested",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
}
|
|
|
|
|
|
def test_scheduler_restart_persists_intent_before_invoking_docker(
|
|
fleet_module, monkeypatch, tmp_path
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
monkeypatch.setattr(fleet_module, "STATE_ROOT", tmp_path)
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
|
|
def observe_durable_intent(args, timeout=20):
|
|
assert args == [
|
|
"docker",
|
|
"restart",
|
|
"--timeout",
|
|
"30",
|
|
"itworx-validation-runner-2",
|
|
]
|
|
assert timeout == 45
|
|
persisted = service._events()
|
|
assert persisted[-1] == {
|
|
"at": persisted[-1]["at"],
|
|
"action": "restart",
|
|
"target": "itworx-validation-runner-2",
|
|
"reason": "stale_scheduler_assignment",
|
|
"result": "intent",
|
|
"ageSeconds": 900,
|
|
}
|
|
return ""
|
|
|
|
monkeypatch.setattr(service, "_run", observe_durable_intent)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert result["schedulerRecovery"]["status"] == "requested"
|
|
|
|
|
|
def test_scheduler_recovery_refuses_restart_when_intent_cannot_be_persisted(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_safe_state_root", lambda: False)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda args, timeout=20: commands.append(args) or ""
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "failed",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"reason": "restart_intent_not_persisted",
|
|
}
|
|
|
|
|
|
def test_scheduler_recovery_refuses_restart_when_intent_write_raises_oserror(
|
|
fleet_module, monkeypatch, tmp_path
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(fleet_module, "STATE_ROOT", tmp_path)
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
fleet_module.os,
|
|
"open",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(
|
|
PermissionError("state storage is unwritable")
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda args, timeout=20: commands.append(args) or ""
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "failed",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"reason": "restart_intent_not_persisted",
|
|
}
|
|
|
|
|
|
def test_scheduler_recovery_keeps_intent_when_outcome_write_fails(
|
|
fleet_module, monkeypatch, tmp_path
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
event_open_count = 0
|
|
real_open = fleet_module.os.open
|
|
monkeypatch.setattr(fleet_module, "STATE_ROOT", tmp_path)
|
|
monkeypatch.setattr(fleet_module, "RECOVERY_BUDGET", 1)
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": True,
|
|
"health": "healthy",
|
|
},
|
|
)
|
|
|
|
def fail_outcome_open(path, flags, mode=0o777):
|
|
nonlocal event_open_count
|
|
if Path(path).name.startswith(".events."):
|
|
event_open_count += 1
|
|
if event_open_count == 2:
|
|
raise PermissionError("outcome storage is unwritable")
|
|
return real_open(path, flags, mode)
|
|
|
|
monkeypatch.setattr(fleet_module.os, "open", fail_outcome_open)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda args, timeout=20: commands.append(args) or ""
|
|
)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == [
|
|
["docker", "restart", "--timeout", "30", "itworx-validation-runner-2"]
|
|
]
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "requested",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
}
|
|
assert [event["action"] for event in service._events()] == ["restart"]
|
|
assert service._recovery_available("itworx-validation-runner-2") is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("running", "health"),
|
|
[
|
|
(True, "starting"),
|
|
(True, "none"),
|
|
(True, "unhealthy"),
|
|
(False, "healthy"),
|
|
],
|
|
)
|
|
def test_scheduler_recovery_rechecks_strict_runner_health_before_restart(
|
|
fleet_module, monkeypatch, running, health
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_container",
|
|
lambda _target: {
|
|
"name": "itworx-validation-runner-2",
|
|
"running": running,
|
|
"health": health,
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda args, timeout=20: commands.append(args) or ""
|
|
)
|
|
monkeypatch.setattr(service, "_record", lambda **_event: True)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "ineligible",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"reason": "docker_health_not_healthy",
|
|
}
|
|
|
|
|
|
def test_scheduler_recovery_rejects_duplicate_fresh_inspect_results(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[tuple[list[str], int]] = []
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
|
|
def duplicate_inspect(args, timeout=20):
|
|
commands.append((args, timeout))
|
|
return "[{}, {}]"
|
|
|
|
monkeypatch.setattr(service, "_run", duplicate_inspect)
|
|
monkeypatch.setattr(service, "_record", lambda **_event: True)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == [
|
|
(["docker", "inspect", "itworx-validation-runner-2"], 20)
|
|
]
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "ineligible",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"reason": "docker_health_not_healthy",
|
|
}
|
|
|
|
|
|
def test_scheduler_recovery_rejects_runner_without_healthy_docker_state(
|
|
fleet_module, monkeypatch
|
|
):
|
|
service = fleet_module.RunnerFleetService()
|
|
snapshot = _healthy_snapshot()
|
|
next(
|
|
runner
|
|
for runner in snapshot["topology"]["runners"]
|
|
if runner["name"] == "itworx-validation-runner-2"
|
|
)["health"] = "unhealthy"
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", lambda: snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(
|
|
service, "_run", lambda args, timeout=20: commands.append(args) or ""
|
|
)
|
|
monkeypatch.setattr(service, "_record", lambda **_event: None)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"] == {
|
|
"status": "ineligible",
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"reason": "docker_health_not_healthy",
|
|
}
|
|
|
|
|
|
def test_scheduler_recovery_refuses_unknown_target(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "Gitea",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: True)
|
|
monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **_event: None)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"]["status"] == "rejected"
|
|
|
|
|
|
def test_scheduler_recovery_obeys_existing_recovery_budget(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_scheduler_recovery_candidate",
|
|
lambda: {
|
|
"target": "itworx-validation-runner-2",
|
|
"jobId": 22931,
|
|
"label": "linux-validation",
|
|
"ageSeconds": 900,
|
|
"reason": "stale_scheduler_assignment",
|
|
},
|
|
)
|
|
monkeypatch.setattr(service, "_recovery_available", lambda _target: False)
|
|
monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **_event: None)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"]["status"] == "budget_exhausted"
|
|
|
|
|
|
def test_scheduler_recovery_does_nothing_without_stale_candidate(fleet_module, monkeypatch):
|
|
service = fleet_module.RunnerFleetService()
|
|
commands: list[list[str]] = []
|
|
monkeypatch.setattr(service, "snapshot", _healthy_snapshot)
|
|
monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None)
|
|
monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "")
|
|
monkeypatch.setattr(service, "_record", lambda **_event: None)
|
|
|
|
result = service.automatic_cycle()
|
|
|
|
assert commands == []
|
|
assert result["schedulerRecovery"] == {"status": "not_needed"}
|