from __future__ import annotations

import importlib.util
import sys
import types
from datetime import UTC, datetime
from pathlib import Path

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_test", FLEET_PATH)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    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 test_dind_uses_private_unix_socket_only_for_trusted_active_specialization(
    fleet_module, monkeypatch
):
    service = fleet_module.RunnerFleetService()
    commands = []
    monkeypatch.setattr(
        service,
        "_run",
        lambda args, **_kwargs: commands.append(args) or "",
    )
    specialization = sys.modules["runner_fleet_specialize"]

    service._dind(["info"])
    assert "DOCKER_HOST=tcp://127.0.0.1:2375" in commands[-1]

    monkeypatch.setattr(specialization, "specialization_state", lambda: "active")
    service._dind(["info"])
    assert f"DOCKER_HOST={fleet_module.DIND_DOCKER_HOST}" in commands[-1]

    for state in ("legacy", "blocked", "in_progress"):
        monkeypatch.setattr(specialization, "specialization_state", lambda value=state: value)
        service._dind(["info"])
        assert "DOCKER_HOST=tcp://127.0.0.1:2375" in commands[-1]


def test_cleanup_candidates_exclude_unknown_attached_and_labeled_networks(
    fleet_module, monkeypatch
):
    service = fleet_module.RunnerFleetService()
    now = 1_800_000_000
    monkeypatch.setattr(fleet_module.time, "time", lambda: now)
    old = "2026-01-01T00:00:00Z"
    monkeypatch.setattr(
        service,
        "_network_inventory",
        lambda: (
            [
                {
                    "id": "a" * 64,
                    "name": "GITEA-ACTIONS-TASK-12-WORKFLOW-ci-JOB-test-network",
                    "created": old,
                    "attached": 0,
                    "labels": {},
                },
                {
                    "id": "b" * 64,
                    "name": "production_default",
                    "created": old,
                    "attached": 0,
                    "labels": {},
                },
                {
                    "id": "c" * 64,
                    "name": "GITEA-ACTIONS-TASK-13-WORKFLOW-ci-JOB-test-network",
                    "created": old,
                    "attached": 1,
                    "labels": {},
                },
                {
                    "id": "d" * 64,
                    "name": "GITEA-ACTIONS-TASK-14-WORKFLOW-ci-JOB-test-network",
                    "created": old,
                    "attached": 0,
                    "labels": {"com.example.keep": "true"},
                },
            ],
            2048,
        ),
    )

    candidates = service.cleanup_candidates()

    assert [candidate["id"] for candidate in candidates] == ["a" * 64]
    assert candidates[0]["reason"] == "empty_gitea_actions_task_network_in_dedicated_dind"


def test_automatic_cycle_only_deletes_exact_preview_candidates(fleet_module, monkeypatch):
    service = fleet_module.RunnerFleetService()
    candidate = {
        "id": "a" * 64,
        "name": "GITEA-ACTIONS-TASK-12-WORKFLOW-ci-JOB-test-network",
        "ageSeconds": 3600,
        "reason": "empty_gitea_actions_task_network_in_dedicated_dind",
    }
    snapshot = {
        "status": "healthy",
        "capacity": _healthy_capacity_snapshot(),
        "networks": {"cleanupCandidates": [candidate]},
        "storage": {"status": "healthy"},
        "topology": {
            "dind": {
                "name": "gitea-runner-dind",
                "running": True,
                "health": "healthy",
                "startedAt": "2026-09-16T18:00:00Z",
                "restartPolicy": "unless-stopped",
            },
            "runners": [],
        },
    }
    calls: list[list[str]] = []
    monkeypatch.setattr(service, "snapshot", lambda: snapshot)
    monkeypatch.setattr(service, "_dind", lambda args, timeout=20: calls.append(args) or "")
    monkeypatch.setattr(service, "_record", lambda **_kwargs: None)

    result = service.automatic_cycle()

    assert result["removedNetworks"] == 1
    assert calls == [["network", "rm", "a" * 64]]


def test_recovery_budget_stops_restart_loops(fleet_module, monkeypatch):
    service = fleet_module.RunnerFleetService()
    now = 1_800_000_000
    monkeypatch.setattr(fleet_module.time, "time", lambda: now)
    monkeypatch.setattr(
        service,
        "_events",
        lambda: [
            {
                "action": "restart",
                "target": "itworx-security-runner",
                "at": "2027-01-15T08:00:00Z",
            }
            for _ in range(fleet_module.RECOVERY_BUDGET)
        ],
    )

    assert service._recovery_available("itworx-security-runner") is False


def test_fresh_starting_runner_is_not_restarted_before_health_grace_expires(
    fleet_module,
):
    fresh = {
        "name": "itworx-validation-runner-2",
        "running": True,
        "health": "starting",
        "startedAt": "1970-01-01T00:16:00Z",
    }
    stale = {**fresh, "startedAt": "1970-01-01T00:13:00Z"}

    assert fleet_module._container_is_healthy(fresh, now=1_000)
    assert not fleet_module._container_is_healthy(stale, now=1_000)


def test_automatic_cycle_starts_only_an_exited_known_runner(fleet_module, monkeypatch):
    service = fleet_module.RunnerFleetService()
    snapshot = {
        "status": "healthy",
        "capacity": _healthy_capacity_snapshot(),
        "networks": {"cleanupCandidates": []},
        "storage": {"status": "healthy"},
        "topology": {
            "dind": {
                "name": "gitea-runner-dind",
                "running": True,
                "health": "healthy",
                "startedAt": "2026-09-16T18:00:00Z",
                "restartPolicy": "unless-stopped",
            },
            "runners": [
                {
                    "name": "itworx-validation-runner-2",
                    "running": False,
                    "health": "unhealthy",
                },
                {
                    "name": "itworx-security-runner",
                    "running": True,
                    "health": "unhealthy",
                },
            ]
        },
    }
    commands: list[list[str]] = []
    events: list[dict[str, object]] = []
    monkeypatch.setattr(service, "snapshot", lambda: snapshot)
    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: events.append(event))

    service.automatic_cycle()

    assert commands == [["docker", "start", "itworx-validation-runner-2"]]
    assert {event["action"] for event in events} == {"start", "restart.deferred"}


@pytest.mark.parametrize("specialization", ["active", "blocked"])
def test_legacy_automatic_cycle_never_observes_or_mutates_after_specialization(
    fleet_module, monkeypatch, specialization
):
    service = fleet_module.RunnerFleetService()
    monkeypatch.setitem(
        sys.modules,
        "runner_fleet_specialize",
        types.SimpleNamespace(specialization_state=lambda: specialization),
    )
    monkeypatch.setattr(service, "snapshot", lambda: pytest.fail("legacy snapshot ran"))
    monkeypatch.setattr(service, "_run", lambda *_args, **_kwargs: pytest.fail("legacy mutation ran"))
    result = service.automatic_cycle()

    assert result["status"] == "deferred"
    assert result["reason"] == "runner_fleet_specialization_owns_recovery"