feat(flr): windows paths, recovery-site gate, stable partition reads (#4)

tests/test_recover.py

@@ -46,39 +46,96 @@ def test_decode_flr_path():
     )
 
 
-def test_resolve_flr_path_picks_browsable_partition():
+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.recover import resolve_flr_path
+    from zerto_rewind_mcp import recover
 
-    root = {
-        "PathItems": [
+    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},
-        ]
-    }
-    listing = {
-        "PathItems": [
-            {"Path": "Volume2-Ext4%2fhome%2fjustin%2f.bashrc", "Type": "File"},
-            {"Path": "Volume2-Ext4%2fhome%2fjustin%2fapp-config.yaml", "Type": "File"},
-        ]
-    }
-
-    class FakeClient:
-        def __init__(self):
-            self.seen = []
-
-        async def browse_flr(self, session_id, path="", recursive=False):
-            self.seen.append(path)
-            return root if path == "" else listing
-
-    client = FakeClient()
-    got = asyncio.run(resolve_flr_path(client, "sess", "/home/justin/app-config.yaml"))
-    assert got == "Volume2-Ext4/home/justin/app-config.yaml"
-    # must not try the unrestorable partition
+        ],
+        {
+            "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
 
@@ -159,3 +216,43 @@ def test_teardown_leaves_pre_existing_sessions_alone():
     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"]