475 lines
19 KiB
Python
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
|