claude
38ba9c1b50
Initial PoC: find_protection, tagged checkpoints, FLR, mutating catalog. Lab 10.9 status enums (0=Initializing, 1=MeetingSLA). Credentials stay in gitignored config.json.
104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
from zerto_rewind_mcp.protection import find_from_rows, group_vm_rows
|
|
from zerto_rewind_mcp.status import (
|
|
BITMAP_SYNC,
|
|
DELTA_SYNC,
|
|
INITIAL_SYNC,
|
|
INITIALIZING,
|
|
MEETING_SLA,
|
|
NOT_MEETING_SLA,
|
|
)
|
|
|
|
|
|
def _row(vm_id, vm_name, vpg_id, vpg_name, status=MEETING_SLA, sub=0):
|
|
return {
|
|
"VmIdentifier": vm_id,
|
|
"VmName": vm_name,
|
|
"VpgIdentifier": vpg_id,
|
|
"VpgName": vpg_name,
|
|
"Status": status,
|
|
"SubStatus": sub,
|
|
}
|
|
|
|
|
|
def test_none():
|
|
result = find_from_rows("web01", [])
|
|
assert result.outcome == "none"
|
|
assert "refuse" in result.message.lower()
|
|
|
|
|
|
def test_unique_two_vpgs():
|
|
rows = [
|
|
_row("vm-1", "web01", "vpg-local", "local-bu"),
|
|
_row("vm-1", "web01", "vpg-dr", "dr-remote"),
|
|
]
|
|
result = find_from_rows("web01", rows)
|
|
assert result.outcome == "ok"
|
|
assert result.vm is not None
|
|
assert result.vm.vm_identifier == "vm-1"
|
|
assert len(result.vm.vpgs) == 2
|
|
assert len(result.taggable_vpgs) == 2
|
|
|
|
|
|
def test_ambiguous_two_vms():
|
|
rows = [
|
|
_row("vm-1", "web01", "vpg-a", "a"),
|
|
_row("vm-2", "web01", "vpg-b", "b"),
|
|
]
|
|
result = find_from_rows("web01", rows)
|
|
assert result.outcome == "ambiguous"
|
|
assert len(result.matches) == 2
|
|
assert result.taggable_vpgs == []
|
|
|
|
|
|
def test_skip_syncing_vpg():
|
|
rows = [
|
|
_row("vm-1", "app", "vpg-ok", "ok", status=MEETING_SLA, sub=0),
|
|
_row("vm-1", "app", "vpg-sync", "syncing", status=MEETING_SLA, sub=INITIAL_SYNC),
|
|
_row("vm-1", "app", "vpg-delta", "delta", status=MEETING_SLA, sub=DELTA_SYNC),
|
|
_row("vm-1", "app", "vpg-bitmap", "bitmap", status=MEETING_SLA, sub=BITMAP_SYNC),
|
|
]
|
|
result = find_from_rows("app", rows)
|
|
assert result.outcome == "ok"
|
|
names = {v.vpg_name for v in result.taggable_vpgs}
|
|
assert names == {"ok"}
|
|
assert "InitialSync" in result.message
|
|
|
|
|
|
def test_not_meeting_sla_is_taggable():
|
|
rows = [_row("vm-1", "app", "vpg-a", "a", status=NOT_MEETING_SLA, sub=0)]
|
|
result = find_from_rows("app", rows)
|
|
assert result.outcome == "ok"
|
|
assert len(result.taggable_vpgs) == 1
|
|
|
|
|
|
def test_initializing_not_taggable():
|
|
rows = [_row("vm-1", "app", "vpg-a", "a", status=INITIALIZING, sub=INITIAL_SYNC)]
|
|
result = find_from_rows("app", rows)
|
|
assert result.outcome == "ok"
|
|
assert result.taggable_vpgs == []
|
|
assert "refuse" in result.message.lower()
|
|
|
|
|
|
def test_all_syncing_ok_but_no_taggable():
|
|
rows = [_row("vm-1", "app", "vpg-a", "a", status=MEETING_SLA, sub=INITIAL_SYNC)]
|
|
result = find_from_rows("app", rows)
|
|
assert result.outcome == "ok"
|
|
assert result.taggable_vpgs == []
|
|
assert "refuse" in result.message.lower()
|
|
|
|
|
|
def test_camel_case_keys():
|
|
rows = [
|
|
{
|
|
"vmIdentifier": "id-9",
|
|
"vmName": "db01",
|
|
"vpgIdentifier": "v1",
|
|
"vpgName": "db-vpg",
|
|
"status": 1,
|
|
"subStatus": 0,
|
|
}
|
|
]
|
|
grouped = group_vm_rows(rows)
|
|
assert grouped[0].vm_name == "db01"
|
|
assert grouped[0].vpgs[0].can_tag
|