Detect Linux volumes from the mount table

CONTRIBUTING.md

@@ -37,7 +37,7 @@ pip install -e ".[dev]"
 `ffmpeg` and `ffprobe` on `PATH` are optional — the suite runs without them.
 
 ```sh
-pytest                      # ~400 tests, about 33s
+pytest                      # ~416 tests, about 33s
 pytest --fuzz               # property tests at 3000 examples each, about 2 min
 ruff check src tests
 pytest --cov=offloader --cov-report=term-missing

README.md

@@ -356,13 +356,13 @@ what makes the report layer testable without moving bytes.
 
 ```sh
 pip install -e ".[dev]"
-pytest                      # 409 tests, ~33s
+pytest                      # 416 tests, ~33s
 pytest --fuzz               # same suite, 3000 examples per property (~2 min)
 ruff check src tests
 pytest --cov=offloader --cov-report=term-missing
 ```
 
-409 tests at 82% line coverage. They cover formatting against the reference's
+416 tests at 82% line coverage. They cover formatting against the reference's
 exact strings, checksum vectors and streaming equivalence, copy/verify
 behaviour including simulated destination corruption, pause/resume/cancel
 concurrency, retry discrimination, BRAW container parsing, ffprobe parsing,

src/offloader/volumes.py

@@ -8,6 +8,7 @@
 
 import os
 import platform
+import re
 import shutil
 import string
 from dataclasses import dataclass
@@ -205,15 +206,134 @@ def _posix_volumes() -> list[Volume]:
     return found
 
 
+#: Linux filesystems that are not storage: kernel interfaces, containers, snap
+#: loop images, and per-user FUSE plumbing (gvfs, portals).
+_LINUX_PSEUDO_FS = {
+    "proc", "sysfs", "devtmpfs", "devpts", "tmpfs", "ramfs", "cgroup", "cgroup2",
+    "overlay", "squashfs", "autofs", "securityfs", "debugfs", "tracefs", "fusectl",
+    "configfs", "pstore", "bpf", "mqueue", "hugetlbfs", "binfmt_misc", "rpc_pipefs",
+    "efivarfs", "nsfs", "fuse.gvfsd-fuse", "fuse.portal", "fuse.snapfuse",
+    "fuse.lxcfs", "fuse.gvfs-fuse-daemon",
+}
+_LINUX_NETWORK_FS = {
+    "nfs", "nfs4", "cifs", "smb3", "smbfs", "9p", "ceph", "glusterfs", "afs",
+    "davfs", "fuse.sshfs", "fuse.rclone",
+}
+_LINUX_OPTICAL_FS = {"iso9660", "udf"}
+_MOUNTINFO = "/proc/self/mountinfo"
+
+#: Mounts under these belong to the OS, not to a person plugging something in.
+_LINUX_SYSTEM_PREFIXES = (
+    "/boot", "/snap", "/var/snap", "/var/lib", "/usr", "/etc", "/proc", "/sys",
+    "/dev", "/run",
+)
+#: Where udisks and friends put removable media. Exempt from the list above,
+#: because /run/media sits under /run.
+_LINUX_MEDIA_PREFIXES = ("/media", "/run/media", "/mnt")
+
+
+def _unescape_mountinfo(field: str) -> str:
+    """mountinfo writes space, tab, newline and backslash as octal escapes."""
+    return re.sub(r"\\([0-7]{3})", lambda m: chr(int(m.group(1), 8)), field)
+
+
+def _under(path: str, prefixes: tuple[str, ...]) -> bool:
+    return any(path == p or path.startswith(p + "/") for p in prefixes)
+
+
+def _linux_device_is_removable(major_minor: str) -> bool:
+    """Whether the block device behind a mount reports itself as removable.
+
+    A partition's `removable` flag lives on its parent disk. Device-mapper
+    devices (LUKS) have none, and read as not removable.
+    """
+    try:
+        device = (Path("/sys/dev/block") / major_minor).resolve()
+    except OSError:
+        return False
+    for candidate in (device, device.parent):
+        try:
+            return (candidate / "removable").read_text().strip() == "1"
+        except OSError:
+            continue
+    return False
+
+
+def _linux_volumes() -> list[Volume]:
+    """Volumes from the kernel's mount table.
+
+    Cards mount at /media/<user>/<LABEL> (or /run/media/<user>/<LABEL>), one
+    level below the directory the macOS-style scan in `_posix_volumes` looks
+    at, so listing children of /media finds the user's directory and never the
+    card. The mount table names the real mount points, and gives the
+    filesystem type as well.
+    """
+    with open(_MOUNTINFO, encoding="utf-8", errors="replace") as handle:
+        lines = handle.read().splitlines()
+
+    found: list[Volume] = []
+    for line in lines:
+        head, separator, tail = line.partition(" - ")
+        left, right = head.split(), tail.split()
+        if not separator or len(left) < 5 or len(right) < 2:
+            continue
+        major_minor, mount_point = left[2], _unescape_mountinfo(left[4])
+        fstype = right[0]
+
+        # The system root is always a candidate, even on overlayfs (containers,
+        # live systems): it is filtered only by the usage check below.
+        if fstype in _LINUX_PSEUDO_FS and mount_point != "/":
+            continue
+        removable_path = _under(mount_point, _LINUX_MEDIA_PREFIXES)
+        if mount_point != "/" and not removable_path and _under(
+                mount_point, _LINUX_SYSTEM_PREFIXES):
+            continue
+
+        root = Path(mount_point)
+        total, free = _usage(root)
+        if total == 0:
+            continue
+
+        if fstype in _LINUX_NETWORK_FS:
+            drive_type = "network"
+        elif fstype in _LINUX_OPTICAL_FS:
+            drive_type = "optical"
+        elif mount_point == "/":
+            drive_type = "fixed"
+        elif removable_path or _linux_device_is_removable(major_minor):
+            drive_type = "removable"
+        else:
+            drive_type = "fixed"
+
+        found.append(Volume(
+            root=root,
+            label=root.name or str(root),
+            filesystem=fstype,
+            total_bytes=total,
+            free_bytes=free,
+            drive_type=drive_type,
+            is_camera_card=detect_camera_card(root, drive_type),
+        ))
+    return found
+
+
 def list_volumes() -> list[Volume]:
     """Every mounted volume, cards first so they are easy to spot.
 
     Deduplicated by resolved root: macOS reaches the boot volume through both
     `/` and `/Volumes/Macintosh HD`, and listing it twice would be noise.
     """
+    system = platform.system()
     try:
-        volumes = (_windows_volumes() if platform.system() == "Windows"
-                   else _posix_volumes())
+        if system == "Windows":
+            volumes = _windows_volumes()
+        elif system == "Linux":
+            try:
+                volumes = _linux_volumes()
+            except OSError:      # no /proc (some containers): scan the usual parents
+                volumes = _posix_volumes()
+        else:
+            volumes = _posix_volumes()
     except Exception:
         volumes = []
 

tests/test_volumes.py

@@ -103,3 +103,88 @@ def test_a_firmlinked_system_volume_is_not_a_card(tmp_path: Path, monkeypatch):
     monkeypatch.setattr(volumes, "system_root", lambda: system)
     assert not volumes.detect_camera_card(system, "fixed")
     assert not volumes.detect_camera_card(alias, "removable")
+
+
+_MOUNTINFO_SAMPLE = r"""\
+26 1 259:2 / / rw,relatime shared:1 - ext4 /dev/nvme0n1p2 rw
+27 26 0:6 / /dev rw - devtmpfs udev rw
+28 26 0:24 / /run rw - tmpfs tmpfs rw
+29 26 259:1 / /boot/efi rw - vfat /dev/nvme0n1p1 rw
+30 26 259:2 /var/lib/snapd/hostfs/x /var/snap/firefox/common/host-hunspell rw - ext4 /dev/nvme0n1p2 rw
+31 28 0:50 / /run/user/1000/gvfs rw - fuse.gvfsd-fuse gvfsd-fuse rw
+32 26 8:17 / /media/owen/A001 rw - exfat /dev/sdb1 rw
+33 26 8:33 / /media/owen/BRAW\040CARD rw - vfat /dev/sdc1 rw
+34 26 8:49 / /mnt/scratch rw - ext4 /dev/sdd1 rw
+35 26 0:60 / /run/media/owen/SD_CARD rw - exfat /dev/sde1 rw
+36 26 8:65 / /home/owen/raid rw - ext4 /dev/sdf1 rw
+37 26 0:70 / /srv/nas rw - nfs4 nas:/export rw
+38 26 11:0 / /media/owen/DISC rw - iso9660 /dev/sr0 ro
+39 26 7:1 / /snap/core/1 ro - squashfs /dev/loop1 ro
+"""
+
+
+@pytest.fixture
+def linux_mounts(tmp_path: Path, monkeypatch):
+    table = tmp_path / "mountinfo"
+    table.write_text(_MOUNTINFO_SAMPLE)
+    monkeypatch.setattr(volumes, "_MOUNTINFO", str(table))
+    monkeypatch.setattr(volumes, "_usage", lambda root: (1000, 400))
+    monkeypatch.setattr(
+        volumes, "_linux_device_is_removable", lambda major_minor: major_minor == "8:65")
+    return {v.root.as_posix(): v for v in volumes._linux_volumes()}
+
+
+def test_linux_finds_cards_below_the_users_media_directory(linux_mounts):
+    card = linux_mounts["/media/owen/A001"]
+    assert (card.label, card.filesystem, card.drive_type) == ("A001", "exfat", "removable")
+    assert "/run/media/owen/SD_CARD" in linux_mounts
+    assert "/media/owen" not in linux_mounts
+
+
+def test_linux_decodes_escaped_mount_points(linux_mounts):
+    assert linux_mounts["/media/owen/BRAW CARD"].label == "BRAW CARD"
+
+
+def test_linux_skips_pseudo_and_system_mounts(linux_mounts):
+    assert set(linux_mounts) == {
+        "/", "/media/owen/A001", "/media/owen/BRAW CARD", "/media/owen/DISC",
+        "/mnt/scratch", "/run/media/owen/SD_CARD", "/home/owen/raid", "/srv/nas",
+    }
+
+
+def test_linux_classifies_drive_types(linux_mounts):
+    assert linux_mounts["/"].drive_type == "fixed"
+    assert linux_mounts["/srv/nas"].drive_type == "network"
+    assert linux_mounts["/media/owen/DISC"].drive_type == "optical"
+    # Outside /media, only the device's own removable flag makes it removable.
+    assert linux_mounts["/home/owen/raid"].drive_type == "removable"
+    assert linux_mounts["/mnt/scratch"].drive_type == "removable"
+
+
+def test_unescape_mountinfo():
+    assert volumes._unescape_mountinfo(r"/a\040b\134c") == "/a b\\c"
+
+
+def _mount_table(tmp_path, monkeypatch, text):
+    table = tmp_path / "mountinfo"
+    table.write_text(text)
+    monkeypatch.setattr(volumes, "_MOUNTINFO", str(table))
+    monkeypatch.setattr(volumes, "_usage", lambda root: (1000, 400))
+    monkeypatch.setattr(volumes, "_linux_device_is_removable", lambda major_minor: False)
+    return {v.root.as_posix(): v for v in volumes._linux_volumes()}
+
+
+def test_linux_keeps_an_overlay_root_as_the_system_volume(tmp_path, monkeypatch):
+    found = _mount_table(tmp_path, monkeypatch,
+                         "1 0 0:50 / / rw - overlay overlay rw,lowerdir=/a\n")
+    assert set(found) == {"/"}
+    assert (found["/"].filesystem, found["/"].drive_type) == ("overlay", "fixed")
+
+
+def test_linux_still_skips_overlay_mounts_elsewhere(tmp_path, monkeypatch):
+    found = _mount_table(
+        tmp_path, monkeypatch,
+        "1 0 0:50 / / rw - overlay overlay rw\n"
+        "2 1 0:51 / /var/lib/docker/overlay2/x/merged rw - overlay overlay rw\n"
+        "3 1 0:52 / /mnt/ctr rw - overlay overlay rw\n")
+    assert set(found) == {"/"}