145 lines
4.6 KiB
Python
145 lines
4.6 KiB
Python
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
|