300 lines
9.5 KiB
Python
300 lines
9.5 KiB
Python
from zerto_rewind_mcp.recover import (
|
|
download_token_from,
|
|
flr_row,
|
|
flr_status,
|
|
session_id_from,
|
|
)
|
|
|
|
|
|
def test_session_id_shapes():
|
|
assert session_id_from("abc") == "abc"
|
|
assert session_id_from({"sessionId": "s1"}) == "s1"
|
|
assert session_id_from({"Identifier": "s2"}) == "s2"
|
|
assert session_id_from({"FlrSessionIdentifier": "s3"}) == "s3"
|
|
|
|
|
|
def test_download_token_shapes():
|
|
assert download_token_from("tok") == "tok"
|
|
assert download_token_from({"downloadToken": "t2"}) == "t2"
|
|
assert download_token_from("v1/downloads/abc").startswith("v1/")
|
|
|
|
|
|
def test_flr_list_status():
|
|
payload = [
|
|
{
|
|
"FlrSessionStatus": "MountCompletedSuccessfully",
|
|
"FlrSessionIdentifier": "sess-1",
|
|
}
|
|
]
|
|
row = flr_row(payload)
|
|
assert flr_status(row).lower() == "mountcompletedsuccessfully"
|
|
|
|
|
|
def test_path_items_shapes():
|
|
from zerto_rewind_mcp.recover import path_items
|
|
|
|
assert path_items({"PathItems": [{"Path": "a"}]}) == [{"Path": "a"}]
|
|
assert path_items([{"Path": "b"}]) == [{"Path": "b"}]
|
|
assert path_items(None) == []
|
|
|
|
|
|
def test_decode_flr_path():
|
|
from zerto_rewind_mcp.recover import _decode_flr_path
|
|
|
|
assert _decode_flr_path("Volume2-Ext4%2fhome%2fjustin%2fapp-config.yaml") == (
|
|
"Volume2-Ext4/home/justin/app-config.yaml"
|
|
)
|
|
|
|
|
|
class _BrowseFake:
|
|
"""Fake FLR mount. root is the partition list; tree maps probe path -> rows."""
|
|
|
|
def __init__(self, root, tree):
|
|
self.root, self.tree, self.seen = root, tree, []
|
|
|
|
async def browse_flr(self, session_id, path="", recursive=False):
|
|
self.seen.append(path)
|
|
if path == "":
|
|
return {"PathItems": self.root}
|
|
return {"PathItems": self.tree.get(path, [])}
|
|
|
|
|
|
def _resolve(client, guest_path):
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp import recover
|
|
|
|
recover.PARTITION_POLL_INTERVAL_S = 0 # do not sleep in tests
|
|
return asyncio.run(recover.resolve_flr_path(client, "sess", guest_path))
|
|
|
|
|
|
def test_resolve_flr_path_linux_prepends_partition():
|
|
client = _BrowseFake(
|
|
[
|
|
{"Path": "Volume1-Unknown", "IsBrowsable": False},
|
|
{"Path": "Volume2-Ext4", "IsBrowsable": True},
|
|
],
|
|
{
|
|
"Volume2-Ext4/home/justin": [
|
|
{"Path": "Volume2-Ext4%2fhome%2fjustin%2f.bashrc", "Type": "File"},
|
|
{"Path": "Volume2-Ext4%2fhome%2fjustin%2fapp-config.yaml", "Type": "File"},
|
|
]
|
|
},
|
|
)
|
|
# returns the raw path browse gave us; download accepts it verbatim
|
|
assert _resolve(client, "/home/justin/app-config.yaml") == (
|
|
"Volume2-Ext4%2fhome%2fjustin%2fapp-config.yaml"
|
|
)
|
|
assert "Volume1-Unknown/home/justin" not in client.seen
|
|
|
|
|
|
def test_resolve_flr_path_windows_does_not_double_the_drive_letter():
|
|
# On Windows the drive letter IS the partition: C%3a decodes to "C:".
|
|
client = _BrowseFake(
|
|
[
|
|
{"Path": "C%3a", "IsBrowsable": True},
|
|
{"Path": "Volume2-FAT32", "IsBrowsable": False},
|
|
],
|
|
{"C:/Users/justin": [{"Path": "C%3a%2fUsers%2fjustin%2fapp.conf", "Type": "File"}]},
|
|
)
|
|
assert _resolve(client, r"C:\Users\justin\app.conf") == "C%3a%2fUsers%2fjustin%2fapp.conf"
|
|
# the bug this guards: probing C:/C:/Users/justin
|
|
assert not any(p.count("C:") > 1 for p in client.seen)
|
|
|
|
|
|
def test_resolve_flr_path_windows_handles_spaces_encoded_as_plus():
|
|
client = _BrowseFake(
|
|
[{"Path": "C%3a", "IsBrowsable": True}],
|
|
{
|
|
"C:/Program Files/app": [
|
|
{"Path": "C%3a%2fProgram+Files%2fapp%2fmy+config.ini", "Type": "File"}
|
|
]
|
|
},
|
|
)
|
|
got = _resolve(client, r"C:\Program Files\app\my config.ini")
|
|
assert got == "C%3a%2fProgram+Files%2fapp%2fmy+config.ini"
|
|
|
|
|
|
def test_wait_partitions_stable_ignores_the_early_mislabelled_list():
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp import recover
|
|
|
|
recover.PARTITION_POLL_INTERVAL_S = 0
|
|
# first poll is the pre-enumeration list, then it settles on the real one
|
|
polls = [
|
|
[{"Path": "Volume4-Unknown", "IsBrowsable": False}],
|
|
[{"Path": "C%3a", "IsBrowsable": True}],
|
|
[{"Path": "C%3a", "IsBrowsable": True}],
|
|
]
|
|
|
|
class Settling:
|
|
async def browse_flr(self, session_id, path="", recursive=False):
|
|
return {"PathItems": polls.pop(0) if polls else [{"Path": "C%3a", "IsBrowsable": True}]}
|
|
|
|
got = asyncio.run(recover.wait_partitions_stable(Settling(), "sess"))
|
|
assert got == ["C%3a"]
|
|
|
|
|
|
def test_session_rows_and_id_shapes():
|
|
from zerto_rewind_mcp.recover import session_id_of, session_rows
|
|
|
|
assert session_rows([{"a": 1}]) == [{"a": 1}]
|
|
assert session_rows({"a": 1}) == [{"a": 1}]
|
|
assert session_rows(None) == []
|
|
assert session_id_of({"FlrSessionIdentifier": "s1"}) == "s1"
|
|
assert session_id_of({"sessionId": "s2"}) == "s2"
|
|
assert session_id_of({}) == ""
|
|
|
|
|
|
def test_is_live_session():
|
|
from zerto_rewind_mcp.recover import is_live_session
|
|
|
|
assert is_live_session({"FlrSessionStatus": "MountCompletedSuccessfully"})
|
|
assert is_live_session({"FlrSessionStatus": "MountInProgress"})
|
|
# unmounted/ended/failed sessions linger as history and hold nothing
|
|
assert not is_live_session({"FlrSessionStatus": "UnmountCompletedSuccessfully"})
|
|
assert not is_live_session({"FlrSessionStatus": "MountFailed"})
|
|
assert not is_live_session({})
|
|
|
|
|
|
class _FakeClient:
|
|
"""Stands in for ZertoClient in teardown tests."""
|
|
|
|
def __init__(self, sessions=None, fail_end=False):
|
|
self._sessions = sessions or []
|
|
self.fail_end = fail_end
|
|
self.ended: list[str] = []
|
|
|
|
async def list_flrs(self):
|
|
return self._sessions
|
|
|
|
async def end_flr(self, session_id):
|
|
from zerto_rewind_mcp.client import ZertoError
|
|
|
|
if self.fail_end:
|
|
raise ZertoError("unmount refused")
|
|
self.ended.append(session_id)
|
|
|
|
|
|
def test_teardown_reports_failure_instead_of_swallowing():
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp.server import _teardown_flr
|
|
|
|
client = _FakeClient(fail_end=True)
|
|
out = asyncio.run(_teardown_flr(client, "sess-1", set()))
|
|
assert out["ok"] is False
|
|
assert out["failed"][0]["session_id"] == "sess-1"
|
|
assert "still be mounted" in out["message"]
|
|
|
|
|
|
def test_teardown_reaps_orphan_when_session_id_never_parsed():
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp.server import _teardown_flr
|
|
|
|
# start_flr succeeded on the ZVM but the response did not parse, so the
|
|
# caller never learned the id. The new live session must still be reaped.
|
|
live = [{"FlrSessionIdentifier": "new-1", "FlrSessionStatus": "MountCompletedSuccessfully"}]
|
|
client = _FakeClient(sessions=live)
|
|
out = asyncio.run(_teardown_flr(client, None, set()))
|
|
assert out["ended"] == ["new-1"]
|
|
assert out["orphans_reaped"] == ["new-1"]
|
|
assert client.ended == ["new-1"]
|
|
|
|
|
|
def test_teardown_leaves_pre_existing_sessions_alone():
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp.server import _teardown_flr
|
|
|
|
live = [
|
|
{"FlrSessionIdentifier": "someone-else", "FlrSessionStatus": "MountCompletedSuccessfully"}
|
|
]
|
|
client = _FakeClient(sessions=live)
|
|
out = asyncio.run(_teardown_flr(client, None, {"someone-else"}))
|
|
assert out["ended"] == []
|
|
assert client.ended == []
|
|
|
|
|
|
class _SiteFake:
|
|
def __init__(self, protected, recovery):
|
|
self._vpg = {
|
|
"VpgName": "demo-vpg",
|
|
"ProtectedSite": {"identifier": protected},
|
|
"RecoverySite": {"identifier": recovery},
|
|
}
|
|
|
|
async def get_vpg(self, vpg_identifier):
|
|
return self._vpg
|
|
|
|
async def get_localsite(self):
|
|
return {"SiteIdentifier": "site-local", "SiteName": "VMware Site"}
|
|
|
|
async def get_peersites(self):
|
|
return [{"SiteIdentifier": "site-aws", "PeerSiteName": "aws-zca"}]
|
|
|
|
|
|
def test_flr_gate_allows_local_replication():
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp.server import _flr_site_gate
|
|
|
|
out = asyncio.run(_flr_site_gate(_SiteFake("site-local", "site-local"), "v1"))
|
|
assert out["ok"] is True
|
|
|
|
|
|
def test_flr_gate_refuses_remote_recovery_site_and_names_it():
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp.server import _flr_site_gate
|
|
|
|
out = asyncio.run(_flr_site_gate(_SiteFake("site-local", "site-aws"), "v1"))
|
|
assert out["ok"] is False
|
|
assert out["not_local_replication"] is True
|
|
# must tell the operator where the operation actually lives
|
|
assert out["recovery_site"] == "aws-zca"
|
|
assert "aws-zca" in out["message"]
|
|
|
|
|
|
def test_encode_recovered_text_and_binary():
|
|
from zerto_rewind_mcp.server import _encode_recovered
|
|
|
|
text = _encode_recovered(b"listen: 0.0.0.0:8443\n")
|
|
assert text["encoding"] == "text"
|
|
assert text["content"] == "listen: 0.0.0.0:8443\n"
|
|
|
|
binary = _encode_recovered(b"\x89PNG\r\n\x1a\n\xff\xfe")
|
|
assert binary["encoding"] == "base64"
|
|
import base64 as b64
|
|
|
|
assert b64.b64decode(binary["content"]) == b"\x89PNG\r\n\x1a\n\xff\xfe"
|
|
|
|
|
|
def test_recover_file_takes_no_caller_destination():
|
|
"""The caller must not choose where bytes land, nor where the ZVM mounts.
|
|
|
|
dest_dir used to be a tool parameter whose value was also passed to the
|
|
appliance as initialDownloadPath.
|
|
"""
|
|
import inspect
|
|
|
|
from zerto_rewind_mcp.server import zerto_recover_file
|
|
|
|
params = set(inspect.signature(zerto_recover_file).parameters)
|
|
assert "dest_dir" not in params
|
|
assert params == {
|
|
"vpg_identifier",
|
|
"vm_identifier",
|
|
"checkpoint_identifier",
|
|
"guest_path",
|
|
"confirmed",
|
|
}
|
|
|
|
|
|
def test_recover_byte_cap_is_configurable_and_has_a_default():
|
|
from zerto_rewind_mcp.server import MAX_RECOVER_BYTES
|
|
|
|
assert MAX_RECOVER_BYTES == 1_048_576
|