80 lines
2.3 KiB
Python
80 lines
2.3 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"
|
|
)
|
|
|
|
|
|
def test_resolve_flr_path_picks_browsable_partition():
|
|
import asyncio
|
|
|
|
from zerto_rewind_mcp.recover import resolve_flr_path
|
|
|
|
root = {
|
|
"PathItems": [
|
|
{"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
|
|
assert "Volume1-Unknown/home/justin" not in client.seen
|