zerto-ai-rewind/tests/test_recover.py

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"]