zerto-ai-rewind/tests/test_checkpoints.py

from datetime import UTC, datetime

import pytest

from zerto_rewind_mcp.checkpoints import (
    TAG_MAX_LEN,
    checkpoint_id,
    checkpoint_tag,
    make_tag,
)

WHEN = datetime(2026, 9, 21, 15, 4, 5, tzinfo=UTC)


def test_tag_describes_agent_and_action():
    tag = make_tag(
        "claude",
        "chg-99",
        WHEN,
        action="edit /home/justin/app-config.yaml",
        vm_name="jp-ubuntu",
    )
    assert tag == (
        "ai:claude | edit /home/justin/app-config.yaml | vm=jp-ubuntu "
        "| change=chg-99 | 20260921T150405Z"
    )


def test_tag_without_action_or_vm():
    assert make_tag("codex", "chg-99", WHEN) == "ai:codex | change=chg-99 | 20260921T150405Z"


def test_tag_keeps_separators_unambiguous():
    # ';' is what Zerto appends to its own tags; '|' is our field separator.
    tag = make_tag("a;b", "c|d", WHEN, action="rm -rf /tmp;x", vm_name="v|m")
    assert tag.count(" | ") == 4
    assert ";" not in tag


def test_tag_collapses_whitespace_and_caps_length():
    tag = make_tag("agent one", "chg 1", WHEN, action="do\n  many   things")
    assert "\n" not in tag
    assert "do many things" in tag
    long_tag = make_tag("a" * 200, "b" * 200, WHEN, action="c" * 400)
    assert len(long_tag) <= TAG_MAX_LEN


def test_checkpoint_row_keys():
    row = {"CheckpointId": "cp-1", "Tag": "ai:x:y:z", "Timestamp": 1}
    assert checkpoint_id(row) == "cp-1"
    assert checkpoint_tag(row) == "ai:x:y:z"
    row2 = {"checkpointId": "cp-2", "tag": "t"}
    assert checkpoint_id(row2) == "cp-2"
    assert checkpoint_tag(row2) == "t"


def _rows(*offsets_seconds):
    from datetime import timedelta

    base = datetime(2026, 9, 22, 12, 0, 0, tzinfo=UTC)
    return [
        {"TimeStamp": (base + timedelta(seconds=o)).isoformat().replace("+00:00", "Z")}
        for o in offsets_seconds
    ]


def test_cadence_measures_the_median_gap():
    from zerto_rewind_mcp.checkpoints import cadence_seconds

    assert cadence_seconds(_rows(0, 5, 10, 15, 20)) == 5.0
    assert cadence_seconds(_rows(0, 60, 120, 180)) == 60.0
    # one irregular gap must not drag the answer around
    assert cadence_seconds(_rows(0, 5, 10, 400, 405, 410)) == 5.0


def test_cadence_is_none_when_unmeasurable():
    from zerto_rewind_mcp.checkpoints import cadence_seconds

    assert cadence_seconds([]) is None
    assert cadence_seconds([{"TimeStamp": "not-a-date"}]) is None
    assert cadence_seconds(_rows(0)) is None  # one checkpoint gives no gap


def test_tag_wait_budget_covers_every_measured_platform():
    """The three cadences measured in one estate, and what each actually needed."""
    from zerto_rewind_mcp.checkpoints import tag_wait_budget

    for cadence, observed_visibility in ((5.0, 4.0), (60.0, 34.0), (630.0, 128.0)):
        budget, interval = tag_wait_budget(cadence)
        assert budget > observed_visibility, (
            f"cadence {cadence}s budgets {budget}s but the tag took {observed_visibility}s"
        )
        assert interval >= 1.5


def test_tag_wait_budget_is_clamped_at_both_ends():
    from zerto_rewind_mcp.checkpoints import (
        DEFAULT_TAG_TIMEOUT_S,
        MAX_TAG_TIMEOUT_S,
        MIN_TAG_TIMEOUT_S,
        tag_wait_budget,
    )

    assert tag_wait_budget(0.1)[0] == MIN_TAG_TIMEOUT_S  # absurdly fast VPG
    assert tag_wait_budget(100_000)[0] == MAX_TAG_TIMEOUT_S  # absurdly slow one
    assert tag_wait_budget(None)[0] == DEFAULT_TAG_TIMEOUT_S
    assert tag_wait_budget(None)[1] == 1.5


def test_wait_for_tag_returns_immediately_when_already_present():
    import asyncio

    from zerto_rewind_mcp.checkpoints import wait_for_tag

    class Client:
        def __init__(self):
            self.calls = 0

        async def list_checkpoints(self, vpg):
            self.calls += 1
            return [{"Tag": "ai:x", "CheckpointId": "7"}]

    c = Client()
    row = asyncio.run(wait_for_tag(c, "vpg", "ai:x"))
    assert row["CheckpointId"] == "7"
    assert c.calls == 1  # no sleep, no second poll


def test_wait_for_tag_error_names_the_measured_cadence():
    import asyncio

    from zerto_rewind_mcp.checkpoints import wait_for_tag
    from zerto_rewind_mcp.client import ZertoError

    class Client:
        async def list_checkpoints(self, vpg):
            return _rows(0, 60, 120, 180)  # 60s cadence, tag never appears

    with pytest.raises(ZertoError) as err:
        # explicit tiny timeout so the test does not actually wait 150s
        asyncio.run(wait_for_tag(Client(), "vpg", "ai:missing", timeout_s=0.01, interval_s=0.01))
    msg = str(err.value)
    assert "about every 60s" in msg
    assert "was short, not that the insert was rejected" in msg