diff --git a/coe_mirror.py b/coe_mirror.py index 83540a2..393af89 100644 --- a/coe_mirror.py +++ b/coe_mirror.py @@ -81,6 +81,8 @@ PENDING_DIR: Path = Path( str(Path.home() / ".local" / "share" / "coe-mirror" / "pending"), ) ).expanduser() +_MOUNT_ROOT_VAL = _resolve("COE_MIRROR_MOUNT_ROOT", _CFG.get("mount_root"), "") +MOUNT_ROOT: Optional[Path] = Path(_MOUNT_ROOT_VAL).expanduser() if _MOUNT_ROOT_VAL else None LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror" LOG_FILE = LOG_DIR / "coe-mirror.log" LOCK_NAME = "coe-mirror.lock" @@ -168,6 +170,14 @@ def release_lock(lock_fd: int) -> None: def probe_target() -> bool: """Probe target directory health. Write tiny file, read back, delete. Return True if healthy.""" try: + if MOUNT_ROOT is not None: + if not MOUNT_ROOT.exists() or not os.path.ismount(str(MOUNT_ROOT)): + logger.warning("mount_root %s is not a live mountpoint; target unavailable; queuing locally", MOUNT_ROOT) + return False + if not (TARGET_DIR == MOUNT_ROOT or TARGET_DIR.is_relative_to(MOUNT_ROOT)): + logger.warning("target_dir %s is not under configured mount_root %s; misconfig; queuing locally", TARGET_DIR, MOUNT_ROOT) + return False + probe_file = TARGET_DIR / ".write_probe" TARGET_DIR.mkdir(parents=True, exist_ok=True) @@ -673,6 +683,11 @@ api_base = "https://git.churchofmalware.org" # Where mirrored zips are stored. One subdirectory per owner is created here. target_dir = "~/coe-mirror" +# Optional: require this path to be a live mountpoint before writing. If set and +# not currently mounted, the run queues to pending instead of creating a phantom +# local directory at the mountpoint. Leave blank to disable. +# mount_root = "~/pCloudDrive" + # Optional: local staging when target_dir is unavailable. # pending_dir = "~/.local/share/coe-mirror/pending" diff --git a/tests/test_coe_mirror.py b/tests/test_coe_mirror.py index 9f397f6..4b74b2e 100644 --- a/tests/test_coe_mirror.py +++ b/tests/test_coe_mirror.py @@ -595,6 +595,51 @@ class TestPlaceIntoPloudSameFileIsNoop: assert test_file.stat().st_size > 0 +class TestMountRootGuard: + """Test probe_target() mount-root verification (phantom-dir regression guard).""" + + def test_no_mount_root_configured_unchanged_behavior(self, mock_paths, setup_logging, monkeypatch): + """MOUNT_ROOT None + writable TARGET_DIR -> True (unchanged behavior).""" + monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", None) + + assert coe_mirror.probe_target() is True + + def test_mount_root_live_target_under_it_returns_true(self, mock_paths, setup_logging, monkeypatch, tmp_path): + """MOUNT_ROOT set, live mount, TARGET_DIR a subdir of it -> True.""" + mount_root = tmp_path / "pCloudDrive" + mount_root.mkdir() + target_dir = mount_root / "Hacking" / "ChurchOfMalware" + monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", mount_root) + monkeypatch.setattr(coe_mirror, "TARGET_DIR", target_dir) + monkeypatch.setattr(coe_mirror.os.path, "ismount", lambda p: True) + + assert coe_mirror.probe_target() is True + + def test_mount_not_live_does_not_create_phantom_dir(self, mock_paths, setup_logging, monkeypatch, tmp_path): + """MOUNT_ROOT set but not mounted -> False, and TARGET_DIR is never created.""" + mount_root = tmp_path / "pCloudDrive" + mount_root.mkdir() + target_dir = mount_root / "Hacking" / "ChurchOfMalware" + monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", mount_root) + monkeypatch.setattr(coe_mirror, "TARGET_DIR", target_dir) + monkeypatch.setattr(coe_mirror.os.path, "ismount", lambda p: False) + + assert coe_mirror.probe_target() is False + assert not target_dir.exists() + + def test_target_dir_not_under_mount_root_returns_false(self, mock_paths, setup_logging, monkeypatch, tmp_path): + """MOUNT_ROOT set, live mount, but TARGET_DIR is unrelated -> False, no dir created.""" + mount_root = tmp_path / "pCloudDrive" + mount_root.mkdir() + target_dir = tmp_path / "unrelated" / "target" + monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", mount_root) + monkeypatch.setattr(coe_mirror, "TARGET_DIR", target_dir) + monkeypatch.setattr(coe_mirror.os.path, "ismount", lambda p: True) + + assert coe_mirror.probe_target() is False + assert not target_dir.exists() + + # Tests for Matrix notification implementation. # Note: Tests requiring monkeypatched subprocess.run must use raising=False because # subprocess.run is not a class attribute on the coe_mirror module by default