"""Auto-upgrade trigger in the heartbeat handler (CAURA-444).

Pure unit-tests for the helpers — no DB / storage roundtrip needed.
The heartbeat handler stitches them together; integration coverage
lives in the existing fleet route tests.
"""

from __future__ import annotations

import pytest

from core_api.routes import fleet as fleet_mod


# ---------------------------------------------------------------------------
# _semver_lt
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "a,b,expected",
    [
        ("2.3.0", "2.4.0", True),
        ("2.4.0", "2.3.0", False),
        ("2.4.0", "2.4.0", False),
        ("2.4", "2.4.1", True),    # zero-pad
        ("2.4.1", "2.4", False),
        ("1.10.0", "1.9.0", False),  # int compare, not lex
        ("1.9.0", "1.10.0", True),
        # Falsy / unparseable inputs never produce a "newer" verdict.
        ("", "2.4.0", False),
        ("2.4.0", "", False),
        (None, "2.4.0", False),
        ("2.4.0", None, False),
        ("dev", "2.4.0", False),
        ("not.a.version", "2.4.0", False),
        # Pure-separator strings with no digits filter to an empty list
        # via `[int(x) for x in s.split(".") if x]`. Without the
        # empty-list guard they'd zero-pad to [0,0,0] and falsely
        # compare as older than any real version, triggering a spurious
        # auto-upgrade. Locked here.
        ("...", "2.4.0", False),
        (".", "2.4.0", False),
        ("....", "2.4.0", False),
    ],
)
def test_semver_lt(a, b, expected):
    assert fleet_mod._semver_lt(a, b) == expected


# ---------------------------------------------------------------------------
# Known-broken denylist (transition guard for v2.3.0)
# ---------------------------------------------------------------------------


def test_v2_3_0_is_in_known_broken_set():
    """The 2.3.0 → 2.4.0 transition is broken (drift-1 + drift-2 in
    the deploy machinery). The denylist must contain it; auto-upgrade
    on these nodes would loop. Operators must manually upgrade.
    """
    assert "2.3.0" in fleet_mod.KNOWN_BROKEN_DEPLOY_VERSIONS


def test_known_broken_set_is_frozen():
    """A frozenset means a runtime mistake (.add()) raises rather than
    silently breaking the safety net.
    """
    assert isinstance(fleet_mod.KNOWN_BROKEN_DEPLOY_VERSIONS, frozenset)


# ---------------------------------------------------------------------------
# _auto_upgrade_enabled_for_tenant
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_auto_upgrade_enabled_default_true(monkeypatch):
    """No tenant override → enabled (the global default)."""
    async def _fake(_db, _tid):
        return {}  # no override

    monkeypatch.setattr(
        "core_api.routes.fleet.get_raw_settings", _fake
    )
    assert (
        await fleet_mod._auto_upgrade_enabled_for_tenant(None, "tenant-1")
        is True
    )


@pytest.mark.asyncio
async def test_auto_upgrade_enabled_override_false(monkeypatch):
    """Tenant override `memclaw.auto_upgrade_enabled = false` → disabled."""
    async def _fake(_db, _tid):
        return {"memclaw": {"auto_upgrade_enabled": False}}

    monkeypatch.setattr(
        "core_api.routes.fleet.get_raw_settings", _fake
    )
    assert (
        await fleet_mod._auto_upgrade_enabled_for_tenant(None, "tenant-1")
        is False
    )


@pytest.mark.asyncio
async def test_auto_upgrade_enabled_override_true(monkeypatch):
    """Tenant override `memclaw.auto_upgrade_enabled = true` → enabled."""
    async def _fake(_db, _tid):
        return {"memclaw": {"auto_upgrade_enabled": True}}

    monkeypatch.setattr(
        "core_api.routes.fleet.get_raw_settings", _fake
    )
    assert (
        await fleet_mod._auto_upgrade_enabled_for_tenant(None, "tenant-1")
        is True
    )


@pytest.mark.asyncio
async def test_auto_upgrade_fail_open_on_settings_error(monkeypatch):
    """If settings resolve raises, default to enabled (cooldown machinery
    on the plugin side prevents loops in the worst case).
    """
    async def _fake(_db, _tid):
        raise RuntimeError("settings backend down")

    monkeypatch.setattr(
        "core_api.routes.fleet.get_raw_settings", _fake
    )
    assert (
        await fleet_mod._auto_upgrade_enabled_for_tenant(None, "tenant-1")
        is True
    )


# ---------------------------------------------------------------------------
# _maybe_queue_auto_upgrade
# ---------------------------------------------------------------------------


class _FakeStorage:
    """Captures create_command + get_pending_commands calls."""

    def __init__(self, pending=None):
        self.pending_commands = pending or []
        self.created_commands: list[dict] = []

    async def get_pending_commands(self, _tenant_id, _node_name):
        return list(self.pending_commands)

    async def create_command(self, data):
        self.created_commands.append(data)
        return {"id": "fake-cmd-1"}


def _body(plugin_version="2.3.0", deploy_blocked_until=None):
    """Mint a HeartbeatIn-shaped Pydantic model with overrides."""
    return fleet_mod.HeartbeatIn(
        tenant_id="tenant-1",
        node_name="node-a",
        plugin_version=plugin_version,
        deploy_blocked_until=deploy_blocked_until,
    )


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_skips_for_known_broken(monkeypatch):
    """Plugin v2.3.0 → no deploy command (loop-prevention denylist)."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(db=None, sc=sc, body=_body("2.3.0"), pending_commands=sc.pending_commands, node_id="node-uuid-1")
    assert sc.created_commands == []


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_skips_when_current(monkeypatch):
    """plugin_version == VERSION → no deploy."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(db=None, sc=sc, body=_body("2.4.0"), pending_commands=sc.pending_commands, node_id="node-uuid-1")
    assert sc.created_commands == []


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_skips_when_newer(monkeypatch):
    """plugin_version > VERSION (dev install) → no downgrade."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(db=None, sc=sc, body=_body("2.5.0-dev"), pending_commands=sc.pending_commands, node_id="node-uuid-1")
    assert sc.created_commands == []


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_skips_when_blocked(monkeypatch):
    """deploy_blocked_until in the future → skip (cooldown signal)."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    # 1 hour in the future — well within the 7-day MAX_BLOCK_MS cap.
    from datetime import UTC, datetime
    near_future_ms = int(datetime.now(UTC).timestamp() * 1000) + 3600_000
    await fleet_mod._maybe_queue_auto_upgrade(
        db=None,
        sc=sc,
        body=_body("2.3.5", deploy_blocked_until=near_future_ms),
        pending_commands=sc.pending_commands, node_id="node-uuid-1",
    )
    assert sc.created_commands == []


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_ignores_absurd_block_cap(monkeypatch):
    """deploy_blocked_until > MAX_BLOCK_MS (7 days) is NOT honored.

    A misbehaving / malicious plugin could pin
    deploy_blocked_until = Number.MAX_SAFE_INTEGER to DoS its own
    upgrade path. The cap forces such absurd values to fall through
    to the normal "queue deploy" branch so the next heartbeat retries
    the upgrade.
    """
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")
    # Disable the pre-manifest-aware floor for this test — we're
    # exercising the absurd-block-cap branch, not the new gate.
    monkeypatch.setattr(fleet_mod, "MIN_AUTO_DEPLOY_PLUGIN_VERSION", "0.0.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    absurd_future_ms = 99999999999999  # year 5138
    await fleet_mod._maybe_queue_auto_upgrade(
        db=None,
        sc=sc,
        body=_body("2.3.5", deploy_blocked_until=absurd_future_ms),
        pending_commands=sc.pending_commands, node_id="node-uuid-1",
    )
    # Absurd value ignored → deploy IS queued (normal path).
    assert len(sc.created_commands) == 1
    assert sc.created_commands[0]["command"] == "deploy"


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_skips_when_disabled(monkeypatch):
    """tenant has auto_upgrade_enabled = false → skip."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")

    async def _disabled(_db, _tid):
        return False

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _disabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(db=None, sc=sc, body=_body("2.3.5"), pending_commands=sc.pending_commands, node_id="node-uuid-1")
    assert sc.created_commands == []


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_skips_when_already_pending(monkeypatch):
    """An existing pending deploy → skip (don't double-queue)."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage(pending=[{"command": "deploy", "payload": {}}])
    await fleet_mod._maybe_queue_auto_upgrade(db=None, sc=sc, body=_body("2.3.5"), pending_commands=sc.pending_commands, node_id="node-uuid-1")
    assert sc.created_commands == []


# ---------------------------------------------------------------------------
# MIN_AUTO_DEPLOY_PLUGIN_VERSION — pre-manifest-aware floor (CAURA-000)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
@pytest.mark.parametrize("stale_version", ["0.98.4", "2.0.0", "2.3.0", "2.4.0", "2.5.0"])
async def test_maybe_queue_auto_upgrade_skips_pre_manifest_aware(monkeypatch, stale_version):
    """Pre-manifest-aware floor blocks every released plugin tag.

    The hardcoded fallback file list in those releases doesn't include
    files added in later releases (e.g. keystones.ts), so auto-deploy
    leaves the plugin unable to load. Same recovery path as
    KNOWN_BROKEN_DEPLOY_VERSIONS — operator runs the install one-liner.
    """
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.6.0")
    monkeypatch.setattr(fleet_mod, "MIN_AUTO_DEPLOY_PLUGIN_VERSION", "2.6.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(
        db=None, sc=sc, body=_body(stale_version),
        pending_commands=sc.pending_commands, node_id="node-uuid-1",
    )
    assert sc.created_commands == [], (
        f"Expected no deploy queued for pre-manifest-aware {stale_version!r}; "
        f"got {sc.created_commands!r}"
    )


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_allowed_at_or_above_floor(monkeypatch):
    """A plugin at the floor exact version OR above is eligible for
    auto-deploy (subject to other gates). Asserts the floor is < rather
    than <=, so floor versions can still be pushed forward."""
    # Target is higher than floor — there's a real upgrade to issue.
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.6.1")
    monkeypatch.setattr(fleet_mod, "MIN_AUTO_DEPLOY_PLUGIN_VERSION", "2.6.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(
        db=None, sc=sc, body=_body("2.6.0"),
        pending_commands=sc.pending_commands, node_id="node-uuid-1",
    )
    assert len(sc.created_commands) == 1, (
        f"Expected deploy queued for floor version 2.6.0; got {sc.created_commands!r}"
    )
    assert sc.created_commands[0]["command"] == "deploy"


def test_min_auto_deploy_constant_shape():
    """The floor is a string-shaped semver and stays a module-level
    constant so operators can find / bump it without searching the file.
    """
    assert isinstance(fleet_mod.MIN_AUTO_DEPLOY_PLUGIN_VERSION, str)
    parts = fleet_mod.MIN_AUTO_DEPLOY_PLUGIN_VERSION.split(".")
    assert len(parts) >= 2
    for p in parts:
        assert p.isdigit(), f"Non-numeric segment in floor: {p!r}"


def test_has_recent_deploy_from_list_handles_non_dict_entries():
    """Pre-fix ``(c or {}).get(...)`` raised AttributeError on a truthy
    non-dict element (string, number, list). The isinstance guard
    treats unexpected types as "not a deploy" rather than crashing
    the heartbeat.
    """
    # Garbage from a misbehaving storage backend — strings, numbers,
    # bare lists. All must be silently skipped, not crash.
    assert (
        fleet_mod._has_recent_deploy_command_from_list(
            ["not-a-dict", 42, ["nested"], None, {}]
        )
        is False
    )
    # Mixed: garbage + a valid deploy → still True (we don't miss the
    # legitimate entry just because adjacent ones are malformed).
    assert (
        fleet_mod._has_recent_deploy_command_from_list(
            ["garbage", {"command": "deploy"}, 99]
        )
        is True
    )


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_queues_deploy_for_old_version(monkeypatch):
    """Happy path: enabled, not blocked, no pending, valid old version → queue."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")
    # Disable the pre-manifest-aware floor for this test — happy-path coverage
    # is for an already-manifest-aware client; the new floor has its own tests.
    monkeypatch.setattr(fleet_mod, "MIN_AUTO_DEPLOY_PLUGIN_VERSION", "0.0.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(db=None, sc=sc, body=_body("2.3.5"), pending_commands=sc.pending_commands, node_id="node-uuid-1")
    assert len(sc.created_commands) == 1
    cmd = sc.created_commands[0]
    assert cmd["command"] == "deploy"
    assert cmd["payload"]["target_version"] == "2.4.0"
    assert cmd["tenant_id"] == "tenant-1"
    # Storage expects ``node_id`` (UUID FK to fleet_nodes.id), not ``node_name``.
    # Pre-fix the queue silently 500'd because ``_filter_fields`` dropped the
    # ``node_name`` key and ``node_id`` was NULL — this assertion locks the fix.
    assert cmd["node_id"] == "node-uuid-1"
    assert "node_name" not in cmd


@pytest.mark.asyncio
async def test_maybe_queue_auto_upgrade_skips_when_plugin_version_missing(monkeypatch):
    """No plugin_version on payload (very old plugin) → skip."""
    monkeypatch.setattr(fleet_mod, "MIN_RECOMMENDED_PLUGIN_VERSION", "2.4.0")

    async def _enabled(_db, _tid):
        return True

    monkeypatch.setattr(fleet_mod, "_auto_upgrade_enabled_for_tenant", _enabled)
    sc = _FakeStorage()
    await fleet_mod._maybe_queue_auto_upgrade(
        db=None,
        sc=sc,
        body=fleet_mod.HeartbeatIn(tenant_id="tenant-1", node_name="node-a"),
        pending_commands=sc.pending_commands, node_id="node-uuid-1",
    )
    assert sc.created_commands == []


# ---------------------------------------------------------------------------
# HeartbeatIn.recall_metrics size cap
# ---------------------------------------------------------------------------


def test_recall_metrics_under_cap_accepted():
    """Normal-sized blob from the plugin passes through unchanged."""
    body = fleet_mod.HeartbeatIn(
        tenant_id="tenant-1",
        node_name="node-a",
        recall_metrics={
            "calls_total": 142,
            "skipped_total": 89,
            "skipped_by_reason": {
                "trivial-ping": 41,
                "below-threshold": 33,
                "slash-command": 8,
                "explicit-recall-trigger": 7,
            },
        },
    )
    assert body.recall_metrics is not None
    assert body.recall_metrics["calls_total"] == 142


def test_recall_metrics_over_cap_rejected():
    """A misbehaving plugin sending a huge counter blob is rejected at
    the API boundary — protects nodes.metadata from unbounded growth.
    """
    import pytest as _pt
    from pydantic import ValidationError

    huge = {"calls_total": 1, "skipped_by_reason": {f"reason-{i}": i for i in range(500)}}
    # 500 keys × ~20 bytes ≈ 10 KB → well over the 4 KB cap.
    with _pt.raises(ValidationError, match="exceeds 4 KB limit"):
        fleet_mod.HeartbeatIn(
            tenant_id="tenant-1",
            node_name="node-a",
            recall_metrics=huge,
        )


def test_recall_metrics_none_accepted():
    """Omitted field is fine (back-compat with older plugins)."""
    body = fleet_mod.HeartbeatIn(tenant_id="tenant-1", node_name="node-a")
    assert body.recall_metrics is None


# ---------------------------------------------------------------------------
# HeartbeatIn.deploy_blocked_until validation
# ---------------------------------------------------------------------------


def test_deploy_blocked_until_positive_accepted():
    """Positive epoch-ms (future) is the normal case."""
    body = fleet_mod.HeartbeatIn(
        tenant_id="tenant-1",
        node_name="node-a",
        deploy_blocked_until=1_700_000_000_000,
    )
    assert body.deploy_blocked_until == 1_700_000_000_000


def test_deploy_blocked_until_none_accepted():
    """Omitted field is fine (older plugins, or no cooldown)."""
    body = fleet_mod.HeartbeatIn(tenant_id="tenant-1", node_name="node-a")
    assert body.deploy_blocked_until is None


def test_deploy_blocked_until_zero_rejected():
    """Zero is not a real cooldown timestamp. Reject at the API boundary
    so it doesn't land in nodes.metadata and mislead operators.
    """
    import pytest as _pt
    from pydantic import ValidationError

    with _pt.raises(ValidationError):
        fleet_mod.HeartbeatIn(
            tenant_id="tenant-1",
            node_name="node-a",
            deploy_blocked_until=0,
        )


def test_deploy_blocked_until_negative_rejected():
    """Negative values are nonsensical — reject."""
    import pytest as _pt
    from pydantic import ValidationError

    with _pt.raises(ValidationError):
        fleet_mod.HeartbeatIn(
            tenant_id="tenant-1",
            node_name="node-a",
            deploy_blocked_until=-1,
        )
