wehub-resource-sync
565 lines
19 KiB
Python
565 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import stat
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from opensquilla.migration import opensquilla_home as home_migration
|
|
from opensquilla.migration import source_snapshot_windows as windows_snapshot
|
|
|
|
|
|
def _information(
|
|
inode: int,
|
|
*,
|
|
directory: bool,
|
|
size: int = 0,
|
|
attributes: int = 0,
|
|
) -> windows_snapshot._HandleInformation:
|
|
if directory:
|
|
attributes |= windows_snapshot._FILE_ATTRIBUTE_DIRECTORY
|
|
file_type = stat.S_IFDIR
|
|
else:
|
|
file_type = stat.S_IFREG
|
|
return windows_snapshot._HandleInformation(
|
|
identity=windows_snapshot.WindowsPathIdentity(
|
|
device=71,
|
|
inode=inode,
|
|
file_type=file_type,
|
|
),
|
|
mode=windows_snapshot._mode_from_attributes(attributes),
|
|
size=size,
|
|
mtime_ns=1_700_000_000_000_000_000 + inode,
|
|
attributes=attributes,
|
|
)
|
|
|
|
|
|
class _FakeWindowsApi:
|
|
def __init__(self) -> None:
|
|
self.root = Path("/fake/source")
|
|
self.data = {
|
|
self.root / "state" / "sessions.db": b"database",
|
|
self.root / "state" / "sessions.db-wal": b"write-ahead-log",
|
|
self.root / "state" / "sessions.db-shm": b"shared-memory",
|
|
}
|
|
self.nodes = {
|
|
Path("/"): _information(1, directory=True),
|
|
Path("/fake"): _information(2, directory=True),
|
|
self.root: _information(3, directory=True),
|
|
self.root / "state": _information(4, directory=True),
|
|
**{
|
|
path: _information(index, directory=False, size=len(value))
|
|
for index, (path, value) in enumerate(self.data.items(), start=5)
|
|
},
|
|
}
|
|
self.children = {
|
|
Path("/"): ("fake",),
|
|
Path("/fake"): ("source",),
|
|
self.root: ("state",),
|
|
self.root / "state": tuple(path.name for path in self.data),
|
|
}
|
|
self._next_handle = 100
|
|
self._handles: dict[int, Path] = {}
|
|
self._offsets: dict[int, int] = {}
|
|
self.open_order: list[Path] = []
|
|
self.open_modes: list[tuple[Path, bool]] = []
|
|
self.read_paths: list[Path] = []
|
|
|
|
def normalize_root(self, path: Path) -> Path:
|
|
assert path == self.root or self.root in path.parents
|
|
return path
|
|
|
|
def open_path(self, path: Path, *, allow_writers: bool) -> int:
|
|
path = Path(path)
|
|
assert path in self.nodes
|
|
if path != Path("/"):
|
|
parent = path.parent
|
|
assert parent in self._handles.values(), f"parent was not pinned: {parent}"
|
|
handle = self._next_handle
|
|
self._next_handle += 1
|
|
self._handles[handle] = path
|
|
self._offsets[handle] = 0
|
|
self.open_order.append(path)
|
|
self.open_modes.append((path, allow_writers))
|
|
return handle
|
|
|
|
def close(self, handle: int) -> None:
|
|
self._handles.pop(handle)
|
|
self._offsets.pop(handle)
|
|
|
|
def information(
|
|
self,
|
|
handle: int,
|
|
*,
|
|
path: Path,
|
|
) -> windows_snapshot._HandleInformation:
|
|
assert self._handles[handle] == path
|
|
return self.nodes[path]
|
|
|
|
def enumerate_names(self, handle: int, *, path: Path) -> tuple[str, ...]:
|
|
assert self._handles[handle] == path
|
|
return tuple(sorted(self.children.get(path, ())))
|
|
|
|
def read(self, handle: int, size: int) -> bytes:
|
|
path = self._handles[handle]
|
|
self.read_paths.append(path)
|
|
data = self.data[path]
|
|
offset = self._offsets[handle]
|
|
chunk = data[offset : offset + size]
|
|
self._offsets[handle] += len(chunk)
|
|
return chunk
|
|
|
|
@property
|
|
def open_handles(self) -> set[int]:
|
|
return set(self._handles)
|
|
|
|
|
|
def _add_gateway_authority(api: _FakeWindowsApi) -> None:
|
|
values = {
|
|
api.root / "state" / "gateway.pid": b"12345\n",
|
|
api.root / "state" / "gateway.pid.lock": b"locked\n",
|
|
}
|
|
start = max(value.identity.inode for value in api.nodes.values()) + 1
|
|
for index, (path, value) in enumerate(values.items(), start=start):
|
|
api.data[path] = value
|
|
api.nodes[path] = _information(index, directory=False, size=len(value))
|
|
api.children[api.root / "state"] = tuple(
|
|
sorted({*api.children[api.root / "state"], *(path.name for path in values)})
|
|
)
|
|
|
|
|
|
def test_public_scan_and_copy_use_pinned_api_for_sqlite_bundle(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
scan_api = _FakeWindowsApi()
|
|
monkeypatch.setattr(windows_snapshot, "_new_api", lambda: scan_api)
|
|
|
|
snapshot = windows_snapshot.scan_windows_source_tree(
|
|
scan_api.root,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
)
|
|
|
|
assert not scan_api.open_handles
|
|
assert (Path("/"), True) in scan_api.open_modes
|
|
assert (Path("/fake"), True) in scan_api.open_modes
|
|
assert (scan_api.root, False) in scan_api.open_modes
|
|
assert (scan_api.root / "state", True) in scan_api.open_modes
|
|
assert (scan_api.root / "state", False) in scan_api.open_modes
|
|
files = [entry for entry in snapshot.entries if entry.entry_type == "file"]
|
|
assert {entry.relative.as_posix() for entry in files} == {
|
|
"state/sessions.db",
|
|
"state/sessions.db-shm",
|
|
"state/sessions.db-wal",
|
|
}
|
|
for entry in files:
|
|
assert entry.digest == hashlib.sha256(scan_api.data[entry.source]).hexdigest()
|
|
assert (entry.source, True) in scan_api.open_modes
|
|
|
|
for entry in files:
|
|
copy_api = _FakeWindowsApi()
|
|
monkeypatch.setattr(windows_snapshot, "_new_api", lambda api=copy_api: api)
|
|
destination = tmp_path / entry.relative
|
|
digest = windows_snapshot.copy_windows_snapshot_file(snapshot, entry, destination)
|
|
assert digest == entry.digest
|
|
assert destination.read_bytes() == copy_api.data[entry.source]
|
|
assert (entry.source, True) in copy_api.open_modes
|
|
assert not copy_api.open_handles
|
|
|
|
|
|
def test_parent_reparse_swap_is_rejected_before_sqlite_leaf_read(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
scan_api = _FakeWindowsApi()
|
|
monkeypatch.setattr(windows_snapshot, "_new_api", lambda: scan_api)
|
|
snapshot = windows_snapshot.scan_windows_source_tree(
|
|
scan_api.root,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
)
|
|
database = next(
|
|
entry for entry in snapshot.entries if entry.relative == Path("state/sessions.db")
|
|
)
|
|
|
|
swapped_api = _FakeWindowsApi()
|
|
old_state = swapped_api.nodes[swapped_api.root / "state"]
|
|
swapped_api.nodes[swapped_api.root / "state"] = windows_snapshot._HandleInformation(
|
|
identity=old_state.identity,
|
|
mode=old_state.mode,
|
|
size=old_state.size,
|
|
mtime_ns=old_state.mtime_ns,
|
|
attributes=old_state.attributes | windows_snapshot._FILE_ATTRIBUTE_REPARSE_POINT,
|
|
)
|
|
monkeypatch.setattr(windows_snapshot, "_new_api", lambda: swapped_api)
|
|
destination = tmp_path / "sessions.db"
|
|
|
|
with pytest.raises(
|
|
windows_snapshot.WindowsSourceSnapshotError,
|
|
match="reparse point",
|
|
):
|
|
windows_snapshot.copy_windows_snapshot_file(snapshot, database, destination)
|
|
|
|
assert not destination.exists()
|
|
assert swapped_api.root / "state" / "sessions.db" not in swapped_api.read_paths
|
|
assert not swapped_api.open_handles
|
|
|
|
|
|
def test_file_writer_sharing_still_rejects_metadata_change_during_read(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
api = _FakeWindowsApi()
|
|
database = api.root / "state" / "sessions.db"
|
|
original_read = api.read
|
|
mutated = False
|
|
|
|
def mutate_after_first_read(handle: int, size: int) -> bytes:
|
|
nonlocal mutated
|
|
chunk = original_read(handle, size)
|
|
if api._handles[handle] == database and not mutated:
|
|
current = api.nodes[database]
|
|
api.nodes[database] = windows_snapshot._HandleInformation(
|
|
identity=current.identity,
|
|
mode=current.mode,
|
|
size=current.size,
|
|
mtime_ns=current.mtime_ns + 100,
|
|
attributes=current.attributes,
|
|
)
|
|
mutated = True
|
|
return chunk
|
|
|
|
api.read = mutate_after_first_read # type: ignore[method-assign]
|
|
monkeypatch.setattr(windows_snapshot, "_new_api", lambda: api)
|
|
|
|
with pytest.raises(
|
|
windows_snapshot.WindowsSourceSnapshotError,
|
|
match="source changed while pinned",
|
|
):
|
|
windows_snapshot.scan_windows_source_tree(
|
|
api.root,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
)
|
|
|
|
assert (database, True) in api.open_modes
|
|
assert not api.open_handles
|
|
|
|
|
|
def test_plain_parent_directory_swap_is_rejected_before_leaf_read(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
scan_api = _FakeWindowsApi()
|
|
monkeypatch.setattr(windows_snapshot, "_new_api", lambda: scan_api)
|
|
snapshot = windows_snapshot.scan_windows_source_tree(
|
|
scan_api.root,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
)
|
|
database = next(
|
|
entry for entry in snapshot.entries if entry.relative == Path("state/sessions.db")
|
|
)
|
|
|
|
swapped_api = _FakeWindowsApi()
|
|
swapped_api.nodes[swapped_api.root / "state"] = _information(9001, directory=True)
|
|
monkeypatch.setattr(windows_snapshot, "_new_api", lambda: swapped_api)
|
|
destination = tmp_path / "sessions.db"
|
|
|
|
with pytest.raises(
|
|
windows_snapshot.WindowsSourceSnapshotError,
|
|
match="source directory changed before copy",
|
|
):
|
|
windows_snapshot.copy_windows_snapshot_file(snapshot, database, destination)
|
|
|
|
assert not destination.exists()
|
|
assert swapped_api.root / "state" / "sessions.db" not in swapped_api.read_paths
|
|
assert not swapped_api.open_handles
|
|
|
|
|
|
def test_bounded_gateway_authority_uses_same_pinned_parent_chain() -> None:
|
|
api = _FakeWindowsApi()
|
|
_add_gateway_authority(api)
|
|
|
|
captured = windows_snapshot._capture_bounded_with_api(
|
|
api,
|
|
api.root / "state",
|
|
names=("gateway.pid", "gateway.pid.lock"),
|
|
max_bytes=1024,
|
|
)
|
|
|
|
assert captured is not None
|
|
assert captured.file("gateway.pid") is not None
|
|
assert captured.file("gateway.pid").data == b"12345\n"
|
|
assert captured.file("gateway.pid.lock") is not None
|
|
assert captured.file("gateway.pid.lock").data == b"locked\n"
|
|
assert (api.root / "state" / "gateway.pid", False) in api.open_modes
|
|
assert (api.root / "state" / "gateway.pid.lock", False) in api.open_modes
|
|
assert not api.open_handles
|
|
|
|
|
|
def test_pinned_ancestor_ignores_metadata_from_unrelated_sibling_activity() -> None:
|
|
api = _FakeWindowsApi()
|
|
ancestor = Path("/fake")
|
|
|
|
with windows_snapshot._open_directory_chain(api, api.root):
|
|
original = api.nodes[ancestor]
|
|
api.nodes[ancestor] = windows_snapshot._HandleInformation(
|
|
identity=original.identity,
|
|
mode=original.mode,
|
|
size=original.size + 1,
|
|
mtime_ns=original.mtime_ns + 1,
|
|
attributes=original.attributes,
|
|
)
|
|
|
|
assert not api.open_handles
|
|
|
|
|
|
def test_pinned_ancestor_still_rejects_identity_replacement() -> None:
|
|
api = _FakeWindowsApi()
|
|
ancestor = Path("/fake")
|
|
|
|
with pytest.raises(
|
|
windows_snapshot.WindowsSourceSnapshotError,
|
|
match="source path component changed",
|
|
):
|
|
with windows_snapshot._open_directory_chain(api, api.root):
|
|
api.nodes[ancestor] = _information(9001, directory=True)
|
|
|
|
assert not api.open_handles
|
|
|
|
|
|
def test_bounded_gateway_authority_rejects_state_parent_reparse() -> None:
|
|
api = _FakeWindowsApi()
|
|
_add_gateway_authority(api)
|
|
old_state = api.nodes[api.root / "state"]
|
|
api.nodes[api.root / "state"] = windows_snapshot._HandleInformation(
|
|
identity=old_state.identity,
|
|
mode=old_state.mode,
|
|
size=old_state.size,
|
|
mtime_ns=old_state.mtime_ns,
|
|
attributes=old_state.attributes | windows_snapshot._FILE_ATTRIBUTE_REPARSE_POINT,
|
|
)
|
|
|
|
with pytest.raises(windows_snapshot.WindowsSourceSnapshotError, match="reparse point"):
|
|
windows_snapshot._capture_bounded_with_api(
|
|
api,
|
|
api.root / "state",
|
|
names=("gateway.pid", "gateway.pid.lock"),
|
|
max_bytes=1024,
|
|
)
|
|
|
|
assert not api.open_handles
|
|
|
|
|
|
def test_migration_dispatch_retains_native_windows_snapshot_authority(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
api = _FakeWindowsApi()
|
|
native = windows_snapshot._scan_with_api(
|
|
api,
|
|
api.root,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
excluded=frozenset(),
|
|
)
|
|
monkeypatch.setattr(home_migration, "os", SimpleNamespace(name="nt"))
|
|
monkeypatch.setattr(home_migration, "scan_windows_source_tree", lambda *args, **kwargs: native)
|
|
|
|
snapshot = home_migration._scan_source_tree(
|
|
api.root,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
)
|
|
|
|
assert snapshot.windows_snapshot is native
|
|
entry = next(item for item in snapshot.entries if item.relative == Path("state/sessions.db"))
|
|
observed: dict[str, object] = {}
|
|
|
|
def copy_native(
|
|
received_snapshot: windows_snapshot.WindowsSourceSnapshot,
|
|
received_entry: windows_snapshot.WindowsManifestEntry,
|
|
destination: Path,
|
|
) -> str:
|
|
observed.update(
|
|
snapshot=received_snapshot,
|
|
entry=received_entry,
|
|
destination=destination,
|
|
)
|
|
return received_entry.digest or ""
|
|
|
|
monkeypatch.setattr(home_migration, "copy_windows_snapshot_file", copy_native)
|
|
destination = tmp_path / "sessions.db"
|
|
digest = home_migration._copy_snapshot_file(snapshot, entry, destination)
|
|
|
|
assert digest == entry.digest
|
|
assert observed == {
|
|
"snapshot": native,
|
|
"entry": next(item for item in native.entries if item.relative == entry.relative),
|
|
"destination": destination,
|
|
}
|
|
|
|
|
|
def test_gateway_authority_dispatch_retains_native_windows_authority(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
api = _FakeWindowsApi()
|
|
_add_gateway_authority(api)
|
|
native = windows_snapshot._capture_bounded_with_api(
|
|
api,
|
|
api.root / "state",
|
|
names=("gateway.pid", "gateway.pid.lock"),
|
|
max_bytes=1024,
|
|
)
|
|
assert native is not None
|
|
monkeypatch.setattr(home_migration, "os", SimpleNamespace(name="nt"))
|
|
monkeypatch.setattr(
|
|
home_migration,
|
|
"capture_windows_bounded_files",
|
|
lambda *args, **kwargs: native,
|
|
)
|
|
|
|
captured = home_migration._capture_legacy_gateway_authority(api.root / "state")
|
|
|
|
assert captured.windows_authority is native
|
|
assert captured.pid_value == 12345
|
|
assert captured.pid is not None
|
|
assert captured.lock is not None
|
|
|
|
|
|
def test_createfile_contract_excludes_delete_share_and_opens_reparse_object() -> None:
|
|
api = object.__new__(windows_snapshot._Win32SourceApi)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def create_file(*args: Any) -> int:
|
|
captured["args"] = args
|
|
return 321
|
|
|
|
api._create_file = create_file
|
|
|
|
assert api.open_path(Path("C:/synthetic/profile"), allow_writers=False) == 321
|
|
args = captured["args"]
|
|
assert args[2] == windows_snapshot._FILE_SHARE_READ
|
|
assert not args[2] & windows_snapshot._FILE_SHARE_DELETE
|
|
assert args[5] & windows_snapshot._FILE_FLAG_OPEN_REPARSE_POINT
|
|
assert args[5] & windows_snapshot._FILE_FLAG_BACKUP_SEMANTICS
|
|
|
|
assert api.open_path(Path("C:/synthetic"), allow_writers=True) == 321
|
|
directory_args = captured["args"]
|
|
assert directory_args[2] & windows_snapshot._FILE_SHARE_WRITE
|
|
assert not directory_args[2] & windows_snapshot._FILE_SHARE_DELETE
|
|
|
|
|
|
def test_directory_enumeration_header_uses_fixed_width_windows_layout() -> None:
|
|
assert windows_snapshot._FileIdBothDirInfoHeader.file_name.offset == 104
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="non-Windows fail-closed contract")
|
|
def test_native_api_has_no_non_windows_path_fallback() -> None:
|
|
with pytest.raises(windows_snapshot.WindowsSourceSnapshotUnavailableError):
|
|
windows_snapshot._new_api()
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "nt", reason="requires real Win32 source handles")
|
|
def test_real_windows_snapshot_copies_database_sidecars(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
state = source / "state"
|
|
state.mkdir(parents=True)
|
|
values = {
|
|
"sessions.db": b"db",
|
|
"sessions.db-wal": b"wal",
|
|
"sessions.db-shm": b"shm",
|
|
}
|
|
for name, value in values.items():
|
|
(state / name).write_bytes(value)
|
|
|
|
snapshot = windows_snapshot.scan_windows_source_tree(
|
|
source,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
)
|
|
for entry in snapshot.entries:
|
|
if entry.entry_type != "file":
|
|
continue
|
|
destination = tmp_path / "copied" / entry.relative
|
|
windows_snapshot.copy_windows_snapshot_file(snapshot, entry, destination)
|
|
assert destination.read_bytes() == values[entry.relative.name]
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "nt", reason="requires real Win32 source handles")
|
|
def test_real_windows_pinned_ancestor_allows_unrelated_sibling_write(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
source.mkdir()
|
|
api = windows_snapshot._new_api()
|
|
|
|
with windows_snapshot._open_directory_chain(api, source):
|
|
sibling = tmp_path / "unrelated-sibling.txt"
|
|
sibling.write_bytes(b"unrelated")
|
|
|
|
assert sibling.read_bytes() == b"unrelated"
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "nt", reason="requires a real Windows junction or symlink")
|
|
def test_real_windows_parent_reparse_to_original_directory_is_rejected(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = tmp_path / "source"
|
|
state = source / "state"
|
|
state.mkdir(parents=True)
|
|
database = state / "sessions.db"
|
|
database.write_bytes(b"stable-database")
|
|
snapshot = windows_snapshot.scan_windows_source_tree(
|
|
source,
|
|
destination_prefix=Path(),
|
|
role="primary",
|
|
)
|
|
entry = next(item for item in snapshot.entries if item.relative == Path("state/sessions.db"))
|
|
|
|
moved = source / "moved-state"
|
|
state.rename(moved)
|
|
try:
|
|
state.symlink_to(moved, target_is_directory=True)
|
|
except OSError as exc:
|
|
pytest.skip(f"creating a directory reparse point is unavailable: {exc}")
|
|
|
|
destination = tmp_path / "copied.db"
|
|
with pytest.raises(
|
|
windows_snapshot.WindowsSourceSnapshotError,
|
|
match="reparse point|source root changed",
|
|
):
|
|
windows_snapshot.copy_windows_snapshot_file(snapshot, entry, destination)
|
|
assert not destination.exists()
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "nt", reason="requires a real Windows junction or symlink")
|
|
def test_real_windows_gateway_authority_parent_reparse_is_rejected(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
state = source / "state"
|
|
state.mkdir(parents=True)
|
|
(state / "gateway.pid").write_text("12345\n", encoding="utf-8")
|
|
captured = windows_snapshot.capture_windows_bounded_files(
|
|
state,
|
|
names=("gateway.pid", "gateway.pid.lock"),
|
|
max_bytes=1024,
|
|
)
|
|
assert captured is not None
|
|
|
|
moved = source / "moved-state"
|
|
state.rename(moved)
|
|
try:
|
|
state.symlink_to(moved, target_is_directory=True)
|
|
except OSError as exc:
|
|
pytest.skip(f"creating a directory reparse point is unavailable: {exc}")
|
|
|
|
with pytest.raises(windows_snapshot.WindowsSourceSnapshotError, match="reparse point"):
|
|
windows_snapshot.capture_windows_bounded_files(
|
|
state,
|
|
names=("gateway.pid", "gateway.pid.lock"),
|
|
max_bytes=1024,
|
|
)
|