mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-03 16:41:26 +01:00
2160 lines
84 KiB
Python
Executable File
2160 lines
84 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Account-free filesystem API contract plus opt-in live FUSE acceptance.
|
|
|
|
Every case runs first against a temporary local directory and then, when asked,
|
|
against the mount. The local run is not just a smoke test of the runner: its
|
|
observations become the reference that the mount is diffed against, so a case
|
|
fails when the mount *differs* from an ordinary filesystem, not only when it
|
|
raises. Differences that are legitimate (device ids, block counts, capabilities
|
|
a network filesystem cannot have) are recorded as notes instead of comparisons,
|
|
which makes the set of accepted divergences explicit and reviewable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import concurrent.futures
|
|
import contextlib
|
|
import ctypes
|
|
import errno
|
|
import faulthandler
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import mmap
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import sqlite3
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
import xml.etree.ElementTree as ElementTree
|
|
|
|
# A wedged FUSE operation is unkillable from Python, so every case runs under a
|
|
# soft alarm and a hard backstop. The soft alarm raises inside the main thread
|
|
# and lets cleanup and reporting run; the backstop dumps every thread's stack
|
|
# and exits, which is the only way to learn where a mount actually hung.
|
|
DEFAULT_TIMEOUT = int(os.environ.get("PDFS_ACCEPTANCE_TIMEOUT", "180"))
|
|
HARD_TIMEOUT_GRACE = 20
|
|
|
|
REFERENCE = "reference"
|
|
LIVE = "live"
|
|
|
|
# Failing an unsupported operation is fine; failing it *dirtily* is not. These
|
|
# are the errno values that mean "this filesystem does not offer that", as
|
|
# opposed to EIO or a hang, which mean something is broken.
|
|
CLEAN_UNSUPPORTED = {
|
|
errno.EPERM,
|
|
errno.ENOSYS,
|
|
errno.EOPNOTSUPP,
|
|
errno.EACCES,
|
|
errno.EINVAL,
|
|
errno.ENOTTY,
|
|
}
|
|
|
|
STALE_ROOT = re.compile(r"^pdfs-acceptance-[0-9a-f]{32}$")
|
|
ANSI = re.compile(r"\x1b\[[0-9;]*m")
|
|
ERROR_LEVEL = re.compile(r"\bERROR\b")
|
|
|
|
FALLOC_FL_KEEP_SIZE = 0x01
|
|
FALLOC_FL_PUNCH_HOLE = 0x02
|
|
RENAME_NOREPLACE = 1 << 0
|
|
RENAME_EXCHANGE = 1 << 1
|
|
AT_FDCWD = -100
|
|
|
|
_libc = ctypes.CDLL(None, use_errno=True)
|
|
|
|
|
|
def _libc_call(name: str, argtypes: list, *args) -> None:
|
|
entry = getattr(_libc, name, None)
|
|
if entry is None:
|
|
raise OSError(errno.ENOSYS, f"{name} is unavailable in libc")
|
|
entry.argtypes = argtypes
|
|
entry.restype = ctypes.c_int
|
|
ctypes.set_errno(0)
|
|
if entry(*args) != 0:
|
|
code = ctypes.get_errno()
|
|
raise OSError(code, os.strerror(code))
|
|
|
|
|
|
def fallocate(fd: int, mode: int, offset: int, length: int) -> None:
|
|
_libc_call(
|
|
"fallocate",
|
|
[ctypes.c_int, ctypes.c_int, ctypes.c_longlong, ctypes.c_longlong],
|
|
ctypes.c_int(fd),
|
|
ctypes.c_int(mode),
|
|
ctypes.c_longlong(offset),
|
|
ctypes.c_longlong(length),
|
|
)
|
|
|
|
|
|
def renameat2(old: Path, new: Path, flags: int) -> None:
|
|
_libc_call(
|
|
"renameat2",
|
|
[ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint],
|
|
ctypes.c_int(AT_FDCWD),
|
|
os.fsencode(old),
|
|
ctypes.c_int(AT_FDCWD),
|
|
os.fsencode(new),
|
|
ctypes.c_uint(flags),
|
|
)
|
|
|
|
|
|
def check(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise AssertionError(message)
|
|
|
|
|
|
def expect_errno(expected: set[int], operation, description: str) -> None:
|
|
try:
|
|
operation()
|
|
except OSError as error:
|
|
check(error.errno in expected, f"{description}: errno {error.errno}, expected {expected}")
|
|
else:
|
|
raise AssertionError(f"{description}: unexpectedly succeeded")
|
|
|
|
|
|
def check_bytes(actual: bytes, expected: bytes, message: str) -> None:
|
|
"""Assert byte equality with a report that is actually diagnosable.
|
|
|
|
"wrong bytes" is useless in a log: a truncated upload, a zero-filled gap and
|
|
a wholly different revision all produce it. Say which one it was.
|
|
"""
|
|
if actual == expected:
|
|
return
|
|
detail = f"{message}: got {len(actual)} bytes, expected {len(expected)}"
|
|
if len(actual) == len(expected):
|
|
offset = next(i for i, (a, b) in enumerate(zip(actual, expected)) if a != b)
|
|
run_end = offset
|
|
while run_end < len(actual) and actual[run_end] != expected[run_end]:
|
|
run_end += 1
|
|
zeros = actual[offset:run_end] == bytes(run_end - offset)
|
|
detail += (
|
|
f"; first difference at {offset} ({run_end - offset} bytes"
|
|
f"{', all zero' if zeros else ''})"
|
|
f"; got {actual[offset:offset + 16]!r} expected {expected[offset:offset + 16]!r}"
|
|
)
|
|
elif expected.startswith(actual):
|
|
detail += "; got a truncated prefix of the expected content"
|
|
elif actual.startswith(expected):
|
|
detail += "; got the expected content plus trailing data"
|
|
raise AssertionError(detail)
|
|
|
|
|
|
def read(path: Path) -> bytes:
|
|
with path.open("rb", buffering=0) as file:
|
|
return file.read()
|
|
|
|
|
|
def write_durable(path: Path, data: bytes) -> None:
|
|
fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
|
|
try:
|
|
view = memoryview(data)
|
|
while view:
|
|
written = os.write(fd, view)
|
|
check(written > 0, "write made no progress")
|
|
view = view[written:]
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
class Observations:
|
|
"""Per-case facts, split into what must match the reference and what may not.
|
|
|
|
`record` is a claim about filesystem semantics: the mount has to agree with
|
|
an ordinary filesystem or the case fails. `note` is context — inode numbers,
|
|
block counts, which optional syscalls exist — that is reported in the diff
|
|
but never fails a run. Choosing between them at the call site is what keeps
|
|
the accepted-divergence list honest; there is no separate allowlist to drift.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.compared: dict[str, object] = {}
|
|
self.noted: dict[str, object] = {}
|
|
self._scope = ""
|
|
|
|
def scope(self, name: str) -> None:
|
|
self._scope = name
|
|
|
|
def _key(self, key: str) -> str:
|
|
return f"{self._scope}.{key}" if self._scope else key
|
|
|
|
def record(self, key: str, value) -> None:
|
|
self.compared[self._key(key)] = value
|
|
|
|
def note(self, key: str, value) -> None:
|
|
self.noted[self._key(key)] = value
|
|
|
|
def stat(self, key: str, path: Path) -> os.stat_result:
|
|
"""Record the portable half of a stat and note the rest."""
|
|
info = os.lstat(path)
|
|
self.record(f"{key}.type", stat.S_IFMT(info.st_mode))
|
|
self.record(f"{key}.size", info.st_size)
|
|
self.record(f"{key}.is_dir", stat.S_ISDIR(info.st_mode))
|
|
self.note(f"{key}.mode", stat.S_IMODE(info.st_mode))
|
|
self.note(f"{key}.nlink", info.st_nlink)
|
|
self.note(f"{key}.blocks", info.st_blocks)
|
|
return info
|
|
|
|
def probe(self, key: str, operation, allow: set[int] = CLEAN_UNSUPPORTED) -> Outcome:
|
|
"""Run an optional operation; demand that failure at least be clean.
|
|
|
|
The reference filesystem supports more than a network filesystem can, so
|
|
the *outcome* is a note. What is asserted is the shape of a refusal: a
|
|
recognised errno, never EIO and never a hang.
|
|
"""
|
|
try:
|
|
value = operation()
|
|
except OSError as error:
|
|
check(
|
|
error.errno in allow,
|
|
f"{key}: refused with errno {error.errno} ({errno.errorcode.get(error.errno)}); "
|
|
f"expected success or one of {sorted(allow)}",
|
|
)
|
|
self.note(key, f"errno:{errno.errorcode.get(error.errno, error.errno)}")
|
|
return Outcome(False, None, error.errno)
|
|
self.note(key, "ok")
|
|
return Outcome(True, value, None)
|
|
|
|
|
|
class Outcome:
|
|
"""The result of an optional operation: did it work, and if not, why."""
|
|
|
|
def __init__(self, ok: bool, value, code: int | None) -> None:
|
|
self.ok = ok
|
|
self.value = value
|
|
self.errno = code
|
|
|
|
def __bool__(self) -> bool:
|
|
return self.ok
|
|
|
|
|
|
class Context:
|
|
"""Everything a case may touch: its sandbox, its recorder, and the daemon."""
|
|
|
|
def __init__(self, root: Path, obs: Observations, kind: str, daemon=None) -> None:
|
|
self.root = root
|
|
self.obs = obs
|
|
self.kind = kind
|
|
self.daemon = daemon
|
|
|
|
@property
|
|
def is_live(self) -> bool:
|
|
return self.kind == LIVE
|
|
|
|
def record(self, key: str, value) -> None:
|
|
self.obs.record(key, value)
|
|
|
|
def note(self, key: str, value) -> None:
|
|
self.obs.note(key, value)
|
|
|
|
def probe(self, key: str, operation, allow: set[int] = CLEAN_UNSUPPORTED):
|
|
return self.obs.probe(key, operation, allow)
|
|
|
|
def stat(self, key: str, path: Path) -> os.stat_result:
|
|
return self.obs.stat(key, path)
|
|
|
|
def require_daemon(self) -> Daemon:
|
|
if self.daemon is None:
|
|
raise Skip("no reachable pdfs daemon; set PDFS_ACCEPTANCE_PDFS")
|
|
return self.daemon
|
|
|
|
|
|
class Skip(Exception):
|
|
"""A case cannot run here, and that is not a failure."""
|
|
|
|
|
|
class TestTimeout(Exception):
|
|
pass
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def time_limit(seconds: int, label: str):
|
|
if seconds <= 0:
|
|
yield
|
|
return
|
|
|
|
def on_alarm(_signum, _frame):
|
|
raise TestTimeout(f"{label} exceeded {seconds}s")
|
|
|
|
previous = signal.signal(signal.SIGALRM, on_alarm)
|
|
faulthandler.dump_traceback_later(seconds + HARD_TIMEOUT_GRACE, exit=True)
|
|
signal.setitimer(signal.ITIMER_REAL, seconds)
|
|
try:
|
|
yield
|
|
finally:
|
|
signal.setitimer(signal.ITIMER_REAL, 0)
|
|
faulthandler.cancel_dump_traceback_later()
|
|
signal.signal(signal.SIGALRM, previous)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Filesystem API contract
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_flags(ctx: Context) -> None:
|
|
root = ctx.root
|
|
path = root / "flags"
|
|
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600)
|
|
os.write(fd, b"abcdef")
|
|
os.close(fd)
|
|
expect_errno({errno.EEXIST}, lambda: os.open(path, os.O_CREAT | os.O_EXCL), "O_EXCL")
|
|
fd = os.open(path, os.O_WRONLY | os.O_APPEND)
|
|
os.lseek(fd, 0, os.SEEK_SET)
|
|
os.write(fd, b"++")
|
|
os.close(fd)
|
|
check(read(path) == b"abcdef++", "O_APPEND ignored")
|
|
ctx.record("append", read(path))
|
|
fd = os.open(path, os.O_WRONLY | os.O_TRUNC)
|
|
os.close(fd)
|
|
check(path.stat().st_size == 0, "O_TRUNC did not truncate")
|
|
ctx.stat("truncated", path)
|
|
|
|
|
|
def test_positioned_and_vectored_io(ctx: Context) -> None:
|
|
root = ctx.root
|
|
path = root / "positioned.bin"
|
|
base = bytes(range(256)) * 32768
|
|
write_durable(path, base)
|
|
fd = os.open(path, os.O_RDWR)
|
|
try:
|
|
check(os.pread(fd, 19, 4093) == base[4093:4112], "pread returned wrong range")
|
|
check(os.pwrite(fd, b"boundary-write", 4093) == 14, "short pwrite")
|
|
if hasattr(os, "pwritev"):
|
|
check(os.pwritev(fd, [b"vector-", b"write"], 65531) == 12, "short pwritev")
|
|
if hasattr(os, "preadv"):
|
|
chunks = [bytearray(7), bytearray(5)]
|
|
check(os.preadv(fd, chunks, 65531) == 12, "short preadv")
|
|
check(b"".join(chunks) == b"vector-write", "preadv data mismatch")
|
|
os.fdatasync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
expected = bytearray(base)
|
|
expected[4093:4107] = b"boundary-write"
|
|
expected[65531:65543] = b"vector-write"
|
|
got = read(path)
|
|
check(got == expected, "positioned write damaged surrounding bytes")
|
|
digest = hashlib.sha256(got).hexdigest()
|
|
ctx.record("digest", digest)
|
|
ctx.stat("file", path)
|
|
print(f" sha256 {digest}")
|
|
|
|
|
|
def test_resize_and_sparse_io(ctx: Context) -> None:
|
|
root = ctx.root
|
|
path = root / "resize"
|
|
write_durable(path, b"0123456789")
|
|
os.truncate(path, 4)
|
|
check(read(path) == b"0123", "shrink did not preserve prefix")
|
|
os.truncate(path, 8193)
|
|
data = read(path)
|
|
check(data[:4] == b"0123" and data[4:] == bytes(8189), "grown range is not zero-filled")
|
|
ctx.record("grown.digest", hashlib.sha256(data).hexdigest())
|
|
|
|
sparse = root / "sparse"
|
|
fd = os.open(sparse, os.O_CREAT | os.O_RDWR, 0o600)
|
|
try:
|
|
os.pwrite(fd, b"head", 0)
|
|
os.pwrite(fd, b"tail", 8 * 1024 * 1024 + 17)
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
check(sparse.stat().st_size == 8 * 1024 * 1024 + 21, "sparse size mismatch")
|
|
fd = os.open(sparse, os.O_RDONLY)
|
|
try:
|
|
check(os.pread(fd, 8, 4) == bytes(8), "sparse hole is not zero-filled")
|
|
check(os.pread(fd, 4, 8 * 1024 * 1024 + 17) == b"tail", "sparse tail missing")
|
|
check(os.pread(fd, 1, sparse.stat().st_size + 4096) == b"", "read beyond EOF not empty")
|
|
finally:
|
|
os.close(fd)
|
|
ctx.stat("sparse", sparse)
|
|
|
|
|
|
def test_mmap_and_copy_paths(ctx: Context) -> None:
|
|
root = ctx.root
|
|
source = root / "mapped"
|
|
data = bytearray((b"mmap-and-sendfile\0" * 65536)[:1024 * 1024])
|
|
write_durable(source, data)
|
|
fd = os.open(source, os.O_RDWR)
|
|
try:
|
|
with mmap.mmap(fd, len(data), access=mmap.ACCESS_WRITE) as mapping:
|
|
mapping[4091:4107] = b"mapped-boundary!"
|
|
mapping.flush()
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
data[4091:4107] = b"mapped-boundary!"
|
|
check(read(source) == data, "shared mmap write/read mismatch")
|
|
|
|
target = root / "sendfile-copy"
|
|
src = os.open(source, os.O_RDONLY)
|
|
dst = os.open(target, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
|
|
try:
|
|
offset = 0
|
|
while offset < len(data):
|
|
count = os.sendfile(dst, src, offset, len(data) - offset)
|
|
check(count > 0, "sendfile made no progress")
|
|
offset += count
|
|
os.fsync(dst)
|
|
finally:
|
|
os.close(src)
|
|
os.close(dst)
|
|
check(read(target) == data, "sendfile copy mismatch")
|
|
ctx.record("digest", hashlib.sha256(read(target)).hexdigest())
|
|
|
|
# coreutils reaches for copy_file_range before read/write, so a filesystem
|
|
# that neither implements it nor refuses it cleanly breaks plain `cp`.
|
|
ranged = root / "copy-file-range"
|
|
src = os.open(source, os.O_RDONLY)
|
|
dst = os.open(ranged, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
|
|
try:
|
|
copied = ctx.probe(
|
|
"copy_file_range",
|
|
lambda: os.copy_file_range(src, dst, len(data)),
|
|
allow=CLEAN_UNSUPPORTED | {errno.EXDEV},
|
|
)
|
|
finally:
|
|
os.close(src)
|
|
os.close(dst)
|
|
if copied.ok and copied.value:
|
|
check(
|
|
read(ranged)[: copied.value] == data[: copied.value],
|
|
"copy_file_range produced wrong bytes",
|
|
)
|
|
|
|
|
|
def test_namespace_and_errors(ctx: Context) -> None:
|
|
root = ctx.root
|
|
a, b = root / "dir-a", root / "dir-b"
|
|
a.mkdir()
|
|
b.mkdir()
|
|
(a / "nested").mkdir()
|
|
write_durable(a / "nested" / "child", b"child")
|
|
write_durable(root / "source", b"source")
|
|
write_durable(root / "victim", b"victim")
|
|
os.replace(root / "source", root / "victim")
|
|
check(read(root / "victim") == b"source" and not (root / "source").exists(), "replace failed")
|
|
os.rename(root / "victim", b / "moved")
|
|
check(read(b / "moved") == b"source", "cross-directory move failed")
|
|
check("moved" in os.listdir(b), "cross-directory move missing from readdir")
|
|
os.rename(b / "moved", b / "moved")
|
|
check(read(b / "moved") == b"source", "same-name rename changed data")
|
|
ctx.record("moved.bytes", read(b / "moved"))
|
|
|
|
expect_errno({errno.ENOTEMPTY, errno.EEXIST}, lambda: os.rmdir(a), "rmdir(non-empty)")
|
|
expect_errno({errno.ENOTDIR}, lambda: os.rmdir(b / "moved"), "rmdir(file)")
|
|
expect_errno({errno.EISDIR, errno.EPERM}, lambda: os.unlink(a), "unlink(directory)")
|
|
expect_errno({errno.EINVAL}, lambda: os.rename(a, a / "nested" / "cycle"), "directory cycle")
|
|
expect_errno({errno.ENOENT}, lambda: os.unlink(root / "absent"), "unlink(absent)")
|
|
expect_errno({errno.ENOENT}, lambda: os.rename(root / "absent", root / "new"), "rename(absent)")
|
|
|
|
empty_src, empty_dst = root / "empty-src", root / "empty-dst"
|
|
empty_src.mkdir()
|
|
empty_dst.mkdir()
|
|
os.replace(empty_src, empty_dst)
|
|
check(empty_dst.is_dir() and not empty_src.exists(), "empty directory replacement failed")
|
|
nonempty = root / "nonempty-dst"
|
|
nonempty.mkdir()
|
|
write_durable(nonempty / "keep", b"keep")
|
|
replacement = root / "replacement-dir"
|
|
replacement.mkdir()
|
|
expect_errno(
|
|
{errno.ENOTEMPTY, errno.EEXIST},
|
|
lambda: os.replace(replacement, nonempty),
|
|
"replace non-empty dir",
|
|
)
|
|
check(read(nonempty / "keep") == b"keep", "failed replacement damaged destination")
|
|
|
|
# renameat2 flags: git and dpkg use NOREPLACE, and a filesystem that ignores
|
|
# the flag instead of refusing it turns a guarded rename into a clobber.
|
|
guard_src, guard_dst = root / "guard-src", root / "guard-dst"
|
|
write_durable(guard_src, b"guard-src")
|
|
write_durable(guard_dst, b"guard-dst")
|
|
guarded = ctx.probe(
|
|
"renameat2.noreplace",
|
|
lambda: renameat2(guard_src, guard_dst, RENAME_NOREPLACE),
|
|
allow=CLEAN_UNSUPPORTED | {errno.EEXIST},
|
|
)
|
|
check(not guarded.ok, "RENAME_NOREPLACE overwrote an existing destination")
|
|
check(read(guard_dst) == b"guard-dst", "RENAME_NOREPLACE damaged the destination")
|
|
ctx.probe("renameat2.exchange", lambda: renameat2(guard_src, guard_dst, RENAME_EXCHANGE))
|
|
|
|
|
|
def test_open_lifetime(ctx: Context) -> None:
|
|
root = ctx.root
|
|
path = root / "open-unlink"
|
|
data = b"held-open\0" * 4096
|
|
write_durable(path, data)
|
|
fd = os.open(path, os.O_RDWR)
|
|
os.unlink(path)
|
|
check(not path.exists(), "unlinked name remains visible")
|
|
check(os.pread(fd, len(data), 0) == data, "open file lost data after unlink")
|
|
os.pwrite(fd, b"still-open", 0)
|
|
check(os.pread(fd, 10, 0) == b"still-open", "unlinked open file is not writable")
|
|
os.close(fd)
|
|
|
|
old, new = root / "open-rename-old", root / "open-rename-new"
|
|
write_durable(old, b"before")
|
|
fd = os.open(old, os.O_RDWR)
|
|
os.rename(old, new)
|
|
os.pwrite(fd, b"after!", 0)
|
|
os.fsync(fd)
|
|
os.close(fd)
|
|
check(not old.exists() and read(new) == b"after!", "open handle did not follow rename")
|
|
ctx.record("renamed.bytes", read(new))
|
|
|
|
|
|
def test_names_and_enumeration(ctx: Context) -> None:
|
|
root = ctx.root
|
|
names = ["space name.txt", "unicodé-文件", ".hidden", "trailing.dot.", "case", "CASE"]
|
|
for index, name in enumerate(names):
|
|
write_durable(root / name, f"name-{index}".encode())
|
|
listed = set(os.listdir(root))
|
|
check(set(names) <= listed, "directory enumeration omitted valid names")
|
|
for name in listed:
|
|
os.lstat(root / name)
|
|
ctx.record("case_sensitive", read(root / "case") != read(root / "CASE"))
|
|
ctx.record("names", sorted(name for name in names))
|
|
expect_errno({errno.ENAMETOOLONG}, lambda: os.open(root / ("x" * 256), os.O_CREAT), "overlong name")
|
|
|
|
|
|
def test_readdir_stability(ctx: Context) -> None:
|
|
"""Enumeration must not duplicate or lose entries when the directory moves.
|
|
|
|
readdir is served off the dispatch loop, so a pass that re-reads the child
|
|
list between two kernel-visible offsets can double-report or skip an entry.
|
|
A lister that sees one name twice deletes it twice.
|
|
"""
|
|
root = ctx.root
|
|
directory = root / "readdir-churn"
|
|
directory.mkdir()
|
|
stable = [f"stable-{index:03d}" for index in range(64)]
|
|
for name in stable:
|
|
write_durable(directory / name, b"x")
|
|
|
|
seen: list[str] = []
|
|
created: list[str] = []
|
|
with os.scandir(directory) as entries:
|
|
for position, entry in enumerate(entries):
|
|
seen.append(entry.name)
|
|
# Mutate mid-enumeration: churn is the point.
|
|
if position == 8:
|
|
for index in range(4):
|
|
name = f"late-{index}"
|
|
write_durable(directory / name, b"y")
|
|
created.append(name)
|
|
if position == 24:
|
|
os.unlink(directory / stable[-1])
|
|
|
|
check(len(seen) == len(set(seen)), f"readdir reported duplicates: {sorted(seen)}")
|
|
survivors = set(os.listdir(directory))
|
|
check(set(created) <= survivors, "entries created during enumeration were lost")
|
|
check(stable[-1] not in survivors, "entry unlinked during enumeration reappeared")
|
|
# An entry that existed unchanged for the whole pass must be reported once.
|
|
missing = set(stable[:-1]) - set(seen)
|
|
check(not missing, f"stable entries vanished from readdir: {sorted(missing)[:10]}")
|
|
ctx.record("no_duplicates", True)
|
|
ctx.record("survivors", len(survivors))
|
|
|
|
# rewinddir: a second full pass must agree with the settled directory.
|
|
check(set(os.listdir(directory)) == survivors, "second enumeration disagreed with the first")
|
|
|
|
|
|
def test_metadata_updates(ctx: Context) -> None:
|
|
"""chmod/utimes must succeed even where they cannot be stored.
|
|
|
|
Drive has no POSIX mode or owner, and `setattr` deliberately accepts and
|
|
ignores them. That is only correct if it reports success: `cp -p`, `tar -x`
|
|
and `rsync -a` all treat a refused chmod as a fatal error.
|
|
"""
|
|
root = ctx.root
|
|
path = root / "metadata"
|
|
write_durable(path, b"metadata")
|
|
|
|
os.chmod(path, 0o640)
|
|
ctx.note("mode.after_chmod", stat.S_IMODE(os.lstat(path).st_mode))
|
|
os.utime(path, (1_600_000_000, 1_600_000_123))
|
|
ctx.note("mtime.after_utime", int(os.lstat(path).st_mtime))
|
|
ctx.record("chmod_and_utime_succeeded", True)
|
|
|
|
directory = root / "metadata-dir"
|
|
directory.mkdir()
|
|
os.chmod(directory, 0o750)
|
|
os.utime(directory, None)
|
|
|
|
# A path-based truncate with no open handle is a separate write path (a
|
|
# shell's `> file`), and it is the one that has to work offline.
|
|
write_durable(path, b"0123456789")
|
|
os.truncate(path, 0)
|
|
check(read(path) == b"", "closed-file truncate to zero did not empty the file")
|
|
ctx.record("closed_truncate.size", os.lstat(path).st_size)
|
|
|
|
with path.open("wb", buffering=0) as handle:
|
|
handle.write(b"redirect")
|
|
check(read(path) == b"redirect", "reopen-and-write after truncate lost bytes")
|
|
ctx.record("redirect.bytes", read(path))
|
|
|
|
|
|
def test_extended_attributes(ctx: Context) -> None:
|
|
"""The thumbnail xattr interface, including its two-call size protocol."""
|
|
root = ctx.root
|
|
path = root / "attributes.txt"
|
|
write_durable(path, b"attributes")
|
|
|
|
expect_errno(
|
|
{errno.ENODATA, errno.ENOATTR} if hasattr(errno, "ENOATTR") else {errno.ENODATA},
|
|
lambda: os.getxattr(path, "user.proton.definitely-absent"),
|
|
"getxattr(unknown name)",
|
|
)
|
|
ctx.probe("setxattr", lambda: os.setxattr(path, "user.pdfs.probe", b"v"))
|
|
|
|
listed = os.listxattr(path)
|
|
ctx.note("listxattr", sorted(listed))
|
|
# Thumbnails exist only for images and video; advertising them on a text
|
|
# file is what made `ls -l` issue a network round trip per file (B5).
|
|
check(
|
|
"user.proton.thumbnail" not in listed and "user.proton.preview" not in listed,
|
|
f"a text file advertised thumbnail attributes: {listed}",
|
|
)
|
|
ctx.record("no_thumbnail_on_text", True)
|
|
|
|
if not ctx.is_live:
|
|
return
|
|
# Opportunistic: if the mount holds a real image, exercise the size protocol
|
|
# against it. Nothing here creates one, because only the server can.
|
|
for candidate in _find_thumbnailable(ctx.root.parent):
|
|
names = os.listxattr(candidate)
|
|
if "user.proton.thumbnail" not in names:
|
|
continue
|
|
size = _xattr_size(candidate, "user.proton.thumbnail")
|
|
if size == 0:
|
|
continue
|
|
expect_errno(
|
|
{errno.ERANGE},
|
|
lambda: _xattr_into(candidate, "user.proton.thumbnail", size - 1),
|
|
"getxattr(undersized buffer)",
|
|
)
|
|
payload = os.getxattr(candidate, "user.proton.thumbnail")
|
|
check(len(payload) == size, "thumbnail length disagreed with its size probe")
|
|
ctx.note("thumbnail.bytes", size)
|
|
return
|
|
ctx.note("thumbnail.bytes", "no thumbnailable file found")
|
|
|
|
|
|
def _find_thumbnailable(directory: Path, limit: int = 200):
|
|
suffixes = {".jpg", ".jpeg", ".png", ".heic", ".mp4", ".mov", ".webp", ".gif"}
|
|
seen = 0
|
|
try:
|
|
entries = sorted(directory.iterdir())
|
|
except OSError:
|
|
return
|
|
for entry in entries:
|
|
if seen >= limit:
|
|
return
|
|
seen += 1
|
|
if entry.is_file() and entry.suffix.lower() in suffixes:
|
|
yield entry
|
|
|
|
|
|
def _xattr_size(path: Path, name: str) -> int:
|
|
fd = os.open(path, os.O_RDONLY)
|
|
try:
|
|
return len(os.getxattr(fd, name))
|
|
except OSError:
|
|
return 0
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def _xattr_into(path: Path, name: str, size: int) -> bytes:
|
|
"""Force the undersized-buffer branch that Python's helper hides."""
|
|
buffer = ctypes.create_string_buffer(max(size, 1))
|
|
_libc.getxattr.argtypes = [
|
|
ctypes.c_char_p,
|
|
ctypes.c_char_p,
|
|
ctypes.c_void_p,
|
|
ctypes.c_size_t,
|
|
]
|
|
_libc.getxattr.restype = ctypes.c_long
|
|
ctypes.set_errno(0)
|
|
written = _libc.getxattr(os.fsencode(path), name.encode(), buffer, ctypes.c_size_t(size))
|
|
if written < 0:
|
|
code = ctypes.get_errno()
|
|
raise OSError(code, os.strerror(code))
|
|
return buffer.raw[:written]
|
|
|
|
|
|
def test_allocation_and_holes(ctx: Context) -> None:
|
|
"""fallocate, including the punched hole that must survive the upload.
|
|
|
|
A punched range reads as zero. The commit path gap-fills unauthored ranges
|
|
from the remote baseline, so if the hole is not claimed as authored the
|
|
original bytes come back and the hole silently disappears after close.
|
|
"""
|
|
root = ctx.root
|
|
path = root / "allocated"
|
|
write_durable(path, b"A" * (256 * 1024))
|
|
|
|
fd = os.open(path, os.O_RDWR)
|
|
try:
|
|
ctx.probe("posix_fallocate", lambda: os.posix_fallocate(fd, 0, 512 * 1024))
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
ctx.note("allocated.size", os.lstat(path).st_size)
|
|
|
|
holed = root / "punched"
|
|
write_durable(holed, b"B" * (192 * 1024))
|
|
fd = os.open(holed, os.O_RDWR)
|
|
punched = False
|
|
try:
|
|
try:
|
|
fallocate(fd, FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE, 64 * 1024, 64 * 1024)
|
|
punched = True
|
|
except OSError as error:
|
|
check(
|
|
error.errno in CLEAN_UNSUPPORTED,
|
|
f"punch hole refused with errno {error.errno}",
|
|
)
|
|
ctx.note("punch_hole", f"errno:{errno.errorcode.get(error.errno)}")
|
|
if punched:
|
|
ctx.note("punch_hole", "ok")
|
|
check(
|
|
os.pread(fd, 64 * 1024, 64 * 1024) == bytes(64 * 1024),
|
|
"punched range did not read as zeros",
|
|
)
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
if punched:
|
|
check(os.lstat(holed).st_size == 192 * 1024, "punch hole with KEEP_SIZE changed the size")
|
|
expected = b"B" * (64 * 1024) + bytes(64 * 1024) + b"B" * (64 * 1024)
|
|
# Reopening is the assertion that matters: it is where a gap-fill from
|
|
# the remote baseline would undo the hole.
|
|
check(read(holed) == expected, "punched hole did not survive close and reopen")
|
|
# A note, not a comparison: whether the filesystem can punch at all is a
|
|
# capability. That it produced the right bytes when it did is asserted
|
|
# above, where the answer does not depend on the reference having agreed
|
|
# to punch too.
|
|
ctx.note("punched.digest", hashlib.sha256(read(holed)).hexdigest())
|
|
|
|
# tar and `cp --sparse` navigate with SEEK_HOLE/SEEK_DATA.
|
|
fd = os.open(holed, os.O_RDONLY)
|
|
try:
|
|
ctx.probe("seek_data", lambda: os.lseek(fd, 0, os.SEEK_DATA), allow=CLEAN_UNSUPPORTED | {errno.ENXIO})
|
|
ctx.probe("seek_hole", lambda: os.lseek(fd, 0, os.SEEK_HOLE), allow=CLEAN_UNSUPPORTED | {errno.ENXIO})
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def test_unsupported_operations(ctx: Context) -> None:
|
|
"""Operations the filesystem does not implement must fail cleanly, not hang.
|
|
|
|
Every one of these is probed by ordinary tools — git, rsync, `cp -a`, `df`,
|
|
installers. An unimplemented FUSE handler that returns EIO, or blocks, turns
|
|
a graceful fallback into a hard failure in software that never asked for the
|
|
feature in the first place.
|
|
"""
|
|
root = ctx.root
|
|
target = root / "link-target"
|
|
write_durable(target, b"link-target")
|
|
|
|
ctx.probe("symlink", lambda: os.symlink("link-target", root / "symlink"))
|
|
if (root / "symlink").is_symlink():
|
|
check(os.readlink(root / "symlink") == "link-target", "readlink returned the wrong target")
|
|
ctx.probe("hardlink", lambda: os.link(target, root / "hardlink"))
|
|
ctx.probe("mkfifo", lambda: os.mkfifo(root / "fifo"))
|
|
ctx.probe("mknod", lambda: os.mknod(root / "node", 0o600 | stat.S_IFCHR, os.makedev(1, 3)))
|
|
|
|
# df and any installer that checks free space call statfs. It must answer.
|
|
info = os.statvfs(root)
|
|
check(info.f_bsize > 0, "statfs reported a zero block size")
|
|
ctx.note("statfs.bsize", info.f_bsize)
|
|
ctx.note("statfs.blocks", info.f_blocks)
|
|
ctx.record("statfs_answers", True)
|
|
|
|
check(os.access(root, os.R_OK | os.X_OK), "access(2) denied read/execute on the test root")
|
|
check(os.access(target, os.R_OK), "access(2) denied read on a readable file")
|
|
|
|
fd = os.open(target, os.O_RDWR)
|
|
try:
|
|
ctx.probe("flock", lambda: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB))
|
|
with contextlib.suppress(OSError):
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
ctx.probe("posix_lock", lambda: fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB, 0, 1))
|
|
with contextlib.suppress(OSError):
|
|
fcntl.lockf(fd, fcntl.LOCK_UN, 0, 1)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
direct = ctx.probe("o_direct", lambda: os.open(target, os.O_RDONLY | os.O_DIRECT))
|
|
if direct.ok:
|
|
with contextlib.suppress(OSError):
|
|
# O_DIRECT demands an aligned buffer; mmap gives a page-aligned one.
|
|
with mmap.mmap(-1, 4096) as buffer:
|
|
os.preadv(direct.value, [buffer], 0)
|
|
os.close(direct.value)
|
|
|
|
# TCGETS against a filesystem must answer ENOTTY, which is what stops fuser
|
|
# from logging a warning every time a program probes whether an fd is a tty.
|
|
fd = os.open(root, os.O_RDONLY)
|
|
try:
|
|
ctx.probe("ioctl_tcgets", lambda: fcntl.ioctl(fd, 0x5401, b"\0" * 64))
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def test_concurrency(ctx: Context) -> None:
|
|
root = ctx.root
|
|
|
|
def independent(index: int) -> None:
|
|
data = bytes([index]) * (131072 + index)
|
|
path = root / f"concurrent-{index}"
|
|
write_durable(path, data)
|
|
check(read(path) == data, f"concurrent file {index} mismatch")
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
list(pool.map(independent, range(1, 17)))
|
|
|
|
shared = root / "concurrent-ranges"
|
|
extent = 64 * 1024
|
|
write_durable(shared, bytes(extent * 8))
|
|
fd = os.open(shared, os.O_RDWR)
|
|
try:
|
|
|
|
def positioned(index: int) -> None:
|
|
payload = bytes([index + 1]) * extent
|
|
check(os.pwrite(fd, payload, index * extent) == extent, f"short concurrent pwrite {index}")
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
list(pool.map(positioned, range(8)))
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
expected = b"".join(bytes([i + 1]) * extent for i in range(8))
|
|
check(read(shared) == expected, "concurrent disjoint writes overlapped or vanished")
|
|
ctx.record("shared.digest", hashlib.sha256(read(shared)).hexdigest())
|
|
|
|
|
|
def test_application_workloads(ctx: Context) -> None:
|
|
"""Real tools, because they combine syscalls in ways a matrix will not.
|
|
|
|
Each of these has broken on a network filesystem before: the editor save
|
|
dance (write temp, fsync, rename over), git's lock files and many small
|
|
objects, tar's metadata restore, sqlite's journal and locking.
|
|
"""
|
|
root = ctx.root
|
|
workloads = root / "workloads"
|
|
workloads.mkdir()
|
|
|
|
# The atomic-save cycle every editor performs, repeated so a stale inode or
|
|
# a mishandled replace shows up rather than passing once by luck.
|
|
document = workloads / "document.txt"
|
|
for revision in range(5):
|
|
temporary = workloads / f".document.txt.{revision}.tmp"
|
|
payload = f"revision {revision}\n".encode() * 128
|
|
write_durable(temporary, payload)
|
|
os.replace(temporary, document)
|
|
check(read(document) == payload, f"atomic save lost revision {revision}")
|
|
ctx.record("atomic_save.bytes", read(document))
|
|
|
|
# tar restores mode and mtime on extract; a filesystem that refuses either
|
|
# makes every extraction fail loudly.
|
|
tree = workloads / "tree"
|
|
(tree / "nested").mkdir(parents=True)
|
|
write_durable(tree / "nested" / "leaf", b"leaf" * 512)
|
|
write_durable(tree / "top", b"top")
|
|
archive = workloads / "tree.tar"
|
|
with tarfile.open(archive, "w") as bundle:
|
|
bundle.add(tree, arcname="tree")
|
|
extracted = workloads / "extracted"
|
|
extracted.mkdir()
|
|
with tarfile.open(archive) as bundle:
|
|
try:
|
|
bundle.extractall(extracted, filter="data")
|
|
except TypeError: # the extraction filter predates Python 3.12
|
|
bundle.extractall(extracted)
|
|
check(
|
|
read(extracted / "tree" / "nested" / "leaf") == b"leaf" * 512,
|
|
"tar extraction produced wrong bytes",
|
|
)
|
|
ctx.record("tar.leaf_size", os.lstat(extracted / "tree" / "nested" / "leaf").st_size)
|
|
|
|
# sqlite exercises locking, journal creation and unlink, and fsync ordering.
|
|
database = workloads / "workload.db"
|
|
connection = sqlite3.connect(database)
|
|
try:
|
|
connection.execute("CREATE TABLE rows (id INTEGER PRIMARY KEY, value TEXT)")
|
|
with connection:
|
|
connection.executemany(
|
|
"INSERT INTO rows (value) VALUES (?)", [(f"row-{index}",) for index in range(500)]
|
|
)
|
|
connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
|
count = connection.execute("SELECT COUNT(*) FROM rows").fetchone()[0]
|
|
finally:
|
|
connection.close()
|
|
check(count == 500, f"sqlite lost rows: {count}")
|
|
ctx.record("sqlite.rows", count)
|
|
connection = sqlite3.connect(database)
|
|
try:
|
|
check(
|
|
connection.execute("PRAGMA integrity_check").fetchone()[0] == "ok",
|
|
"sqlite integrity check failed after reopen",
|
|
)
|
|
finally:
|
|
connection.close()
|
|
ctx.record("sqlite.integrity", "ok")
|
|
|
|
if shutil.which("git"):
|
|
repository = workloads / "repository"
|
|
repository.mkdir()
|
|
_run_tool(["git", "init", "--quiet", "."], repository)
|
|
_run_tool(["git", "config", "user.email", "acceptance@example.invalid"], repository)
|
|
_run_tool(["git", "config", "user.name", "pdfs acceptance"], repository)
|
|
for index in range(3):
|
|
write_durable(repository / f"file-{index}", f"content {index}\n".encode() * 64)
|
|
_run_tool(["git", "add", "-A"], repository)
|
|
_run_tool(["git", "commit", "--quiet", "-m", f"commit {index}"], repository)
|
|
status = _run_tool(["git", "status", "--porcelain"], repository)
|
|
check(status.strip() == "", f"git saw a dirty tree after committing: {status!r}")
|
|
_run_tool(["git", "fsck", "--no-progress"], repository)
|
|
ctx.record("git.clean_status", True)
|
|
else:
|
|
ctx.note("git", "not installed")
|
|
|
|
if shutil.which("rsync"):
|
|
mirror = workloads / "rsync-mirror"
|
|
mirror.mkdir()
|
|
_run_tool(["rsync", "-a", f"{tree}/", f"{mirror}/"], workloads)
|
|
check(
|
|
read(mirror / "nested" / "leaf") == b"leaf" * 512,
|
|
"rsync mirrored the wrong bytes",
|
|
)
|
|
ctx.record("rsync.leaf_size", os.lstat(mirror / "nested" / "leaf").st_size)
|
|
else:
|
|
ctx.note("rsync", "not installed")
|
|
|
|
|
|
def _run_tool(command: list[str], cwd: Path) -> str:
|
|
result = subprocess.run(command, cwd=cwd, text=True, capture_output=True)
|
|
if result.returncode:
|
|
detail = (result.stderr or result.stdout).strip()
|
|
raise AssertionError(f"{' '.join(command)} failed in {cwd}: {detail}")
|
|
return result.stdout
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Regressions from docs/BUGS.md
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def test_regression_b7_renamed_directory(ctx: Context) -> None:
|
|
"""A renamed directory must stay traversable (B7).
|
|
|
|
The reported failure was a directory that appeared in `ls` but could not be
|
|
entered, because rename replaced the inode the kernel still had cached.
|
|
"""
|
|
root = ctx.root
|
|
original = root / "b7 original dir"
|
|
original.mkdir()
|
|
write_durable(original / "child file.txt", b"b7-child")
|
|
(original / "sub dir").mkdir()
|
|
write_durable(original / "sub dir" / "deep", b"b7-deep")
|
|
|
|
before = os.lstat(original).st_ino
|
|
renamed = root / "b7 renamed dïr ünï"
|
|
os.rename(original, renamed)
|
|
|
|
check(renamed.is_dir(), "renamed directory is not a directory")
|
|
check(set(os.listdir(renamed)) == {"child file.txt", "sub dir"}, "renamed directory lost children")
|
|
check(read(renamed / "child file.txt") == b"b7-child", "child unreadable after directory rename")
|
|
check(read(renamed / "sub dir" / "deep") == b"b7-deep", "grandchild unreadable after rename")
|
|
ctx.record("children", sorted(os.listdir(renamed)))
|
|
ctx.note("inode_stable", os.lstat(renamed).st_ino == before)
|
|
|
|
# Writing through the new path proves the entry is not merely readable.
|
|
write_durable(renamed / "child file.txt", b"b7-rewritten")
|
|
check(read(renamed / "child file.txt") == b"b7-rewritten", "write through renamed directory lost")
|
|
|
|
|
|
def test_regression_b69_identical_rewrite(ctx: Context) -> None:
|
|
"""Rewriting identical bytes must not fork a `(sync-conflict …)` copy (B69).
|
|
|
|
Revision identity used to be `(mtime, size)`, so a rewrite that changed the
|
|
mtime looked like a divergent revision and the sync engine kept both.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
root = ctx.root
|
|
path = root / "b69-identical.bin"
|
|
payload = b"b69-identical-payload\n" * 4096
|
|
|
|
write_durable(path, payload)
|
|
daemon.wait_for_queue()
|
|
time.sleep(1)
|
|
write_durable(path, payload)
|
|
daemon.wait_for_queue()
|
|
|
|
conflicts = _conflict_copies(root)
|
|
check(not conflicts, f"identical rewrite produced conflict copies: {conflicts}")
|
|
check_bytes(read(path), payload, "identical rewrite changed the file contents")
|
|
ctx.record("conflicts", [])
|
|
|
|
|
|
def test_regression_b70_transient_download_name(ctx: Context) -> None:
|
|
"""An in-flight browser download must not be sealed as a revision (B70).
|
|
|
|
A `.crdownload` is a partial file the browser renames on completion. Sealing
|
|
it produced an upload of the partial bytes and then a conflict fork against
|
|
the finished name.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
root = ctx.root
|
|
partial = root / "b70-download.zip.crdownload"
|
|
final = root / "b70-download.zip"
|
|
payload = b"b70-final-payload\n" * 8192
|
|
|
|
# Write the partial in chunks with a pause, the way a download lands.
|
|
fd = os.open(partial, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
|
|
try:
|
|
half = len(payload) // 2
|
|
os.write(fd, payload[:half])
|
|
os.fsync(fd)
|
|
time.sleep(2)
|
|
os.write(fd, payload[half:])
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
os.rename(partial, final)
|
|
daemon.wait_for_queue()
|
|
|
|
listing = set(os.listdir(root))
|
|
check(final.name in listing, "completed download is missing after rename")
|
|
check(partial.name not in listing, "transient download name survived the rename")
|
|
conflicts = _conflict_copies(root)
|
|
check(not conflicts, f"download rename produced conflict copies: {conflicts}")
|
|
check_bytes(read(final), payload, "completed download has the wrong bytes")
|
|
ctx.record("final.digest", hashlib.sha256(read(final)).hexdigest())
|
|
ctx.record("conflicts", [])
|
|
|
|
|
|
def test_regression_b74_rename_after_close(ctx: Context) -> None:
|
|
"""Write, close, rename: the content must reach Drive (B74).
|
|
|
|
Not a `.crdownload` and not an overwrite — the plain temp-file dance that
|
|
every editor, browser, `rsync` and `git` performs. Renaming a file whose
|
|
create is still queued lost the staged bytes and published an empty node,
|
|
so this is the narrowest reproduction of that loss.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
root = ctx.root
|
|
payload = b"b74-rename-after-close\n" * 6144
|
|
|
|
temporary = root / "b74-staging.tmp"
|
|
final = root / "b74-final.bin"
|
|
write_durable(temporary, payload)
|
|
os.rename(temporary, final)
|
|
daemon.wait_for_queue()
|
|
# The queue reports empty the moment the create lands; the revision that
|
|
# carries the bytes can still be in flight behind it.
|
|
time.sleep(5)
|
|
daemon.wait_for_queue()
|
|
|
|
check(final.is_file(), "renamed file is missing after the queue drained")
|
|
check_bytes(read(final), payload, "renamed file lost its content")
|
|
check(not _conflict_copies(root), "rename after close produced conflict copies")
|
|
ctx.record("digest", hashlib.sha256(read(final)).hexdigest())
|
|
|
|
|
|
SHARED_DIR_NAME = "Shared with me"
|
|
|
|
|
|
def _locations(daemon: Daemon) -> list[dict]:
|
|
"""Every local Proton Drive location the daemon reports."""
|
|
return json.loads(daemon.command("locations", json_output=True)).get("items", [])
|
|
|
|
|
|
def _shared_with_me(daemon: Daemon) -> list[dict]:
|
|
return json.loads(daemon.command("shared-with-me", json_output=True)).get("entries", [])
|
|
|
|
|
|
def _shared_root(daemon: Daemon, mount: Path, role: str) -> Path:
|
|
"""The local path of an accepted share held at `role`.
|
|
|
|
Skips rather than fails when the account has no such share: the role cases
|
|
need a *second* account to have shared something at that role, which most
|
|
runs will not have (mount-architecture.md §7).
|
|
"""
|
|
entries = [e for e in _shared_with_me(daemon) if e.get("role") == role]
|
|
if not entries:
|
|
raise Skip(f"no accepted {role}-role share on this account")
|
|
for entry in entries:
|
|
# `path` is mount-relative and filled in only once the synthetic
|
|
# directory has interned the node, so list it first.
|
|
with contextlib.suppress(OSError):
|
|
os.listdir(mount / SHARED_DIR_NAME)
|
|
relative = entry.get("path") or ""
|
|
if relative:
|
|
candidate = mount / relative
|
|
if candidate.exists():
|
|
return candidate
|
|
raise Skip(f"{role}-role share is not resident under {SHARED_DIR_NAME}/")
|
|
|
|
|
|
def test_regression_b34_viewer_share_is_read_only(ctx: Context) -> None:
|
|
"""A viewer-role share must be read-only *and* queue nothing (B34).
|
|
|
|
The harm B34 names is not the mode bits: it is that a local write into a
|
|
read-only share was accepted, entered `pending_op`, and then failed 403 in
|
|
the drain forever. So step 4 — the queue depth being unchanged across the
|
|
attempts — is the assertion that distinguishes the fix from cosmetics.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
mount = daemon.enclosing_mount(ctx.root)
|
|
root = _shared_root(daemon, mount, "viewer")
|
|
|
|
before = daemon.status().get("mount") or {}
|
|
mode = stat.S_IMODE(os.stat(root).st_mode)
|
|
check(mode == 0o555, f"viewer share is {mode:o}, expected 555")
|
|
check(not os.access(root, os.W_OK), "viewer share reports itself writable")
|
|
|
|
target = root / "b34-should-not-exist"
|
|
expect_errno({errno.EACCES}, lambda: os.mkdir(target), "mkdir in a viewer share")
|
|
expect_errno(
|
|
{errno.EACCES},
|
|
lambda: os.open(target, os.O_CREAT | os.O_WRONLY, 0o600),
|
|
"create in a viewer share",
|
|
)
|
|
for child in sorted(root.iterdir())[:1]:
|
|
child_mode = stat.S_IMODE(os.stat(child).st_mode)
|
|
expected = 0o555 if child.is_dir() else 0o444
|
|
check(child_mode == expected, f"{child.name} is {child_mode:o}, expected {expected:o}")
|
|
expect_errno(
|
|
{errno.EACCES},
|
|
lambda: os.open(child, os.O_WRONLY),
|
|
"open-for-write in a viewer share",
|
|
)
|
|
expect_errno(
|
|
{errno.EACCES},
|
|
lambda: os.rename(child, child.with_name("b34-renamed")),
|
|
"rename in a viewer share",
|
|
)
|
|
expect_errno(
|
|
{errno.EACCES},
|
|
lambda: (os.rmdir(child) if child.is_dir() else os.unlink(child)),
|
|
"remove from a viewer share",
|
|
)
|
|
|
|
after = daemon.status().get("mount") or {}
|
|
for key in ("pending_uploads", "pending_changes"):
|
|
check(
|
|
before.get(key, 0) == after.get(key, 0),
|
|
f"a refused write changed {key}: {before.get(key)} -> {after.get(key)}",
|
|
)
|
|
ctx.record("viewer.mode", f"{mode:o}")
|
|
ctx.record("viewer.queue_unchanged", True)
|
|
|
|
|
|
def test_regression_b34b_editor_share_still_writes(ctx: Context) -> None:
|
|
"""The fail-open half of B34: an editor share stays writable.
|
|
|
|
A permission model that denies everything would pass the viewer case and
|
|
break the feature, so this is the guard that the enforcement is scoped.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
mount = daemon.enclosing_mount(ctx.root)
|
|
root = _shared_root(daemon, mount, "editor")
|
|
|
|
is_dir = root.is_dir()
|
|
mode = stat.S_IMODE(os.stat(root).st_mode)
|
|
expected = 0o755 if is_dir else 0o644
|
|
check(mode == expected, f"editor share is {mode:o}, expected {expected:o}")
|
|
check(os.access(root, os.W_OK), "editor share reports itself unwritable")
|
|
ctx.record("editor.mode", f"{mode:o}")
|
|
|
|
# A shared *file* is someone else's document: proving the mode bits and the
|
|
# kernel's own W_OK answer is as far as this case goes. Only a shared folder
|
|
# gets a real write, and only into a file this suite created.
|
|
if not is_dir:
|
|
ctx.note("editor.write", "skipped: the editor share is a file")
|
|
return
|
|
|
|
payload = b"b34b-editor-write\n" * 64
|
|
target = root / "b34b-editor-write.bin"
|
|
write_durable(target, payload)
|
|
daemon.wait_for_queue()
|
|
try:
|
|
check_bytes(read(target), payload, "editor-share write did not read back")
|
|
finally:
|
|
with contextlib.suppress(OSError):
|
|
os.unlink(target)
|
|
daemon.wait_for_queue()
|
|
|
|
|
|
def test_shared_directory_contract(ctx: Context) -> None:
|
|
"""The synthetic `Shared with me/` directory behaves like a real directory.
|
|
|
|
It is a real `Node` on a reserved volume rather than a FUSE special case, so
|
|
it must enumerate, stat and refuse mutation like any other read-only folder —
|
|
and must not be creatable or removable by the user.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
mount = daemon.enclosing_mount(ctx.root)
|
|
shared = mount / SHARED_DIR_NAME
|
|
if not shared.is_dir():
|
|
raise Skip(f"{SHARED_DIR_NAME}/ is not present in this mount")
|
|
|
|
mode = stat.S_IMODE(os.stat(shared).st_mode)
|
|
check(mode == 0o555, f"{SHARED_DIR_NAME}/ is {mode:o}, expected 555")
|
|
# Enumerating must not raise: it is the operation the whole feature exists for.
|
|
names = sorted(os.listdir(shared))
|
|
expect_errno(
|
|
{errno.EACCES},
|
|
lambda: os.mkdir(shared / "b34-synthetic-child"),
|
|
f"mkdir inside {SHARED_DIR_NAME}/",
|
|
)
|
|
expect_errno(
|
|
{errno.EACCES, errno.ENOTEMPTY, errno.EBUSY},
|
|
lambda: os.rmdir(shared),
|
|
f"rmdir {SHARED_DIR_NAME}/",
|
|
)
|
|
expect_errno(
|
|
{errno.EACCES, errno.EEXIST},
|
|
lambda: os.rename(shared, mount / "b34-renamed-shared"),
|
|
f"rename {SHARED_DIR_NAME}/",
|
|
)
|
|
ctx.record("shared.mode", f"{mode:o}")
|
|
ctx.note("shared.entries", len(names))
|
|
|
|
|
|
def test_regression_b79_ondemand_location_accepts_writes(ctx: Context) -> None:
|
|
"""An on-demand device folder must accept writes (B79).
|
|
|
|
Every mount holds its own inode space, and the queue guard intersects the
|
|
access each one reports. The primary mount hydrates the device folder's
|
|
nodes parentless (their device root is never a persisted node), so a
|
|
fail-closed classification there denied every write in the *secondary*
|
|
mount while its own state said Owner. The symptom is EACCES on a plain
|
|
`touch` in `~/Documents` while `~/ProtonDrive` accepts one.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
locations = [
|
|
location
|
|
for location in _locations(daemon)
|
|
if location.get("kind", {}).get("kind") == "device"
|
|
and location.get("mode") == "ondemand"
|
|
and location.get("mounted")
|
|
and location.get("access") == "rw"
|
|
]
|
|
if not locations:
|
|
raise Skip("no mounted on-demand device folder on this machine")
|
|
|
|
root = Path(locations[0]["local_path"])
|
|
payload = b"b79-ondemand-write\n" * 128
|
|
directory = root / "pdfs-b79-acceptance"
|
|
target = directory / "write.bin"
|
|
try:
|
|
os.mkdir(directory)
|
|
write_durable(target, payload)
|
|
daemon.wait_for_queue()
|
|
check_bytes(read(target), payload, "on-demand write did not read back")
|
|
finally:
|
|
with contextlib.suppress(OSError):
|
|
os.unlink(target)
|
|
with contextlib.suppress(OSError):
|
|
os.rmdir(directory)
|
|
with contextlib.suppress(Exception):
|
|
daemon.wait_for_queue()
|
|
ctx.record("b79.location", str(root))
|
|
|
|
|
|
def _conflict_copies(root: Path) -> list[str]:
|
|
return sorted(name for name in os.listdir(root) if "(sync-conflict" in name)
|
|
|
|
|
|
def test_durability_across_restart(ctx: Context) -> None:
|
|
"""Written bytes must survive a daemon restart (staging/ and recovery/).
|
|
|
|
Those directories can hold the only copy of a file whose upload has not
|
|
finished. This is the one case that proves the queue is persistent rather
|
|
than merely in-memory.
|
|
"""
|
|
daemon = ctx.require_daemon()
|
|
if not daemon.may_restart:
|
|
raise Skip("restart drill not requested; pass --durability")
|
|
root = ctx.root
|
|
path = root / "durability.bin"
|
|
payload = hashlib.sha256(b"durability").digest() * 8192
|
|
write_durable(path, payload)
|
|
daemon.wait_for_queue()
|
|
|
|
mountpoint = daemon.enclosing_mount(root)
|
|
daemon.restart()
|
|
daemon.wait_for_mount(mountpoint)
|
|
daemon.wait_for_queue()
|
|
|
|
check(path.is_file(), "file vanished across a daemon restart")
|
|
check(read(path) == payload, "file contents changed across a daemon restart")
|
|
check(not _conflict_copies(root), "restart produced conflict copies")
|
|
ctx.record("digest", hashlib.sha256(read(path)).hexdigest())
|
|
|
|
|
|
class Case:
|
|
def __init__(self, name: str, run, kinds: tuple[str, ...] = (REFERENCE, LIVE)) -> None:
|
|
self.name = name
|
|
self.run = run
|
|
self.kinds = kinds
|
|
|
|
|
|
TESTS = [
|
|
Case("creation and open flags", test_create_flags),
|
|
Case("positioned and vectored I/O", test_positioned_and_vectored_io),
|
|
Case("truncate, growth, sparse ranges, and EOF", test_resize_and_sparse_io),
|
|
Case("mmap and zero-copy copy paths", test_mmap_and_copy_paths),
|
|
Case("namespace operations and POSIX errors", test_namespace_and_errors),
|
|
Case("open-handle lifetime across unlink and rename", test_open_lifetime),
|
|
Case("names, lookup, stat, and enumeration", test_names_and_enumeration),
|
|
Case("readdir stability under concurrent mutation", test_readdir_stability),
|
|
Case("metadata updates and path-based truncate", test_metadata_updates),
|
|
Case("extended attributes and the thumbnail interface", test_extended_attributes),
|
|
Case("allocation, punched holes, and hole-seeking", test_allocation_and_holes),
|
|
Case("unsupported operations refuse cleanly", test_unsupported_operations),
|
|
Case("independent and shared-file concurrency", test_concurrency),
|
|
Case("application workloads (editor, tar, sqlite, git, rsync)", test_application_workloads),
|
|
Case("regression B7: renamed directory stays traversable", test_regression_b7_renamed_directory),
|
|
Case("regression B69: identical rewrite makes no conflict", test_regression_b69_identical_rewrite, (LIVE,)),
|
|
Case("regression B70: transient download name is not sealed", test_regression_b70_transient_download_name, (LIVE,)),
|
|
Case("regression B74: rename after close keeps its content", test_regression_b74_rename_after_close, (LIVE,)),
|
|
Case("regression B34: a viewer share is read-only and queues nothing", test_regression_b34_viewer_share_is_read_only, (LIVE,)),
|
|
Case("regression B34b: an editor share still writes", test_regression_b34b_editor_share_still_writes, (LIVE,)),
|
|
Case("synthetic Shared-with-me directory contract", test_shared_directory_contract, (LIVE,)),
|
|
Case("regression B79: an on-demand device folder accepts writes", test_regression_b79_ondemand_location_accepts_writes, (LIVE,)),
|
|
Case("durability across a daemon restart", test_durability_across_restart, (LIVE,)),
|
|
]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Runner
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
class Result:
|
|
def __init__(self, name: str, target: str, status: str, seconds: float, message: str = "") -> None:
|
|
self.name = name
|
|
self.target = target
|
|
self.status = status
|
|
self.seconds = seconds
|
|
self.message = message
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"name": self.name,
|
|
"target": self.target,
|
|
"status": self.status,
|
|
"seconds": round(self.seconds, 3),
|
|
"message": self.message,
|
|
}
|
|
|
|
|
|
class Run:
|
|
def __init__(self, label: str, kind: str, root: Path) -> None:
|
|
self.label = label
|
|
self.kind = kind
|
|
self.root = root
|
|
self.results: list[Result] = []
|
|
self.compared: dict[str, object] = {}
|
|
self.noted: dict[str, object] = {}
|
|
self.ran: set[str] = set()
|
|
self.digest = ""
|
|
|
|
@property
|
|
def failed(self) -> bool:
|
|
return any(result.status in {"fail", "timeout"} for result in self.results)
|
|
|
|
|
|
REPORT: list[Run] = []
|
|
|
|
|
|
def select_cases(kind: str) -> list[Case]:
|
|
selected = os.environ.get("PDFS_ACCEPTANCE_ONLY")
|
|
cases = [case for case in TESTS if kind in case.kinds]
|
|
if not selected:
|
|
return cases
|
|
needle = selected.lower()
|
|
# A filter that names a live-only case (every regression case is one) matches
|
|
# nothing in the reference run, which is not an error — the reference target
|
|
# simply has nothing to do. Only a filter that matches *no case at all* is a
|
|
# typo worth failing on.
|
|
check(
|
|
any(needle in case.name.lower() for case in TESTS),
|
|
f"PDFS_ACCEPTANCE_ONLY={selected!r} matched no tests",
|
|
)
|
|
return [case for case in cases if needle in case.name.lower()]
|
|
|
|
|
|
def reap_stale_roots(parent: Path, max_age: int) -> None:
|
|
"""Remove abandoned roots from crashed runs so preconditions stay meaningful.
|
|
|
|
Only exact `pdfs-acceptance-<32 hex>` names are eligible, and only once they
|
|
are older than the age cutoff, so a concurrent run's tree is never touched.
|
|
"""
|
|
if max_age <= 0:
|
|
return
|
|
cutoff = time.time() - max_age
|
|
try:
|
|
entries = list(parent.iterdir())
|
|
except OSError as error:
|
|
print(f"WARNING: could not scan {parent} for stale roots: {error}")
|
|
return
|
|
for entry in entries:
|
|
if not STALE_ROOT.match(entry.name):
|
|
continue
|
|
try:
|
|
if not entry.is_dir() or entry.stat().st_mtime > cutoff:
|
|
continue
|
|
except OSError:
|
|
continue
|
|
print(f"[reap] removing abandoned acceptance root {entry}")
|
|
shutil.rmtree(entry, ignore_errors=True)
|
|
|
|
|
|
def run_contract(
|
|
parent: Path,
|
|
label: str,
|
|
kind: str = REFERENCE,
|
|
daemon=None,
|
|
timeout: int = DEFAULT_TIMEOUT,
|
|
fail_fast: bool = False,
|
|
) -> Run:
|
|
reap_stale_roots(parent, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600")))
|
|
root = parent / f"pdfs-acceptance-{uuid.uuid4().hex}"
|
|
root.mkdir()
|
|
run = Run(label, kind, root)
|
|
REPORT.append(run)
|
|
print(f"[target] {label}: {parent}")
|
|
obs = Observations()
|
|
try:
|
|
for case in select_cases(kind):
|
|
obs.scope(case.name)
|
|
context = Context(root, obs, kind, daemon)
|
|
started = time.monotonic()
|
|
print(f" [test] {case.name}")
|
|
try:
|
|
with time_limit(timeout, case.name):
|
|
case.run(context)
|
|
except Skip as reason:
|
|
elapsed = time.monotonic() - started
|
|
print(f" SKIP {reason}")
|
|
run.results.append(Result(case.name, label, "skip", elapsed, str(reason)))
|
|
continue
|
|
except TestTimeout as reason:
|
|
elapsed = time.monotonic() - started
|
|
print(f" TIMEOUT {reason}")
|
|
run.results.append(Result(case.name, label, "timeout", elapsed, str(reason)))
|
|
# A timeout means the mount may be wedged; further cases would
|
|
# only produce noise, and cleanup already has to fight for it.
|
|
break
|
|
except BaseException as error: # noqa: BLE001 - reported, then continued
|
|
elapsed = time.monotonic() - started
|
|
detail = f"{type(error).__name__}: {error}"
|
|
print(f" FAIL {detail}")
|
|
run.results.append(Result(case.name, label, "fail", elapsed, detail))
|
|
if fail_fast:
|
|
raise
|
|
continue
|
|
elapsed = time.monotonic() - started
|
|
print(f" ok {elapsed:.2f}s")
|
|
run.results.append(Result(case.name, label, "pass", elapsed))
|
|
run.ran.add(case.name)
|
|
run.compared = dict(obs.compared)
|
|
run.noted = dict(obs.noted)
|
|
digest_path = root / "positioned.bin"
|
|
if digest_path.exists():
|
|
run.digest = hashlib.sha256(read(digest_path)).hexdigest()
|
|
return run
|
|
except BaseException:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
raise
|
|
|
|
|
|
def compare_runs(reference: Run, target: Run) -> list[str]:
|
|
"""Diff the mount's recorded semantics against the local filesystem's.
|
|
|
|
Only keys from cases that passed on both sides are compared: a case that
|
|
failed or was skipped has already been reported, and its partial
|
|
observations would just repeat that failure in a less useful form.
|
|
"""
|
|
shared = reference.ran & target.ran
|
|
divergences: list[str] = []
|
|
for key, expected in reference.compared.items():
|
|
case = key.split(".", 1)[0]
|
|
if case not in shared:
|
|
continue
|
|
if key not in target.compared:
|
|
divergences.append(f"{key}: missing on {target.label}")
|
|
continue
|
|
actual = target.compared[key]
|
|
if actual != expected:
|
|
divergences.append(f"{key}: {target.label}={actual!r} reference={expected!r}")
|
|
for key in target.compared:
|
|
case = key.split(".", 1)[0]
|
|
if case in shared and key not in reference.compared:
|
|
divergences.append(f"{key}: present on {target.label}, absent from reference")
|
|
return divergences
|
|
|
|
|
|
def note_capability_differences(reference: Run, target: Run) -> list[str]:
|
|
lines = []
|
|
for key, expected in reference.noted.items():
|
|
if key in target.noted and target.noted[key] != expected:
|
|
lines.append(f"{key}: {target.label}={target.noted[key]!r} reference={expected!r}")
|
|
return lines
|
|
|
|
|
|
def write_reports(json_path: Path | None, junit_path: Path | None) -> None:
|
|
if json_path:
|
|
payload = {
|
|
"runs": [
|
|
{
|
|
"label": run.label,
|
|
"kind": run.kind,
|
|
"results": [result.as_dict() for result in run.results],
|
|
"observations": {
|
|
key: _jsonable(value) for key, value in sorted(run.compared.items())
|
|
},
|
|
"notes": {key: _jsonable(value) for key, value in sorted(run.noted.items())},
|
|
}
|
|
for run in REPORT
|
|
]
|
|
}
|
|
json_path.write_text(json.dumps(payload, indent=2, sort_keys=True))
|
|
print(f"[report] wrote {json_path}")
|
|
if junit_path:
|
|
suites = ElementTree.Element("testsuites")
|
|
for run in REPORT:
|
|
suite = ElementTree.SubElement(
|
|
suites,
|
|
"testsuite",
|
|
name=run.label,
|
|
tests=str(len(run.results)),
|
|
failures=str(sum(1 for r in run.results if r.status in {"fail", "timeout"})),
|
|
skipped=str(sum(1 for r in run.results if r.status == "skip")),
|
|
time=f"{sum(r.seconds for r in run.results):.3f}",
|
|
)
|
|
for result in run.results:
|
|
case = ElementTree.SubElement(
|
|
suite,
|
|
"testcase",
|
|
name=result.name,
|
|
classname=run.label,
|
|
time=f"{result.seconds:.3f}",
|
|
)
|
|
if result.status in {"fail", "timeout"}:
|
|
failure = ElementTree.SubElement(case, "failure", type=result.status)
|
|
failure.text = result.message
|
|
elif result.status == "skip":
|
|
ElementTree.SubElement(case, "skipped", message=result.message)
|
|
ElementTree.ElementTree(suites).write(junit_path, encoding="unicode", xml_declaration=True)
|
|
print(f"[report] wrote {junit_path}")
|
|
|
|
|
|
def _jsonable(value):
|
|
if isinstance(value, (bytes, bytearray)):
|
|
return value.decode("utf-8", "backslashreplace")
|
|
if isinstance(value, (set, frozenset)):
|
|
return sorted(value)
|
|
return value
|
|
|
|
|
|
def report_timings(budget: float) -> list[str]:
|
|
slow = []
|
|
for run in REPORT:
|
|
for result in run.results:
|
|
if result.status == "pass" and budget and result.seconds > budget:
|
|
slow.append(f"{run.label}: {result.name} took {result.seconds:.1f}s (budget {budget:.0f}s)")
|
|
return slow
|
|
|
|
|
|
class JournalWatch:
|
|
"""Fail a run that leaves errors in the daemon's journal.
|
|
|
|
A suite can pass every assertion while the daemon logs a stream of failures
|
|
behind it; that has happened here before, and it was only noticed by reading
|
|
the journal by hand.
|
|
"""
|
|
|
|
def __init__(self, unit: str = "proton-drive.service") -> None:
|
|
self.unit = unit
|
|
self.since = 0.0
|
|
|
|
def start(self) -> None:
|
|
if shutil.which("journalctl") is None:
|
|
print("WARNING: journalctl is unavailable; skipping the journal check")
|
|
self.since = 0.0
|
|
return
|
|
self.since = time.time()
|
|
|
|
def errors(self) -> list[str]:
|
|
if not self.since:
|
|
return []
|
|
result = subprocess.run(
|
|
[
|
|
"journalctl",
|
|
"--user",
|
|
"-u",
|
|
self.unit,
|
|
"--since",
|
|
f"@{int(self.since)}",
|
|
"--no-pager",
|
|
"-o",
|
|
"cat",
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
if result.returncode:
|
|
return [f"journalctl failed: {result.stderr.strip()}"]
|
|
# Filtering on syslog priority (`-p err`) finds nothing: the daemon writes
|
|
# tracing levels as text on stdout, which journald files at the unit's
|
|
# default priority. Match the level token instead, after stripping the
|
|
# ANSI colouring tracing emits when it thinks it has a terminal.
|
|
errors = []
|
|
for line in result.stdout.splitlines():
|
|
plain = ANSI.sub("", line)
|
|
if ERROR_LEVEL.search(plain):
|
|
errors.append(plain.strip())
|
|
return errors
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Daemon control
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
class Daemon:
|
|
"""The `pdfs` CLI, plus the unit control the durability drill needs."""
|
|
|
|
def __init__(self, timeout: int, may_restart: bool = False) -> None:
|
|
self.timeout = timeout
|
|
self.may_restart = may_restart
|
|
self.pdfs = os.environ.get("PDFS_ACCEPTANCE_PDFS", "pdfs")
|
|
self.unit = os.environ.get("PDFS_ACCEPTANCE_UNIT", "proton-drive.service")
|
|
|
|
@classmethod
|
|
def discover(cls, timeout: int, may_restart: bool = False) -> Daemon | None:
|
|
daemon = cls(timeout, may_restart)
|
|
if shutil.which(daemon.pdfs) is None and not Path(daemon.pdfs).is_file():
|
|
return None
|
|
try:
|
|
daemon.status()
|
|
except Exception:
|
|
return None
|
|
return daemon
|
|
|
|
def command(self, *args: str, json_output: bool = False) -> str:
|
|
command = [self.pdfs]
|
|
if json_output:
|
|
command.append("--json")
|
|
command.extend(args)
|
|
result = subprocess.run(command, text=True, capture_output=True, timeout=self.timeout)
|
|
if result.returncode:
|
|
detail = result.stderr.strip() or result.stdout.strip()
|
|
raise RuntimeError(f"{' '.join(command)} failed: {detail}")
|
|
return result.stdout
|
|
|
|
def status(self) -> dict:
|
|
return json.loads(self.command("status", json_output=True))
|
|
|
|
def wait_for_queue(self) -> None:
|
|
deadline = time.monotonic() + self.timeout
|
|
last = None
|
|
while time.monotonic() < deadline:
|
|
last = self.status().get("mount") or {}
|
|
if last.get("pending_uploads", 0) == 0 and last.get("pending_changes", 0) == 0:
|
|
return
|
|
time.sleep(1)
|
|
raise TimeoutError(f"daemon mutation queue did not drain: {last}")
|
|
|
|
def enclosing_mount(self, path: Path) -> Path:
|
|
current = path.resolve()
|
|
while current != current.parent:
|
|
if os.path.ismount(current):
|
|
return current
|
|
current = current.parent
|
|
return current
|
|
|
|
def restart(self) -> None:
|
|
print(f"[restart] systemctl --user restart {self.unit}")
|
|
result = subprocess.run(
|
|
["systemctl", "--user", "restart", self.unit],
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=self.timeout,
|
|
)
|
|
if result.returncode:
|
|
raise RuntimeError(f"could not restart {self.unit}: {result.stderr.strip()}")
|
|
|
|
def wait_for_mount(self, mountpoint: Path) -> None:
|
|
deadline = time.monotonic() + self.timeout
|
|
while time.monotonic() < deadline:
|
|
if os.path.ismount(mountpoint):
|
|
with contextlib.suppress(OSError):
|
|
os.listdir(mountpoint)
|
|
return
|
|
time.sleep(1)
|
|
raise TimeoutError(f"{mountpoint} did not come back within {self.timeout}s")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Live and managed modes
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def wait_for_copy(root: Path, relative: Path, digest: str, timeout: int) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
target = root / relative
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
if target.is_file() and hashlib.sha256(read(target)).hexdigest() == digest:
|
|
return
|
|
except OSError:
|
|
pass
|
|
time.sleep(2)
|
|
raise AssertionError(f"{target} did not converge byte-for-byte within {timeout}s")
|
|
|
|
|
|
class ManagedSyncPair:
|
|
"""Two sync registrations created by this run and removed on exit."""
|
|
|
|
def __init__(self, paths: list[Path], timeout: int) -> None:
|
|
self.paths = [path.resolve() for path in paths]
|
|
self.timeout = timeout
|
|
self.pdfs = os.environ.get("PDFS_ACCEPTANCE_PDFS", "pdfs")
|
|
self.ids: dict[Path, int] = {}
|
|
self.sentinels = {
|
|
path: hashlib.sha256(f"pdfs-preservation:{path}:{uuid.uuid4()}".encode()).digest()
|
|
* 4096
|
|
for path in self.paths
|
|
}
|
|
|
|
def command(self, *args: str, json_output: bool = False) -> str:
|
|
command = [self.pdfs]
|
|
if json_output:
|
|
command.append("--json")
|
|
command.extend(args)
|
|
result = subprocess.run(command, text=True, capture_output=True)
|
|
if result.returncode:
|
|
detail = result.stderr.strip() or result.stdout.strip()
|
|
raise RuntimeError(f"{' '.join(command)} failed: {detail}")
|
|
return result.stdout
|
|
|
|
def folders(self) -> list[dict]:
|
|
value = json.loads(self.command("sync", "list", json_output=True))
|
|
return value["items"]
|
|
|
|
def wait_for_queue(self) -> None:
|
|
deadline = time.monotonic() + self.timeout
|
|
last = None
|
|
while time.monotonic() < deadline:
|
|
value = json.loads(self.command("status", json_output=True))
|
|
last = value.get("mount") or {}
|
|
if last.get("pending_uploads", 0) == 0 and last.get("pending_changes", 0) == 0:
|
|
return
|
|
time.sleep(1)
|
|
raise TimeoutError(f"daemon mutation queue did not drain: {last}")
|
|
|
|
def remove_tree(self, root: Path) -> None:
|
|
deadline = time.monotonic() + self.timeout
|
|
while True:
|
|
try:
|
|
shutil.rmtree(root)
|
|
return
|
|
except FileNotFoundError:
|
|
return
|
|
except OSError as error:
|
|
if error.errno != errno.ENOTEMPTY or time.monotonic() >= deadline:
|
|
survivors = []
|
|
if root.exists():
|
|
for directory, dirs, files in os.walk(root):
|
|
survivors.extend(str(Path(directory) / name) for name in dirs + files)
|
|
raise OSError(
|
|
error.errno,
|
|
f"{error.strerror}; surviving test entries: {survivors[:50]}",
|
|
error.filename,
|
|
) from error
|
|
# A queued namespace operation can land between rmtree's
|
|
# enumeration and rmdir. Re-walk the test-owned tree.
|
|
time.sleep(0.25)
|
|
|
|
def validate(self) -> None:
|
|
if shutil.which(self.pdfs) is None and not Path(self.pdfs).is_file():
|
|
raise RuntimeError(
|
|
f"cannot find {self.pdfs!r}; set PDFS_ACCEPTANCE_PDFS to the CLI binary"
|
|
)
|
|
if self.paths[0] == self.paths[1]:
|
|
raise ValueError("managed live paths must be different directories")
|
|
deadline = time.monotonic() + self.timeout
|
|
last_error = None
|
|
while True:
|
|
try:
|
|
current = self.folders()
|
|
break
|
|
except Exception as error:
|
|
last_error = error
|
|
if time.monotonic() >= deadline:
|
|
raise TimeoutError(
|
|
f"pdfs daemon did not become ready within {self.timeout}s: {last_error}"
|
|
) from error
|
|
time.sleep(1)
|
|
registered = {Path(item["local_path"]).resolve() for item in current}
|
|
for path in self.paths:
|
|
reap_stale_roots(path, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600")))
|
|
check(path.is_dir(), f"managed path is not a directory: {path}")
|
|
check(not any(path.iterdir()), f"managed path is not empty: {path}")
|
|
check(path not in registered, f"managed path is already registered: {path}")
|
|
check(not is_mountpoint(path), f"managed path is already a mount: {path}")
|
|
|
|
def wait_for(self, path: Path, *, mode: str | None = None) -> dict:
|
|
deadline = time.monotonic() + self.timeout
|
|
last = None
|
|
while time.monotonic() < deadline:
|
|
for item in self.folders():
|
|
if Path(item["local_path"]).resolve() != path:
|
|
continue
|
|
last = item
|
|
if (
|
|
item["state"] == "idle"
|
|
and item.get("pending_mode") is None
|
|
and (mode is None or item["mode"] == mode)
|
|
):
|
|
return item
|
|
if item["state"] in {"error", "conflict"}:
|
|
raise RuntimeError(f"sync folder {path} entered {item['state']}: {item}")
|
|
time.sleep(2)
|
|
raise TimeoutError(f"sync folder {path} did not become idle in mode {mode}: {last}")
|
|
|
|
def create(self) -> None:
|
|
self.validate()
|
|
for path in self.paths:
|
|
print(f"[setup] registering empty sync folder {path}")
|
|
self.command("sync", "add", str(path))
|
|
item = self.wait_for(path, mode="mirror")
|
|
self.ids[path] = int(item["id"])
|
|
for path in self.paths:
|
|
write_durable(path / "pdfs-mode-preservation.bin", self.sentinels[path])
|
|
self.force_sync(path)
|
|
|
|
def force_sync(self, path: Path) -> None:
|
|
before = int(self.wait_for(path)["last_sync"])
|
|
# last_sync has one-second resolution. Ensure this requested pass cannot
|
|
# finish with the same timestamp and look indistinguishable from no pass.
|
|
while int(time.time()) <= before:
|
|
time.sleep(0.1)
|
|
self.command("sync", "now", str(self.ids[path]))
|
|
deadline = time.monotonic() + self.timeout
|
|
last = None
|
|
while time.monotonic() < deadline:
|
|
last = next(
|
|
(item for item in self.folders() if Path(item["local_path"]).resolve() == path),
|
|
None,
|
|
)
|
|
if last and last["state"] == "idle" and int(last["last_sync"]) > before:
|
|
return
|
|
if last and last["state"] in {"error", "conflict"}:
|
|
raise RuntimeError(f"forced sync for {path} entered {last['state']}: {last}")
|
|
time.sleep(1)
|
|
raise TimeoutError(f"forced sync for {path} did not complete: {last}")
|
|
|
|
def set_modes(self, modes: tuple[str, str]) -> None:
|
|
changed_to_mirror: set[Path] = set()
|
|
for path, mode in zip(self.paths, modes, strict=True):
|
|
current = self.wait_for(path)
|
|
if current["mode"] != mode:
|
|
print(f"[setup] switching {path} to {mode}")
|
|
self.command("sync", "mode", str(self.ids[path]), mode)
|
|
if mode == "mirror":
|
|
changed_to_mirror.add(path)
|
|
for path, mode in zip(self.paths, modes, strict=True):
|
|
self.wait_for(path, mode=mode)
|
|
# The mode row flips before the asynchronous restore pass starts,
|
|
# and its prior idle/last_sync values remain visible meanwhile.
|
|
# Demand a completed pass before inspecting restored local bytes.
|
|
if path in changed_to_mirror:
|
|
self.force_sync(path)
|
|
mounted = is_mountpoint(path)
|
|
check(mounted == (mode == "ondemand"), f"{path}: mode is {mode}, mounted={mounted}")
|
|
check(
|
|
read(path / "pdfs-mode-preservation.bin") == self.sentinels[path],
|
|
f"{path}: preservation sentinel changed or vanished after switch to {mode}",
|
|
)
|
|
|
|
def cleanup(self) -> None:
|
|
# An async add can create its row and then time out before create() has
|
|
# recorded the id. validate() proved these exact paths were unregistered
|
|
# at entry, so rows now at those paths belong to this run.
|
|
try:
|
|
for item in self.folders():
|
|
path = Path(item["local_path"]).resolve()
|
|
if path in self.paths:
|
|
self.ids[path] = int(item["id"])
|
|
except Exception as error:
|
|
print(f"WARNING: could not rediscover managed sync folders: {error}")
|
|
for path, folder_id in reversed(list(self.ids.items())):
|
|
try:
|
|
print(f"[cleanup] removing test sync folder {path}")
|
|
self.command("sync", "rm", str(folder_id), "--delete-remote")
|
|
except Exception as error:
|
|
print(f"WARNING: cleanup failed for sync folder {folder_id}: {error}")
|
|
continue
|
|
# Unregistering a mirror intentionally preserves its local copy.
|
|
# Remove only the sentinel owned by this harness so the next run's
|
|
# strict empty-directory precondition remains meaningful. Unknown
|
|
# survivors are left untouched and will fail validate() next time.
|
|
sentinel = path / "pdfs-mode-preservation.bin"
|
|
try:
|
|
if sentinel.exists() and read(sentinel) == self.sentinels[path]:
|
|
sentinel.unlink()
|
|
except OSError as error:
|
|
print(f"WARNING: could not remove managed sentinel {sentinel}: {error}")
|
|
|
|
|
|
def is_mountpoint(path: Path) -> bool:
|
|
"""True for a distinct mount, including FUSE mounts over an existing dir."""
|
|
return os.path.ismount(path)
|
|
|
|
|
|
def run_managed_matrix(paths: list[Path], reference: Run, args) -> None:
|
|
timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "180"))
|
|
pair = ManagedSyncPair(paths, timeout)
|
|
daemon = Daemon.discover(timeout, may_restart=args.durability)
|
|
roots: list[Path] = []
|
|
try:
|
|
pair.create()
|
|
matrices = [
|
|
("ondemand", "ondemand"),
|
|
("ondemand", "mirror"),
|
|
("mirror", "ondemand"),
|
|
("mirror", "mirror"),
|
|
]
|
|
for modes in matrices:
|
|
print(f"[matrix] first={modes[0]}, second={modes[1]}")
|
|
pair.set_modes(modes)
|
|
for path, mode in zip(pair.paths, modes, strict=True):
|
|
run = run_contract(
|
|
path,
|
|
f"managed {mode} {path.name}",
|
|
kind=LIVE,
|
|
daemon=daemon,
|
|
timeout=args.timeout,
|
|
fail_fast=args.fail_fast,
|
|
)
|
|
roots.append(run.root)
|
|
summarize_divergence(reference, run)
|
|
pair.wait_for_queue()
|
|
pair.remove_tree(run.root)
|
|
roots.remove(run.root)
|
|
pair.wait_for_queue()
|
|
# Mirror changes are asynchronous; force and await a pass before
|
|
# changing its mode so authored bytes/deletions cannot be lost.
|
|
if mode == "mirror":
|
|
pair.force_sync(path)
|
|
finally:
|
|
for root in roots:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
pair.cleanup()
|
|
|
|
|
|
def summarize_divergence(reference: Run, target: Run) -> None:
|
|
divergences = compare_runs(reference, target)
|
|
capabilities = note_capability_differences(reference, target)
|
|
if capabilities:
|
|
print(f" [capabilities] {target.label} differs from the reference filesystem:")
|
|
for line in capabilities:
|
|
print(f" - {line}")
|
|
if divergences:
|
|
print(f" [DIVERGENCE] {target.label} does not match the reference filesystem:")
|
|
for line in divergences:
|
|
print(f" ! {line}")
|
|
target.results.append(
|
|
Result(
|
|
"differential comparison against the reference filesystem",
|
|
target.label,
|
|
"fail",
|
|
0.0,
|
|
"; ".join(divergences),
|
|
)
|
|
)
|
|
else:
|
|
print(f" [differential] {target.label} matches the reference filesystem")
|
|
target.results.append(
|
|
Result(
|
|
"differential comparison against the reference filesystem",
|
|
target.label,
|
|
"pass",
|
|
0.0,
|
|
)
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
live_mode = parser.add_mutually_exclusive_group()
|
|
live_mode.add_argument("--live", nargs="+", type=Path, metavar="MOUNTPOINT")
|
|
live_mode.add_argument(
|
|
"--managed-live",
|
|
nargs=2,
|
|
type=Path,
|
|
metavar=("EMPTY_DIR_A", "EMPTY_DIR_B"),
|
|
)
|
|
parser.add_argument("--list", action="store_true", help="print case names and exit")
|
|
parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="per-case seconds")
|
|
parser.add_argument("--fail-fast", action="store_true", help="stop at the first failure")
|
|
parser.add_argument("--report-json", type=Path, metavar="PATH")
|
|
parser.add_argument("--report-junit", type=Path, metavar="PATH")
|
|
parser.add_argument(
|
|
"--budget",
|
|
type=float,
|
|
default=float(os.environ.get("PDFS_ACCEPTANCE_BUDGET", "0")),
|
|
help="report any case slower than this many seconds",
|
|
)
|
|
parser.add_argument(
|
|
"--journal-check",
|
|
action="store_true",
|
|
help="fail if the daemon logs errors during the run",
|
|
)
|
|
parser.add_argument(
|
|
"--durability",
|
|
action="store_true",
|
|
help="restart the daemon mid-suite and verify written bytes survive",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.list:
|
|
for case in TESTS:
|
|
print(f"{','.join(case.kinds):<18} {case.name}")
|
|
return 0
|
|
|
|
journal = JournalWatch(os.environ.get("PDFS_ACCEPTANCE_UNIT", "proton-drive.service"))
|
|
if args.journal_check:
|
|
journal.start()
|
|
|
|
with tempfile.TemporaryDirectory(prefix="pdfs-api-reference-") as directory:
|
|
reference = run_contract(
|
|
Path(directory),
|
|
"local filesystem API reference (no account)",
|
|
kind=REFERENCE,
|
|
timeout=args.timeout,
|
|
fail_fast=args.fail_fast,
|
|
)
|
|
shutil.rmtree(reference.root, ignore_errors=True)
|
|
if reference.failed:
|
|
print("FAIL: the account-free contract failed against an ordinary filesystem")
|
|
else:
|
|
print("[pass] account-free filesystem API contract")
|
|
|
|
try:
|
|
if args.managed_live:
|
|
run_managed_matrix(args.managed_live, reference, args)
|
|
elif args.live:
|
|
run_live(args.live, reference, args)
|
|
finally:
|
|
outcome = finish(args, journal)
|
|
return outcome
|
|
|
|
|
|
def run_live(mountpoints: list[Path], reference: Run, args) -> None:
|
|
completed: list[tuple[Path, Run]] = []
|
|
convergence = os.environ.get("PDFS_ACCEPTANCE_CONVERGENCE", "0") == "1"
|
|
timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "120"))
|
|
daemon = Daemon.discover(timeout, may_restart=args.durability)
|
|
if daemon is None:
|
|
print("WARNING: no reachable pdfs daemon; sync regressions will be skipped")
|
|
try:
|
|
# Views of one remote folder must not independently create the same
|
|
# fixtures. Exercise the primary, then observe that exact tree through
|
|
# every secondary. Without convergence mode, each mount is independent
|
|
# and receives the full contract.
|
|
targets = mountpoints[:1] if convergence and len(mountpoints) > 1 else mountpoints
|
|
for mountpoint in targets:
|
|
run = run_contract(
|
|
mountpoint.resolve(),
|
|
f"live FUSE {mountpoint}",
|
|
kind=LIVE,
|
|
daemon=daemon,
|
|
timeout=args.timeout,
|
|
fail_fast=args.fail_fast,
|
|
)
|
|
completed.append((mountpoint.resolve(), run))
|
|
summarize_divergence(reference, run)
|
|
if convergence and len(mountpoints) > 1:
|
|
print(" [test] cross-mount remote convergence")
|
|
source = completed[0][1]
|
|
relative = source.root.relative_to(completed[0][0]) / "positioned.bin"
|
|
for mountpoint in mountpoints[1:]:
|
|
wait_for_copy(mountpoint.resolve(), relative, source.digest, timeout)
|
|
finally:
|
|
for _, run in completed:
|
|
shutil.rmtree(run.root, ignore_errors=True)
|
|
|
|
|
|
def finish(args, journal: JournalWatch) -> int:
|
|
write_reports(args.report_json, args.report_junit)
|
|
|
|
slow = report_timings(args.budget)
|
|
if slow:
|
|
print("[timing] cases over budget:")
|
|
for line in slow:
|
|
print(f" - {line}")
|
|
|
|
journal_errors = journal.errors() if args.journal_check else []
|
|
if journal_errors:
|
|
print(f"[journal] the daemon logged {len(journal_errors)} error(s) during the run:")
|
|
for line in journal_errors[:20]:
|
|
print(f" ! {line}")
|
|
|
|
failures = [
|
|
(run.label, result)
|
|
for run in REPORT
|
|
for result in run.results
|
|
if result.status in {"fail", "timeout"}
|
|
]
|
|
print()
|
|
for run in REPORT:
|
|
counts: dict[str, int] = {}
|
|
for result in run.results:
|
|
counts[result.status] = counts.get(result.status, 0) + 1
|
|
detail = ", ".join(f"{count} {status}" for status, count in sorted(counts.items()))
|
|
print(f"[summary] {run.label}: {detail}")
|
|
|
|
if failures or journal_errors:
|
|
for label, result in failures:
|
|
print(f"FAIL {label}: {result.name}: {result.message}")
|
|
return 1
|
|
print("PASS: acceptance suite completed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except KeyboardInterrupt:
|
|
print("\ninterrupted", file=sys.stderr)
|
|
sys.exit(130)
|
|
except BrokenPipeError:
|
|
os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
|
|
sys.exit(0)
|
|
except (AssertionError, RuntimeError, TimeoutError, ValueError) as failure:
|
|
# Setup and selection problems are the operator's to fix; a traceback
|
|
# buries the one line that says what to change.
|
|
print(f"error: {failure}", file=sys.stderr)
|
|
sys.exit(2)
|