#!/usr/bin/env python3
#
# Copyright (c) 2025-2026 openmediavault plugin developers
#
# This file is licensed under the terms of the GNU General Public
# License version 2. This program is licensed "as is" without any
# warranty of any kind, whether express or implied.
#
# version: 0.2.2
#
# Test script for omv-writecache
# Uses the config at /etc/omv-writecache/config.yaml and tests
# mount, flush, and unmount operations with the configured policies.
#
import os
import sys
import time
import errno
import shutil
import subprocess
import hashlib
import json
import tempfile
import stat
import threading
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Tuple, Optional, Set

# -------- Settings --------
CONFIG_PATH = Path("/etc/omv-writecache/config.yaml")
TEST_LOG = Path("/tmp/omv-writecache-test.log")
SCRIPT_PATH = "/usr/sbin/omv-writecache"
TEST_VERSION = "0.2.2"

# Optional: only create test files in these paths (leave empty to allow all)
ALLOWED_FOR_FILE_CREATION = set()  # e.g., {"/var/log", "/var/tmp"} to restrict

# -------- Colors / logging --------
class Colors:
    RED = '\033[91m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    MAGENTA = '\033[95m'
    CYAN = '\033[96m'
    WHITE = '\033[97m'
    RESET = '\033[0m'
    BOLD = '\033[1m'

def colored(text: str, color: str) -> str:
    if sys.stdout.isatty():
        return f"{color}{text}{Colors.RESET}"
    return text

def run_sh(cmd, timeout: int = 20):
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False)
        return (r.returncode == 0, r.stdout, r.stderr)
    except Exception as e:
        return (False, "", str(e))

def postfix_available() -> bool:
    """Return True if postfix binary is present."""
    return shutil.which("postfix") is not None

def postfix_unit_disabled() -> bool:
    """
    Return True if the postfix systemd unit is explicitly disabled.
    """
    ok, out, err = run_sh(["systemctl", "is-enabled", "postfix"], timeout=5)
    state = (out or "").strip()
    return state == "disabled"

def check_postfix(stage: str) -> tuple[bool, str]:
    """
    Check Postfix health after a given stage (mount/flush/unmount/baseline).

    We treat as 'healthy' if:
      - postfix is not installed, OR
      - postfix systemd unit is disabled (explicitly skipped), OR
      - postfix status / postqueue / postfix check all succeed.
    """
    if postfix_unit_disabled():
        msg = "Postfix systemd unit is disabled; skipping postfix checks."
        log_test(msg + f" (stage={stage})", "WARN")
        return True, msg

    overall_ok = True
    messages: list[str] = []

    # 1) postfix status
    ok_status, out_status, err_status = run_sh(["postfix", "status"], timeout=10)
    msg_status = (out_status or err_status or "").strip()
    if ok_status:
        log_test(f"[Postfix] status OK after {stage}: {msg_status}", "SUCCESS")
    else:
        log_test(f"[Postfix] status FAILED after {stage}: {msg_status}", "FAIL")
        overall_ok = False
    messages.append(f"status: {'OK' if ok_status else 'FAIL'} ({msg_status})")

    # 2) postqueue -p (queue listing)
    ok_queue, out_queue, err_queue = run_sh(["postqueue", "-p"], timeout=15)
    msg_queue = (out_queue or err_queue or "").strip()
    if ok_queue:
        log_test(f"[Postfix] queue listing OK after {stage}", "SUCCESS")
    else:
        log_test(f"[Postfix] queue listing FAILED after {stage}: {msg_queue}", "FAIL")
        overall_ok = False
    messages.append(f"queue: {'OK' if ok_queue else 'FAIL'}")

    # 3) postfix check (config/files/spool sanity)
    ok_check, out_check, err_check = run_sh(["postfix", "check"], timeout=20)
    msg_check = (out_check or err_check or "").strip()
    if ok_check:
        log_test(f"[Postfix] check OK after {stage}", "SUCCESS")
    else:
        log_test(f"[Postfix] check FAILED after {stage}: {msg_check}", "FAIL")
        overall_ok = False
    messages.append(f"check: {'OK' if ok_check else 'FAIL'} ({msg_check})")

    summary = "; ".join(messages)
    if overall_ok:
        log_test(f"Postfix health OK after {stage}: {summary}", "SUCCESS")
    else:
        log_test(f"Postfix health FAILED after {stage}: {summary}", "FAIL")

    return overall_ok, summary

def journal_has(tag: str, needle: str, since_epoch: float | None = None, until_epoch: float | None = None) -> bool:
    args = ["journalctl", "-b", "-t", tag, "--no-pager"]
    if since_epoch is not None:
        args += ["--since", f"@{since_epoch:.3f}"]
    if until_epoch is not None:
        args += ["--until", f"@{until_epoch:.3f}"]
    ok, out, _ = run_sh(args, timeout=30)
    return ok and (needle in out)

def log_test(msg: str, level: str = "INFO"):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    color_map = {
        "INFO": Colors.CYAN,
        "SUCCESS": Colors.GREEN,
        "FAIL": Colors.RED,
        "WARN": Colors.YELLOW,
        "DEBUG": Colors.WHITE
    }
    color = color_map.get(level, Colors.WHITE)
    print(f"[{colored(timestamp, Colors.WHITE)}] {colored(level, color)}: {msg}")
    try:
        with open(TEST_LOG, "a") as f:
            f.write(f"[{timestamp}] {level}: {msg}\n")
    except Exception:
        pass

# -------- Cleanup Tracker --------
class CleanupTracker:
    """Tracks all ephemeral files and directories created during testing"""
    def __init__(self):
        self.files: Set[Path] = set()
        self.directories: Set[Path] = set()

    def track_file(self, path: Path):
        """Track a file for cleanup"""
        self.files.add(path)

    def track_directory(self, path: Path):
        """Track a directory for cleanup"""
        self.directories.add(path)

    def cleanup_all(self):
        """Clean up all tracked files and directories"""
        cleaned_files = 0
        cleaned_dirs = 0
        errors = []

        # Clean files first
        for fpath in sorted(self.files):
            try:
                if fpath.exists():
                    fpath.unlink()
                    log_test(f"Removed file :: {fpath}", "INFO")
                    cleaned_files += 1
            except Exception as e:
                errors.append(f"File {fpath}: {e}")

        # Clean directories (deepest first) - ONLY remove if they are test directories
        for dpath in sorted(self.directories, key=lambda p: len(p.parts), reverse=True):
            try:
                if dpath.exists() and dpath.is_dir():
                    # Safety check: only remove directories with "omvwc_" in the name
                    if "omvwc_" in dpath.name or dpath.name.startswith("test_"):
                        # Try to remove if empty
                        try:
                            dpath.rmdir()
                            log_test(f"Removed directory :: {dpath}", "INFO")
                            cleaned_dirs += 1
                        except OSError:
                            # Not empty, force remove
                            log_test(f"Force remove directory :: {dpath}", "INFO")
                            shutil.rmtree(dpath, ignore_errors=True)
                            if not dpath.exists():
                                cleaned_dirs += 1
            except Exception as e:
                errors.append(f"Directory {dpath}: {e}")

        if cleaned_files > 0 or cleaned_dirs > 0:
            log_test(f"Cleaned up {cleaned_files} files and {cleaned_dirs} directories", "INFO")

        if errors:
            log_test(f"Cleanup errors: {len(errors)}", "WARN")
            for err in errors[:5]:  # Show first 5 errors
                log_test(f"  {err}", "DEBUG")

        # Clear tracking sets
        self.files.clear()
        self.directories.clear()

# -------- Helpers --------
def load_config() -> dict:
    """Load the omv-writecache YAML config and return it as a dict."""
    if not CONFIG_PATH.exists():
        raise RuntimeError(f"Config not found: {CONFIG_PATH}")
    try:
        import yaml
        cfg = yaml.safe_load(CONFIG_PATH.read_text()) or {}
        if not isinstance(cfg, dict):
            raise RuntimeError("Config YAML root is not a mapping/object.")
        return cfg
    except Exception as e:
        raise RuntimeError(f"Failed to read YAML config: {e}")

def get_workspace_info(cfg: dict) -> Tuple[str, Path]:
    """Return (workspace_type, workspace_root). Defaults match omv-writecache."""
    ws_type = str(cfg.get("workspace_type", "tmpfs")).strip().lower()
    if ws_type not in ("tmpfs", "path", "zram"):
        ws_type = "tmpfs"
    ws_root = Path(str(cfg.get("workspace_root", "/run/omv-writecache")).strip() or "/run/omv-writecache")
    return ws_type, ws_root

def load_config_paths(cfg: Optional[dict] = None) -> Dict[str, str]:
    """
    Read /etc/omv-writecache/config.yaml and return {path: policy}.
    Expects a 'paths' multi-line string with lines like:
        /var/log = flush
        /var/lib/docker = flush owner=root:docker mode=0710

    The optional owner= and mode= attributes are stripped; only the policy
    keyword (flush/drop/persist/writeback) is stored in the returned dict.
    """
    if cfg is None:
        cfg = load_config()

    paths_block = cfg.get("paths", "") or ""
    result: Dict[str, str] = {}
    for raw in str(paths_block).splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        if "=" not in line:
            continue
        left, right = line.split("=", 1)
        path = left.strip()
        # policy is the first token; remaining tokens are owner=/mode= attrs
        policy = right.strip().split()[0].lower() if right.strip() else ""
        if path and policy:
            result[path] = policy
    if not result:
        raise RuntimeError("No paths parsed from config 'paths' block.")
    return result

def require_root():
    if os.geteuid() != 0:
        print(colored("This test must be run as root (sudo).", Colors.RED))
        sys.exit(1)

def ensure_script():
    if not Path(SCRIPT_PATH).exists():
        print(colored(f"Script not found at {SCRIPT_PATH}. Edit SCRIPT_PATH in this test.", Colors.RED))
        sys.exit(1)
    os.chmod(SCRIPT_PATH, 0o755)

def run_cmd(cmd: str, extra_args: Optional[List[str]] = None) -> Tuple[bool, str]:
    """
    Run omv-writecache subcommand from a safe cwd so unmount/remount
    won't invalidate the child's working directory.
    extra_args is a list of additional flags, e.g. ["--restart", "--daily"].
    """
    # Resolve script path to an absolute path; fall back to /usr/sbin/omv-writecache
    sp = Path(SCRIPT_PATH)
    if not sp.is_absolute():
        try:
            sp = sp.resolve(strict=True)
        except Exception:
            pass
    if not sp.exists():
        sp = Path("/usr/sbin/omv-writecache")

    # Build command (use sudo if not root)
    full_cmd = [str(sp), cmd] + (extra_args or [])
    if os.geteuid() != 0:
        full_cmd = ["sudo"] + full_cmd

    log_test(f"Running: {' '.join(full_cmd)}", "DEBUG")

    # Keep a sane env (preserve PATH); force safe cwd
    env = os.environ.copy()

    try:
        res = subprocess.run(
            full_cmd,
            capture_output=True,
            text=True,
            timeout=120,
            cwd="/",              # <- critical: safe working directory
            env=env
        )
        out = (res.stdout or "") + (res.stderr or "")
        if res.returncode != 0:
            log_test(f"Command failed ({cmd}): {out}", "WARN")
            return False, out
        return True, out
    except subprocess.TimeoutExpired:
        log_test(f"Command timed out: {cmd}", "FAIL")
        return False, "Timeout"
    except Exception as e:
        log_test(f"Command error: {e}", "FAIL")
        return False, str(e)

def check_overlay_mounted(path: str) -> bool:
    try:
        with open("/proc/mounts") as f:
            for line in f:
                parts = line.split()
                if len(parts) >= 3 and parts[1] == path and parts[2] == "overlay":
                    return True
    except Exception:
        pass
    return False

def overlay_lowerdir(path: str) -> Optional[str]:
    """Return the first lowerdir of the overlay mounted at 'path', or None."""
    try:
        with open("/proc/mounts") as f:
            for line in f:
                parts = line.split()
                if len(parts) >= 4 and parts[1] == path and parts[2] == "overlay":
                    for kv in parts[3].split(","):
                        if kv.startswith("lowerdir="):
                            return kv[len("lowerdir="):].split(":")[0]
    except Exception:
        pass
    return None

def findmnt_fstype(target: str) -> Optional[str]:
    """Return the source filesystem type of the mount at 'target', or None."""
    ok, out, _ = run_sh(["findmnt", "-no", "FSTYPE", target])
    out = (out or "").strip()
    return out if ok and out else None

def load_writecache_module():
    """Import the omv-writecache script as a module so its functions can be unit
    tested directly. The script guards its CLI entry behind __name__ == '__main__',
    so importing it has no side effects."""
    import importlib.util
    import importlib.machinery
    sp = SCRIPT_PATH if Path(SCRIPT_PATH).exists() else "/usr/sbin/omv-writecache"
    loader = importlib.machinery.SourceFileLoader("omv_writecache_mod", sp)
    spec = importlib.util.spec_from_loader(loader.name, loader)
    mod = importlib.util.module_from_spec(spec)
    loader.exec_module(mod)
    return mod

def get_workspace_usage(workspace_root: Path) -> Tuple[int, int]:
    """Return (used_bytes, total_bytes) for the filesystem containing workspace_root."""
    try:
        du = shutil.disk_usage(str(workspace_root))
        return int(du.used), int(du.total)
    except Exception:
        return 0, 0


# -------- Test file helper --------
class TestFile:
    def __init__(self, path: Path, content: Optional[str] = None, tracker: Optional[CleanupTracker] = None):
        self.path = path
        self.content = content or f"Test content for {path.name} at {time.time()}"
        self.checksum = hashlib.md5(self.content.encode()).hexdigest()
        self.tracker = tracker

    def create(self):
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.path.write_text(self.content)
        if self.tracker:
            self.tracker.track_file(self.path)
        os.sync()

    def exists(self) -> bool:
        return self.path.exists()

    def verify(self) -> bool:
        if not self.exists():
            return False
        try:
            current = self.path.read_text()
            return hashlib.md5(current.encode()).hexdigest() == self.checksum
        except Exception:
            return False

    def delete(self):
        try:
            if self.exists():
                self.path.unlink()
        except Exception:
            pass

# -------- Test runner --------
class WriteCacheTest:
    def __init__(self):
        self.cfg = load_config()
        self.workspace_type, self.workspace_root = get_workspace_info(self.cfg)
        self.config_paths = load_config_paths(self.cfg)
        self.test_files: Dict[str, List[TestFile]] = {}
        self.cleanup_tracker = CleanupTracker()

    def create_test_files(self, tag: str):
        log_test(f"Creating test files: {tag}", "INFO")
        created = 0
        for path, policy in self.config_paths.items():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                log_test(f"Skipping file creation in {path} (not in ALLOWED_FOR_FILE_CREATION)", "DEBUG")
                continue
            base = Path(path)
            files: List[TestFile] = []
            for i in range(2):
                fname = f"omvwc_{tag}_{i}_{int(time.time())}.txt"
                tf = TestFile(base / fname, tracker=self.cleanup_tracker)
                try:
                    tf.create()
                    files.append(tf)
                    created += 1
                    log_test(f"Created {tf.path} (policy={policy})", "DEBUG")
                except Exception as e:
                    log_test(f"Failed to create {tf.path}: {e}", "WARN")
            # subdir file
            subdir = base / f"omvwc_subdir_{tag}"
            try:
                subdir.mkdir(parents=True, exist_ok=True)
                self.cleanup_tracker.track_directory(subdir)
                subfile = TestFile(subdir / "subfile.txt", tracker=self.cleanup_tracker)
                subfile.create()
                files.append(subfile)
                created += 1
            except Exception as e:
                log_test(f"Failed to create subdir in {base}: {e}", "WARN")
            self.test_files[f"{path}:{tag}"] = files
        log_test(f"Created {created} test files total", "SUCCESS")

    def verify_files(self, tag: str, should_exist: bool) -> bool:
        ok_all = True
        for path, _ in self.config_paths.items():
            key = f"{path}:{tag}"
            files = self.test_files.get(key, [])
            for tf in files:
                exists = tf.exists()
                valid = tf.verify() if exists else False
                if should_exist:
                    if not (exists and valid):
                        ok_all = False
                        log_test(f"Verify FAIL (missing/corrupt): {tf.path}", "FAIL")
                else:
                    if exists:
                        ok_all = False
                        log_test(f"Verify FAIL (should be gone): {tf.path}", "FAIL")
        return ok_all

    def verify_persistence_per_policy(self, tag: str) -> bool:
        """
        After UNMOUNT, verify files according to policy:
          - drop -> files should NOT exist
          - flush/persist/writeback -> files SHOULD exist (assuming flush_on_shutdown=True)
        """
        ok_all = True
        for path, policy in self.config_paths.items():
            key = f"{path}:{tag}"
            files = self.test_files.get(key, [])

            expected_exists = policy in ("flush", "persist", "writeback")
            path_ok = True

            for tf in files:
                exists = tf.exists()
                valid = tf.verify() if exists else False

                if expected_exists:
                    if not (exists and valid):
                        path_ok = False
                        ok_all = False
                        log_test(f"[{policy}] expected PRESENT after unmount: {tf.path}", "FAIL")
                else:
                    # drop: must be gone
                    if exists:
                        path_ok = False
                        ok_all = False
                        log_test(f"[{policy}] expected ABSENT after unmount: {tf.path}", "FAIL")

            if path_ok:
                if expected_exists:
                    log_test(f"[{policy}] persisted OK: {path}", "SUCCESS")
                else:
                    log_test(f"[{policy}] dropped OK: {path}", "SUCCESS")

        return ok_all

    def delete_test_files(self, tag: str):
        log_test(f"Deleting test files: {tag}", "INFO")
        for path, _ in self.config_paths.items():
            key = f"{path}:{tag}"
            files = self.test_files.get(key, [])
            for tf in files:
                tf.delete()

    def test_concurrent_operations(self) -> bool:
        """Test that lock mechanism prevents concurrent operations"""
        log_test("\n>>> Test: concurrent operations (lock)", "INFO")

        def try_flush():
            ok, _ = run_cmd("flush")
            return ok

        results = []
        threads = []

        # Start two flush operations simultaneously
        for i in range(2):
            t = threading.Thread(target=lambda: results.append(try_flush()))
            threads.append(t)
            t.start()

        for t in threads:
            t.join(timeout=60)

        # Exactly one should succeed (one gets lock, other is blocked)
        if len(results) == 2:
            success = results[0] != results[1]  # XOR
            log_test(f"Concurrent lock test: {'OK' if success else 'FAIL'} (results={results})",
                     "SUCCESS" if success else "FAIL")
            return success
        else:
            log_test("Concurrent test incomplete", "FAIL")
            return False

    def test_disk_space_exhaustion(self) -> bool:
        """Test behavior when workspace runs out of space"""
        log_test("\n>>> Test: disk space exhaustion", "INFO")

        if self.workspace_type == "path":
            log_test("Skip disk exhaustion test on workspace type path", "WARN")
            return True  # Skip test

        # Find a writable path
        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for disk exhaustion test", "WARN")
            return True  # Skip test

        used, total = get_workspace_usage(self.workspace_root)
        if total == 0:
            log_test("Could not determine workspace filesystem size", "WARN")
            return True  # Skip test

        available = total - used
        log_test(f"workspace available: {available / 1024 / 1024:.1f}MB", "DEBUG")

        # Try to fill most of available space (leave 5MB buffer)
        fill_size = max(0, available - 5 * 1024 * 1024)
        if fill_size < 1024 * 1024:  # Less than 1MB available
            log_test("Tmpfs too full for exhaustion test", "WARN")
            return True

        fill_file = test_path / f"omvwc_fill_{int(time.time())}.dat"
        self.cleanup_tracker.track_file(fill_file)

        try:
            # Create large file. Reaching ENOSPC IS the condition under test: a
            # fixed-size workspace (tmpfs or zram) must refuse writes gracefully
            # rather than hang or corrupt. ext4-backed zram fills a few MB short
            # of (total - used) due to filesystem metadata overhead, so treat
            # ENOSPC during the fill as expected and continue to the checks below.
            written = 0
            try:
                with open(fill_file, "wb") as f:
                    chunk = b"x" * (1024 * 1024)  # 1MB chunks
                    while written < fill_size:
                        to_write = min(len(chunk), fill_size - written)
                        f.write(chunk[:to_write])
                        written += to_write
            except OSError as e:
                if e.errno != errno.ENOSPC:
                    raise
                log_test(f"Workspace exhausted as expected at {written / 1024 / 1024:.1f}MB (ENOSPC)", "DEBUG")

            log_test(f"Created {written / 1024 / 1024:.1f}MB fill file", "DEBUG")

            # Try to create more files (should handle gracefully)
            small_file = test_path / f"omvwc_small_{int(time.time())}.txt"
            self.cleanup_tracker.track_file(small_file)
            try:
                small_file.write_text("small content")
                created = True
            except Exception as e:
                log_test(f"Expected failure on full disk: {e}", "DEBUG")
                created = False

            # System should still be operational
            ok, _ = run_cmd("status")

            # Cleanup
            try:
                fill_file.unlink()
                if small_file.exists():
                    small_file.unlink()
            except Exception:
                pass

            log_test("Disk exhaustion test: OK (system remained stable)", "SUCCESS")
            return ok

        except Exception as e:
            log_test(f"Disk exhaustion test error: {e}", "WARN")
            try:
                fill_file.unlink()
            except Exception:
                pass
            return False

    def test_special_permissions(self) -> bool:
        """Test preservation of special permissions and ownership"""
        log_test("\n>>> Test: special permissions and ownership", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for permissions test", "WARN")
            return True

        test_file = test_path / f"omvwc_perms_{int(time.time())}.txt"
        self.cleanup_tracker.track_file(test_file)

        try:
            # Create file with special permissions
            test_file.write_text("permission test")

            # Set various permission bits
            os.chmod(test_file, 0o4755)  # setuid
            original_stat = os.stat(test_file)

            log_test(f"Created file with mode {oct(original_stat.st_mode)}", "DEBUG")

            # Flush to write through
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during permissions test", "FAIL")
                test_file.unlink()
                return False

            # Verify permissions preserved
            new_stat = os.stat(test_file)
            perms_ok = (original_stat.st_mode == new_stat.st_mode)

            test_file.unlink()

            log_test(f"Permissions preservation: {'OK' if perms_ok else 'FAIL'}",
                     "SUCCESS" if perms_ok else "FAIL")
            return perms_ok

        except Exception as e:
            log_test(f"Permissions test error: {e}", "WARN")
            try:
                test_file.unlink()
            except Exception:
                pass
            return False

    def test_symlinks(self) -> bool:
        """Test symlink handling"""
        log_test("\n>>> Test: symlink handling", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for symlink test", "WARN")
            return True

        target_file = test_path / f"omvwc_target_{int(time.time())}.txt"
        symlink_file = test_path / f"omvwc_link_{int(time.time())}.txt"
        self.cleanup_tracker.track_file(target_file)
        self.cleanup_tracker.track_file(symlink_file)

        try:
            # Create target and symlink
            target_file.write_text("symlink target")
            symlink_file.symlink_to(target_file.name)  # Relative symlink

            # Verify symlink works
            content = symlink_file.read_text()
            symlink_ok = (content == "symlink target" and symlink_file.is_symlink())

            # Flush
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during symlink test", "FAIL")
                target_file.unlink()
                symlink_file.unlink()
                return False

            # Verify symlink still works after flush
            still_ok = symlink_file.is_symlink() and symlink_file.read_text() == "symlink target"

            target_file.unlink()
            symlink_file.unlink()

            result = symlink_ok and still_ok
            log_test(f"Symlink test: {'OK' if result else 'FAIL'}",
                     "SUCCESS" if result else "FAIL")
            return result

        except Exception as e:
            log_test(f"Symlink test error: {e}", "WARN")
            try:
                target_file.unlink()
                symlink_file.unlink()
            except Exception:
                pass
            return False

    def test_deep_directories(self) -> bool:
        """Test deep directory hierarchies"""
        log_test("\n>>> Test: deep directory hierarchy", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for deep directory test", "WARN")
            return True

        # Create deep hierarchy (30 levels)
        base_dir = test_path / f"omvwc_deep_{int(time.time())}"
        self.cleanup_tracker.track_directory(base_dir)
        current = base_dir
        depth = 30

        try:
            for i in range(depth):
                current = current / f"level_{i}"
                self.cleanup_tracker.track_directory(current)
            current.mkdir(parents=True, exist_ok=True)

            # Create file at deepest level
            deep_file = current / "deep_file.txt"
            self.cleanup_tracker.track_file(deep_file)
            deep_file.write_text("deep content")

            # Flush
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during deep directory test", "FAIL")
                shutil.rmtree(base_dir, ignore_errors=True)
                return False

            # Verify file still exists
            still_exists = deep_file.exists() and deep_file.read_text() == "deep content"

            shutil.rmtree(base_dir, ignore_errors=True)

            log_test(f"Deep directory test: {'OK' if still_exists else 'FAIL'}",
                     "SUCCESS" if still_exists else "FAIL")
            return still_exists

        except Exception as e:
            log_test(f"Deep directory test error: {e}", "WARN")
            try:
                shutil.rmtree(base_dir, ignore_errors=True)
            except Exception:
                pass
            return False

    def test_whiteout_scenarios(self) -> bool:
        """Test complex whiteout scenarios"""
        log_test("\n>>> Test: whiteout scenarios", "INFO")

        test_path = None
        for path, policy in self.config_paths.items():
            if policy in ("flush", "persist", "writeback"):
                if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                    continue
                test_path = Path(path)
                break

        if not test_path:
            log_test("No suitable path for whiteout test", "WARN")
            return True

        test_file = test_path / f"omvwc_whiteout_{int(time.time())}.txt"
        self.cleanup_tracker.track_file(test_file)

        try:
            # Create, delete, recreate cycle
            test_file.write_text("first version")
            time.sleep(0.1)
            test_file.unlink()
            time.sleep(0.1)
            test_file.write_text("second version")

            # Flush should handle this correctly
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during whiteout test", "FAIL")
                return False

            # File should exist with second version
            exists_ok = test_file.exists()
            content_ok = test_file.read_text() == "second version" if exists_ok else False

            test_file.unlink()

            result = exists_ok and content_ok
            log_test(f"Whiteout test: {'OK' if result else 'FAIL'}",
                     "SUCCESS" if result else "FAIL")
            return result

        except Exception as e:
            log_test(f"Whiteout test error: {e}", "WARN")
            try:
                test_file.unlink()
            except Exception:
                pass
            return False

    def test_status_command(self) -> bool:
        """Test status command output"""
        log_test("\n>>> Test: status command", "INFO")

        ok, output = run_cmd("status")
        if not ok:
            log_test("Status command failed", "FAIL")
            return False

        # Check for expected mounts
        has_workspace = "WriteCache: ACTIVE" in output

        # Count overlay mounts
        overlay_count = output.count("overlay")
        expected_count = len(self.config_paths)

        count_ok = overlay_count >= expected_count  # May have more due to subdirs

        log_test(f"Status: active={has_workspace}, overlays={overlay_count}>={expected_count}", "DEBUG")

        result = has_workspace and count_ok
        log_test(f"Status command test: {'OK' if result else 'FAIL'}",
                 "SUCCESS" if result else "FAIL")
        return result

    def test_large_files(self) -> bool:
        """Test handling of larger files"""
        log_test("\n>>> Test: large file handling", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for large file test", "WARN")
            return True

        # Check available space
        used, total = get_workspace_usage(self.workspace_root)
        if total == 0:
            log_test("Could not determine workspace filesystem size", "WARN")
            return True

        available = total - used
        # Use 10MB or 10% of available, whichever is smaller
        size = min(10 * 1024 * 1024, available // 10)

        if size < 1024 * 1024:  # Less than 1MB
            log_test("Not enough space for large file test", "WARN")
            return True

        large_file = test_path / f"omvwc_large_{int(time.time())}.dat"
        self.cleanup_tracker.track_file(large_file)

        try:
            # Create file with known pattern
            chunk = b"A" * (1024 * 1024)  # 1MB of 'A'
            checksum = hashlib.md5()

            with open(large_file, "wb") as f:
                written = 0
                while written < size:
                    to_write = min(len(chunk), size - written)
                    data = chunk[:to_write]
                    f.write(data)
                    checksum.update(data)
                    written += to_write

            original_checksum = checksum.hexdigest()
            log_test(f"Created {size / 1024 / 1024:.1f}MB file", "DEBUG")

            # Flush
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during large file test", "FAIL")
                large_file.unlink()
                return False

            # Verify integrity
            verify_checksum = hashlib.md5()
            with open(large_file, "rb") as f:
                while True:
                    data = f.read(1024 * 1024)
                    if not data:
                        break
                    verify_checksum.update(data)

            integrity_ok = (verify_checksum.hexdigest() == original_checksum)

            large_file.unlink()

            log_test(f"Large file test: {'OK' if integrity_ok else 'FAIL'}",
                     "SUCCESS" if integrity_ok else "FAIL")
            return integrity_ok

        except Exception as e:
            log_test(f"Large file test error: {e}", "WARN")
            try:
                large_file.unlink()
            except Exception:
                pass
            return False

    def test_rapid_operations(self) -> bool:
        """Test rapid create/modify/delete operations"""
        log_test("\n>>> Test: rapid I/O operations", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for rapid I/O test", "WARN")
            return True

        base_name = f"omvwc_rapid_{int(time.time())}"
        files_created = []

        try:
            # Rapid create
            for i in range(50):
                f = test_path / f"{base_name}_{i}.txt"
                f.write_text(f"content {i}")
                files_created.append(f)
                self.cleanup_tracker.track_file(f)

            # Rapid modify
            for f in files_created[:25]:
                f.write_text(f"modified {f.name}")

            # Rapid delete
            for f in files_created[25:]:
                f.unlink()
                files_created.remove(f)

            # Flush
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during rapid I/O test", "FAIL")
                for f in files_created:
                    try:
                        f.unlink()
                    except Exception:
                        pass
                return False

            # Verify remaining files
            all_ok = True
            for f in files_created:
                if not f.exists() or not f.read_text().startswith("modified"):
                    all_ok = False
                    break

            # Cleanup
            for f in files_created:
                try:
                    f.unlink()
                except Exception:
                    pass

            log_test(f"Rapid I/O test: {'OK' if all_ok else 'FAIL'}",
                     "SUCCESS" if all_ok else "FAIL")
            return all_ok

        except Exception as e:
            log_test(f"Rapid I/O test error: {e}", "WARN")
            for f in files_created:
                try:
                    f.unlink()
                except Exception:
                    pass
            return False

    def test_double_mount(self) -> bool:
        """Test mounting when already mounted"""
        log_test("\n>>> Test: double mount prevention", "INFO")

        # Ensure mounted
        ok1, _ = run_cmd("mount")
        if not ok1:
            log_test("Initial mount failed", "FAIL")
            return False

        # Try to mount again (should be idempotent)
        ok2, output2 = run_cmd("mount")

        # Should succeed or handle gracefully
        # Check that overlays are still mounted
        still_mounted = True
        for path in self.config_paths.keys():
            if not check_overlay_mounted(path):
                still_mounted = False
                break

        result = ok2 and still_mounted
        log_test(f"Double mount test: {'OK' if result else 'FAIL'}",
                 "SUCCESS" if result else "FAIL")
        return result

    def test_empty_directories(self) -> bool:
        """Test handling of empty directories"""
        log_test("\n>>> Test: empty directories", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for empty directory test", "WARN")
            return True

        empty_dir = test_path / f"omvwc_empty_{int(time.time())}"
        self.cleanup_tracker.track_directory(empty_dir)

        try:
            empty_dir.mkdir(parents=True, exist_ok=True)

            # Flush
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during empty directory test", "FAIL")
                shutil.rmtree(empty_dir, ignore_errors=True)
                return False

            # Directory should still exist
            still_exists = empty_dir.exists() and empty_dir.is_dir()

            shutil.rmtree(empty_dir, ignore_errors=True)

            log_test(f"Empty directory test: {'OK' if still_exists else 'FAIL'}",
                     "SUCCESS" if still_exists else "FAIL")
            return still_exists

        except Exception as e:
            log_test(f"Empty directory test error: {e}", "WARN")
            try:
                shutil.rmtree(empty_dir, ignore_errors=True)
            except Exception:
                pass
            return False

    def test_long_filenames(self) -> bool:
        """Test handling of long filenames (near 255 char limit)"""
        log_test("\n>>> Test: long filenames", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for long filename test", "WARN")
            return True

        # Create 250 char filename (leave room for extension)
        long_name = "omvwc_" + "x" * 240 + ".txt"
        long_file = test_path / long_name
        self.cleanup_tracker.track_file(long_file)

        try:
            long_file.write_text("long filename content")

            # Flush
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during long filename test", "FAIL")
                long_file.unlink()
                return False

            # Verify
            still_exists = long_file.exists() and long_file.read_text() == "long filename content"

            long_file.unlink()

            log_test(f"Long filename test: {'OK' if still_exists else 'FAIL'}",
                     "SUCCESS" if still_exists else "FAIL")
            return still_exists

        except Exception as e:
            log_test(f"Long filename test error: {e}", "WARN")
            try:
                long_file.unlink()
            except Exception:
                pass
            return False

    def test_unicode_filenames(self) -> bool:
        """Test handling of unicode/special characters in filenames"""
        log_test("\n>>> Test: unicode filenames", "INFO")

        test_path = None
        for path in self.config_paths.keys():
            if ALLOWED_FOR_FILE_CREATION and path not in ALLOWED_FOR_FILE_CREATION:
                continue
            test_path = Path(path)
            break

        if not test_path:
            log_test("No suitable path for unicode filename test", "WARN")
            return True

        # Test various unicode characters
        test_names = [
            f"omvwc_emoji_😀_{int(time.time())}.txt",
            f"omvwc_chinese_测试_{int(time.time())}.txt",
            f"omvwc_accents_café_{int(time.time())}.txt",
        ]

        files = []
        try:
            for name in test_names:
                f = test_path / name
                f.write_text(f"content for {name}")
                files.append(f)
                self.cleanup_tracker.track_file(f)

            # Flush
            ok, _ = run_cmd("flush")
            if not ok:
                log_test("Flush failed during unicode filename test", "FAIL")
                for f in files:
                    try:
                        f.unlink()
                    except Exception:
                        pass
                return False

            # Verify all files
            all_ok = all(f.exists() for f in files)

            # Cleanup
            for f in files:
                try:
                    f.unlink()
                except Exception:
                    pass

            log_test(f"Unicode filename test: {'OK' if all_ok else 'FAIL'}",
                     "SUCCESS" if all_ok else "FAIL")
            return all_ok

        except Exception as e:
            log_test(f"Unicode filename test error: {e}", "WARN")
            for f in files:
                try:
                    f.unlink()
                except Exception:
                    pass
            return False

    def test_parse_services_daily_actions(self) -> bool:
        """Verify that the new *daily action keywords are parsed without warnings."""
        log_test("\n>>> Test: parse_services accepts daily action keywords", "INFO")
        import yaml as _yaml

        cfg_path = Path("/etc/omv-writecache/config.yaml")
        original_text = cfg_path.read_text() if cfg_path.exists() else None
        cfg = load_config()

        test_services = (
            "cron = restartdaily\n"
            "rsyslog = reloaddaily\n"
            "docker = stopdailystart\n"
            "postfix = restart\n"
            "nginx = reload\n"
        )
        cfg["services"] = test_services

        sp = Path(SCRIPT_PATH) if Path(SCRIPT_PATH).exists() else Path("/usr/sbin/omv-writecache")
        try:
            cfg_path.parent.mkdir(parents=True, exist_ok=True)
            cfg_path.write_text(_yaml.dump(cfg))
            res = subprocess.run(
                [str(sp), "status"],
                capture_output=True, text=True, timeout=10, cwd="/"
            )
            out = (res.stdout or "") + (res.stderr or "")
        except Exception as e:
            log_test(f"Run error: {e}", "FAIL")
            return False
        finally:
            if original_text is not None:
                cfg_path.write_text(original_text)

        # No PARSE_SERVICE_WARN should appear for the known actions
        bad = [line for line in out.splitlines() if "PARSE_SERVICE_WARN" in line]
        if bad:
            for b in bad:
                log_test(f"Unexpected parse warning: {b}", "FAIL")
            return False

        log_test("All daily action keywords parsed without warnings", "SUCCESS")
        return True

    def test_daily_flag_service_filtering(self) -> bool:
        """
        Verify that *daily service actions only fire when --daily is passed.
        Injects fake service names into the config, runs flush with and without
        --daily, and checks stdout for SERVICE_ log entries.
        The service names are deliberately non-existent, so they are never
        'active' and the script logs a SERVICE_<action>_SKIP line instead of
        calling systemctl. The --daily gate is applied before the active check,
        so a *daily entry produces no line at all without --daily.
        """
        log_test("\n>>> Test: --daily flag controls *daily service actions", "INFO")
        import yaml as _yaml

        cfg_path = Path("/etc/omv-writecache/config.yaml")
        original_text = cfg_path.read_text() if cfg_path.exists() else None
        cfg = load_config()

        always_svc = "omv-writecache-test-always"
        daily_svc  = "omv-writecache-test-daily"
        stop_svc   = "omv-writecache-test-stopdaily"

        cfg["services"] = (
            f"{always_svc} = restart\n"
            f"{daily_svc} = restartdaily\n"
            f"{stop_svc} = stopdailystart\n"
        )

        sp = Path(SCRIPT_PATH) if Path(SCRIPT_PATH).exists() else Path("/usr/sbin/omv-writecache")

        def run_flush(daily: bool) -> str:
            args = ["--restart", "--daily"] if daily else ["--restart"]
            try:
                res = subprocess.run(
                    [str(sp), "flush"] + args,
                    capture_output=True, text=True, timeout=60, cwd="/"
                )
                return (res.stdout or "") + (res.stderr or "")
            except Exception as e:
                return str(e)

        ok = True
        try:
            cfg_path.parent.mkdir(parents=True, exist_ok=True)
            cfg_path.write_text(_yaml.dump(cfg))

            # --- Run without --daily ---
            out_no_daily = run_flush(daily=False)
            log_test("Ran flush --restart (no --daily)", "DEBUG")

            always_seen = f"SERVICE_RESTART_SKIP: {always_svc}" in out_no_daily
            daily_seen  = daily_svc in out_no_daily
            stop_seen   = stop_svc  in out_no_daily

            log_test(f"  '{always_svc}' considered (expected yes): {'OK' if always_seen else 'FAIL'}",
                     "SUCCESS" if always_seen else "FAIL")
            log_test(f"  '{daily_svc}' untouched (expected yes):   {'OK' if not daily_seen else 'FAIL'}",
                     "SUCCESS" if not daily_seen else "FAIL")
            log_test(f"  '{stop_svc}' untouched (expected yes):    {'OK' if not stop_seen else 'FAIL'}",
                     "SUCCESS" if not stop_seen else "FAIL")

            ok &= always_seen and not daily_seen and not stop_seen

            # --- Run with --daily ---
            out_daily = run_flush(daily=True)
            log_test("Ran flush --restart --daily", "DEBUG")

            always_seen2 = f"SERVICE_RESTART_SKIP: {always_svc}" in out_daily
            daily_seen2  = f"SERVICE_RESTART_SKIP: {daily_svc}"  in out_daily
            stop_seen2   = f"SERVICE_STOP_SKIP: {stop_svc}"      in out_daily
            # non-existent units must never be started/stopped for real
            stop_started2 = f"SERVICE_START: {stop_svc}"         in out_daily

            log_test(f"  '{always_svc}' considered (expected yes): {'OK' if always_seen2 else 'FAIL'}",
                     "SUCCESS" if always_seen2 else "FAIL")
            log_test(f"  '{daily_svc}' considered (expected yes):  {'OK' if daily_seen2 else 'FAIL'}",
                     "SUCCESS" if daily_seen2 else "FAIL")
            log_test(f"  '{stop_svc}' stop considered (expected yes): {'OK' if stop_seen2 else 'FAIL'}",
                     "SUCCESS" if stop_seen2 else "FAIL")
            log_test(f"  '{stop_svc}' never started (expected yes): {'OK' if not stop_started2 else 'FAIL'}",
                     "SUCCESS" if not stop_started2 else "FAIL")

            ok &= always_seen2 and daily_seen2 and stop_seen2 and not stop_started2

        finally:
            if original_text is not None:
                cfg_path.write_text(original_text)
            elif cfg_path.exists():
                cfg_path.unlink()

        log_test("--daily service filtering: " + ("OK" if ok else "FAIL"),
                 "SUCCESS" if ok else "FAIL")
        return ok

    def test_services_only_touch_running_units(self) -> bool:
        """stop_services()/restart_services() must only act on units that are
        actually running, and restart_services() must start back only the ones
        this run stopped -- so a flush never resurrects rrdcached (or anything)
        on a box where OMV monitoring is deliberately disabled (issue #7)."""
        log_test("\n>>> Test: services only act on running units", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        calls: List[Tuple[str, str]] = []
        active = {"running-svc"}  # only this one is "active"

        orig_active = mod._unit_is_active
        orig_systemctl = mod._run_systemctl
        orig_daily = mod.DAILY_FLUSH
        try:
            mod._unit_is_active = lambda name: name in active
            def fake_systemctl(action, name):
                calls.append((action, name))
                return True
            mod._run_systemctl = fake_systemctl
            mod.DAILY_FLUSH = True

            cfg = {"services": (
                "running-svc = stopstart\n"
                "stopped-svc = stopstart\n"
                "disabled-monitoring = restart\n"
            )}

            stopped = mod.stop_services(cfg)
            mod.restart_services(cfg, stopped)
        finally:
            mod._unit_is_active = orig_active
            mod._run_systemctl = orig_systemctl
            mod.DAILY_FLUSH = orig_daily

        ok = True

        stopped_running = ("stop", "running-svc") in calls
        started_running = ("start", "running-svc") in calls
        log_test(f"  running stopstart unit stopped+started: "
                 f"{'OK' if stopped_running and started_running else 'FAIL'}",
                 "SUCCESS" if stopped_running and started_running else "FAIL")
        ok &= stopped_running and started_running

        touched_stopped = any(n == "stopped-svc" for _, n in calls)
        log_test(f"  inactive stopstart unit never touched: "
                 f"{'OK' if not touched_stopped else 'FAIL'}",
                 "SUCCESS" if not touched_stopped else "FAIL")
        ok &= not touched_stopped

        touched_disabled = any(n == "disabled-monitoring" for _, n in calls)
        log_test(f"  inactive restart unit never started: "
                 f"{'OK' if not touched_disabled else 'FAIL'}",
                 "SUCCESS" if not touched_disabled else "FAIL")
        ok &= not touched_disabled

        returned_only_stopped = (stopped == {"running-svc"})
        log_test(f"  stop_services() returned exactly what it stopped: "
                 f"{'OK' if returned_only_stopped else 'FAIL'} ({sorted(stopped)})",
                 "SUCCESS" if returned_only_stopped else "FAIL")
        ok &= returned_only_stopped

        log_test("services only act on running units: " + ("OK" if ok else "FAIL"),
                 "SUCCESS" if ok else "FAIL")
        return ok

    def test_reconfigure_unmount_no_kill(self) -> bool:
        """unmount(cfg, allow_kill=False) -- the --reconfigure path taken on a
        settings change -- must never 'fuser -km' a busy overlay, while a plain
        unmount (shutdown/reboot) still escalates. A busy overlay must still
        lazy-detach so the reconfigure can proceed (issue #8)."""
        log_test("\n>>> Test: --reconfigure unmount skips the kill escalation", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        target = "/srv/wc-reconfigure-test"
        cfg = {
            "workspace_type": "tmpfs",
            "workspace_root": "/run/omv-writecache-test",
            "paths": f"{target} = flush\n",
        }

        def collect(allow_kill: bool) -> List[List[str]]:
            cmds: List[List[str]] = []

            class FakeResult:
                returncode = 1  # every umount "fails" -> full ladder is walked
                stdout = ""

            orig = {k: getattr(mod, k) for k in
                    ("run", "is_mountpoint", "is_overlay_mounted", "get_mounts")}
            try:
                mod.run = lambda c, check=True, quiet=False: (cmds.append(list(c)) or FakeResult())
                mod.is_mountpoint = lambda p: False
                mod.is_overlay_mounted = lambda p: str(p) == target
                mod.get_mounts = lambda: []
                mod.unmount(cfg, allow_kill=allow_kill)
            finally:
                for k, v in orig.items():
                    setattr(mod, k, v)
            return cmds

        recfg = collect(allow_kill=False)
        shutd = collect(allow_kill=True)

        def has(cmds, pred):
            return any(pred(c) for c in cmds)

        ok = True

        recfg_no_fuser = not has(recfg, lambda c: c[:1] == ["fuser"])
        log_test(f"  --reconfigure never runs fuser: {'OK' if recfg_no_fuser else 'FAIL'}",
                 "SUCCESS" if recfg_no_fuser else "FAIL")
        ok &= recfg_no_fuser

        recfg_lazy = has(recfg, lambda c: c[:1] == ["umount"] and "-l" in c)
        log_test(f"  --reconfigure still lazy-detaches a busy overlay: {'OK' if recfg_lazy else 'FAIL'}",
                 "SUCCESS" if recfg_lazy else "FAIL")
        ok &= recfg_lazy

        shutd_fuser = has(shutd, lambda c: c[:1] == ["fuser"])
        log_test(f"  plain unmount still escalates to fuser: {'OK' if shutd_fuser else 'FAIL'}",
                 "SUCCESS" if shutd_fuser else "FAIL")
        ok &= shutd_fuser

        log_test("--reconfigure unmount kill guard: " + ("OK" if ok else "FAIL"),
                 "SUCCESS" if ok else "FAIL")
        return ok

    def test_reconfigure_flag_parsing(self) -> bool:
        """main() must parse 'remount --reconfigure' into RECONFIGURE=True and
        drive cmd_unmount(); cmd_mount(), so the salt reconfigure state actually
        reaches unmount(allow_kill=False) (issue #8)."""
        log_test("\n>>> Test: --reconfigure flag is recognised by main()", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        import os as _os
        seen = {"reconfigure": None, "order": []}

        saved = {k: getattr(mod, k) for k in
                 ("cmd_unmount", "cmd_mount", "check_and_create_lock",
                  "release_lock", "ensure_persistent_journal", "reclaim_pending_zram",
                  "_unit_is_active", "_run_systemctl")}
        saved_argv = sys.argv[:]
        saved_geteuid = _os.geteuid
        try:
            def fake_unmount():
                seen["reconfigure"] = mod.RECONFIGURE
                seen["order"].append("unmount")
            mod.cmd_unmount = fake_unmount
            mod.cmd_mount = lambda: seen["order"].append("mount")
            mod.check_and_create_lock = lambda name: None
            mod.release_lock = lambda: None
            mod.ensure_persistent_journal = lambda: None
            mod.reclaim_pending_zram = lambda: None
            # remount now also wraps the Services list around unmount/mount
            # (like cmd_flush already did); never let this test touch the
            # real host's units.
            mod._unit_is_active = lambda name: False
            mod._run_systemctl = lambda action, name: True
            mod.RECONFIGURE = False
            _os.geteuid = lambda: 0
            sys.argv = ["omv-writecache", "remount", "--reconfigure"]
            mod.main()
        except SystemExit:
            pass
        finally:
            for k, v in saved.items():
                setattr(mod, k, v)
            sys.argv = saved_argv
            _os.geteuid = saved_geteuid
            mod.RECONFIGURE = False

        ok = True

        parsed = (seen["reconfigure"] is True)
        log_test(f"  RECONFIGURE set True during cmd_unmount: {'OK' if parsed else 'FAIL'}",
                 "SUCCESS" if parsed else "FAIL")
        ok &= parsed

        ordered = (seen["order"] == ["unmount", "mount"])
        log_test(f"  remount ran unmount then mount: {'OK' if ordered else 'FAIL'} ({seen['order']})",
                 "SUCCESS" if ordered else "FAIL")
        ok &= ordered

        log_test("--reconfigure flag parsing: " + ("OK" if ok else "FAIL"),
                 "SUCCESS" if ok else "FAIL")
        return ok

    def test_private_tmp_detection(self) -> bool:
        """A remount must detect running units with PrivateTmp=yes and warn
        about them instead of restarting them -- their host-backed private
        tmp directory under /tmp or /var/tmp is wiped and recreated by the
        remount, so their NEXT reload/restart (not this one) would otherwise
        fail with 226/NAMESPACE with no admin-visible explanation (issue #10).
        """
        log_test("\n>>> Test: PrivateTmp units are detected and warned about, "
                 "not restarted, on remount", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        import os as _os

        running_units = ["has-private-tmp.service", "no-private-tmp.service",
                          "unrelated.service"]
        private_tmp_units = {"has-private-tmp.service"}

        class FakeResult:
            def __init__(self, stdout):
                self.stdout = stdout
                self.returncode = 0

        def fake_run(cmd, **kwargs):
            if cmd[:2] == ["systemctl", "list-units"]:
                return FakeResult("\n".join(f"{u} loaded active running" for u in running_units) + "\n")
            if cmd[:2] == ["systemctl", "show"]:
                unit = cmd[2]
                return FakeResult("yes\n" if unit in private_tmp_units else "no\n")
            return FakeResult("")

        saved = {k: getattr(mod, k) for k in
                 ("cmd_unmount", "cmd_mount", "check_and_create_lock",
                  "release_lock", "ensure_persistent_journal", "reclaim_pending_zram", "subprocess",
                  "load_config", "_run_systemctl", "_unit_is_active", "log")}
        saved_argv = sys.argv[:]
        saved_geteuid = _os.geteuid

        restarted: List[Tuple[str, str]] = []
        warnings: List[str] = []
        detected_with_var_tmp = None
        scan_calls = {"count": 0}

        class FakeSubprocess:
            @staticmethod
            def run(cmd, **kwargs):
                scan_calls["count"] += 1
                return fake_run(cmd, **kwargs)
            PIPE = mod.subprocess.PIPE
            STDOUT = mod.subprocess.STDOUT
            DEVNULL = mod.subprocess.DEVNULL

        def run_main():
            # main() always ends in sys.exit() with the command's status.
            try:
                mod.main()
            except SystemExit:
                pass

        try:
            mod.cmd_unmount = lambda: None
            mod.cmd_mount = lambda: None
            mod.check_and_create_lock = lambda name: None
            mod.release_lock = lambda: None
            mod.ensure_persistent_journal = lambda: None
            mod.reclaim_pending_zram = lambda: None
            mod.subprocess = FakeSubprocess
            mod._unit_is_active = lambda name: True

            def fake_systemctl(action, name):
                restarted.append((action, name))
                return True
            mod._run_systemctl = fake_systemctl

            def fake_log(msg, level="INFO"):
                if level == "WARN":
                    warnings.append(msg)
            mod.log = fake_log

            # Unit-test the scan directly while the fake systemctl is in place.
            detected_with_var_tmp = mod.find_active_private_tmp_units()

            # A path under /var/tmp is configured, no matching Services entry:
            # main() must run the scan and only warn, never restart, about the
            # unit it finds.
            mod.load_config = lambda: {"paths": "/var/tmp = drop\n"}
            mod.RECONFIGURE = False
            _os.geteuid = lambda: 0
            sys.argv = ["omv-writecache", "remount", "--reconfigure"]
            run_main()

            scans_with_var_tmp = scan_calls["count"]
            warned_when_unhandled = list(warnings)
            restarted_when_unhandled = list(restarted)

            # No /tmp or /var/tmp path configured: the scan must be skipped
            # entirely so a settings change unrelated to PrivateTmp never
            # spends time probing every running unit's PrivateTmp property.
            mod.load_config = lambda: {"paths": "/var/log = drop\n"}
            sys.argv = ["omv-writecache", "remount", "--reconfigure"]
            run_main()

            scans_without_var_tmp = scan_calls["count"] - scans_with_var_tmp

            # Admin has listed the affected unit in Services: main() must
            # restart it (via restart_services, right after cmd_mount) and
            # drop it from the warning -- it is handled now, not just flagged.
            restarted.clear()
            warnings.clear()
            mod.load_config = lambda: {
                "paths": "/var/tmp = drop\n",
                "services": "has-private-tmp.service = restart\n",
            }
            sys.argv = ["omv-writecache", "remount", "--reconfigure"]
            run_main()

            restarted_when_handled = list(restarted)
            warned_when_handled = list(warnings)
        finally:
            for k, v in saved.items():
                setattr(mod, k, v)
            sys.argv = saved_argv
            _os.geteuid = saved_geteuid
            mod.RECONFIGURE = False

        ok = True

        detected_ok = (set(detected_with_var_tmp or []) == private_tmp_units)
        log_test(f"  find_active_private_tmp_units() returns only PrivateTmp=yes units: "
                 f"{'OK' if detected_ok else 'FAIL'} ({detected_with_var_tmp})",
                 "SUCCESS" if detected_ok else "FAIL")
        ok &= detected_ok

        never_restarted = (restarted_when_unhandled == [])
        log_test(f"  remount never auto-restarts an unhandled unit: "
                 f"{'OK' if never_restarted else 'FAIL'} ({restarted_when_unhandled})",
                 "SUCCESS" if never_restarted else "FAIL")
        ok &= never_restarted

        warned_ok = any("has-private-tmp.service" in w for w in warned_when_unhandled)
        log_test(f"  remount warns about an unhandled unit: "
                 f"{'OK' if warned_ok else 'FAIL'} ({warned_when_unhandled})",
                 "SUCCESS" if warned_ok else "FAIL")
        ok &= warned_ok

        scanned_when_relevant = scans_with_var_tmp > 0
        log_test(f"  remount scans for PrivateTmp units when /var/tmp is managed: "
                 f"{'OK' if scanned_when_relevant else 'FAIL'}",
                 "SUCCESS" if scanned_when_relevant else "FAIL")
        ok &= scanned_when_relevant

        skipped_when_irrelevant = (scans_without_var_tmp == 0)
        log_test(f"  remount skips the scan when no /tmp or /var/tmp path is managed: "
                 f"{'OK' if skipped_when_irrelevant else 'FAIL'}",
                 "SUCCESS" if skipped_when_irrelevant else "FAIL")
        ok &= skipped_when_irrelevant

        restarted_ok = ("restart", "has-private-tmp.service") in restarted_when_handled
        log_test(f"  remount restarts a unit the admin listed in Services: "
                 f"{'OK' if restarted_ok else 'FAIL'} ({restarted_when_handled})",
                 "SUCCESS" if restarted_ok else "FAIL")
        ok &= restarted_ok

        not_warned_ok = not any("has-private-tmp.service" in w for w in warned_when_handled)
        log_test(f"  remount does not warn about a unit the admin already handles: "
                 f"{'OK' if not_warned_ok else 'FAIL'} ({warned_when_handled})",
                 "SUCCESS" if not_warned_ok else "FAIL")
        ok &= not_warned_ok

        log_test("PrivateTmp detection: " + ("OK" if ok else "FAIL"),
                 "SUCCESS" if ok else "FAIL")
        return ok

    def test_dropin_file(self) -> bool:
        """Verify the workspace mount drop-in file exists with correct content (RAM-backed: tmpfs/zram)."""
        log_test("\n>>> Test: workspace mount drop-in file", "INFO")

        if self.workspace_type == "path":
            log_test("Skip: workspace is disk-backed (path)", "WARN")
            return True

        dropin = Path(r"/etc/systemd/system/run-omv\x2dwritecache.mount.d/10-before-setup.conf")
        if not dropin.exists():
            log_test(f"Drop-in not found: {dropin}", "FAIL")
            return False

        content = dropin.read_text()
        has_nodeps = "DefaultDependencies=no" in content
        has_before = "Before=omv-writecache-setup.service" in content

        log_test(f"Drop-in DefaultDependencies=no: {'OK' if has_nodeps else 'FAIL'}", "SUCCESS" if has_nodeps else "FAIL")
        log_test(f"Drop-in Before=omv-writecache-setup.service: {'OK' if has_before else 'FAIL'}", "SUCCESS" if has_before else "FAIL")
        return has_nodeps and has_before

    def test_setup_unit_kill_mode(self) -> bool:
        """A stop timeout must SIGTERM only the script (which defers it until the
        path is consistent), never its rsync child, and never SIGKILL either."""
        log_test("\n>>> Test: setup unit kill mode", "INFO")

        unit = Path("/etc/systemd/system/omv-writecache-setup.service")
        if not unit.exists():
            log_test(f"Unit not found: {unit}", "FAIL")
            return False

        content = unit.read_text()
        has_mixed = "KillMode=mixed" in content
        no_sigkill = "SendSIGKILL=no" in content

        log_test(f"KillMode=mixed: {'OK' if has_mixed else 'FAIL'}", "SUCCESS" if has_mixed else "FAIL")
        log_test(f"SendSIGKILL=no: {'OK' if no_sigkill else 'FAIL'}", "SUCCESS" if no_sigkill else "FAIL")
        return has_mixed and no_sigkill

    def test_mount_unit_ordering(self) -> bool:
        """After mount, verify systemd applied the drop-in to the workspace mount unit."""
        log_test("\n>>> Test: workspace mount unit ordering (drop-in applied)", "INFO")

        if self.workspace_type == "path":
            log_test("Skip: workspace is disk-backed (path)", "WARN")
            return True

        ok_esc, unit_name, _ = run_sh(
            ["systemd-escape", "--path", "--suffix=mount", str(self.workspace_root)],
            timeout=5
        )
        if not ok_esc or not unit_name.strip():
            log_test("Could not determine unit name via systemd-escape", "FAIL")
            return False
        unit_name = unit_name.strip()
        log_test(f"Workspace mount unit: {unit_name}", "DEBUG")

        ok_show, out, _ = run_sh(
            ["systemctl", "show", unit_name, "--no-pager",
             "--property=Before,Conflicts,DefaultDependencies"],
            timeout=10
        )
        if not ok_show:
            log_test(f"systemctl show failed for {unit_name} (unit not active?)", "WARN")
            return True  # can't verify if unit doesn't exist yet

        log_test(f"Unit properties:\n{out.strip()}", "DEBUG")

        before_ok = "omv-writecache-setup.service" in out

        conflicts_line = next((l for l in out.splitlines() if l.startswith("Conflicts=")), "")
        no_umount_conflict = "umount.target" not in conflicts_line

        log_test(f"Before= includes omv-writecache-setup.service: {'OK' if before_ok else 'FAIL'}", "SUCCESS" if before_ok else "FAIL")
        log_test(f"Conflicts= excludes umount.target: {'OK' if no_umount_conflict else 'FAIL'}", "SUCCESS" if no_umount_conflict else "FAIL")
        return before_ok and no_umount_conflict

    def test_workspace_survives_flush(self) -> bool:
        """RAM-backed workspace (tmpfs/zram) must remain mounted throughout a flush cycle."""
        log_test("\n>>> Test: workspace survives flush", "INFO")

        if self.workspace_type == "path":
            log_test("Skip: workspace is disk-backed (path)", "WARN")
            return True

        def is_mounted(path: Path) -> bool:
            try:
                with open("/proc/mounts") as f:
                    return any(line.split()[1] == str(path) for line in f)
            except Exception:
                return False

        before = is_mounted(self.workspace_root)
        if not before:
            log_test(f"Workspace not mounted before flush: {self.workspace_root}", "FAIL")
            return False

        ok, _ = run_cmd("flush")

        after = is_mounted(self.workspace_root)

        log_test(f"Workspace mounted before flush: {'OK' if before else 'FAIL'}", "SUCCESS" if before else "FAIL")
        log_test(f"Workspace mounted after flush:  {'OK' if after else 'FAIL'}", "SUCCESS" if after else "FAIL")
        return ok and after

    def test_flush_preserves_dir_ownership(self) -> bool:
        """
        Flush must not alter the managed directory's own uid/gid/mode.

        When an app chmod/chowns the managed path while the overlay is active,
        overlayfs records the change in the upper directory's own attributes.
        rsync -a then propagates those onto the lower (real) directory at the
        end of the transfer.  The fix saves/restores lower root's stat around
        rsync so only the *contents* are affected.
        """
        log_test("\n>>> Test: flush preserves directory ownership", "INFO")

        flush_paths = [(p, m) for p, m in self.config_paths.items()
                       if m in ("flush", "persist", "writeback")]
        if not flush_paths:
            log_test("Skip: no flush/persist/writeback paths configured", "WARN")
            return True

        ok_all = True
        for path_str, _ in flush_paths:
            p = Path(path_str)
            try:
                # Stat now (overlay root with no copy-up yet == lower's attrs).
                orig = os.stat(str(p))
                orig_mode = stat.S_IMODE(orig.st_mode)
                orig_uid  = orig.st_uid
                orig_gid  = orig.st_gid
                log_test(f"  {path_str}: original uid={orig_uid} gid={orig_gid} mode={oct(orig_mode)}", "INFO")

                # Simulate an app chowning/chmodding the managed directory while
                # the overlay is active.  overlayfs records this by setting the
                # upper directory's own mode/owner, which rsync then copies to lower.
                os.chmod(str(p), 0o777)
                os.chown(str(p), 0, 0)
                log_test(f"  {path_str}: set to root:root 0o777 via overlay to simulate app behaviour", "INFO")

                ok, _ = run_cmd("flush")
                if not ok:
                    log_test(f"  {path_str}: flush command failed", "FAIL")
                    ok_all = False
                    try:
                        os.chmod(str(p), orig_mode)
                        os.chown(str(p), orig_uid, orig_gid)
                    except Exception:
                        pass
                    continue

                # After flush the upper is cleared and the overlay is remounted;
                # the overlay root now shows the real directory's attributes.
                after = os.stat(str(p))
                after_mode = stat.S_IMODE(after.st_mode)
                after_uid  = after.st_uid
                after_gid  = after.st_gid

                if after_mode == orig_mode and after_uid == orig_uid and after_gid == orig_gid:
                    log_test(f"  {path_str}: dir ownership preserved OK "
                             f"(uid={after_uid} gid={after_gid} mode={oct(after_mode)})", "SUCCESS")
                else:
                    log_test(f"  {path_str}: dir ownership changed! "
                             f"expected uid={orig_uid} gid={orig_gid} mode={oct(orig_mode)}, "
                             f"got uid={after_uid} gid={after_gid} mode={oct(after_mode)}", "FAIL")
                    ok_all = False
                    # Restore so subsequent tests are not disrupted.
                    try:
                        os.chmod(str(p), orig_mode)
                        os.chown(str(p), orig_uid, orig_gid)
                    except Exception:
                        pass

            except PermissionError as e:
                log_test(f"  {path_str}: permission error (requires root): {e}", "WARN")
            except Exception as e:
                log_test(f"  {path_str}: unexpected error: {e}", "FAIL")
                ok_all = False

        return ok_all

    def test_enforced_dir_ownership(self) -> bool:
        """
        When a path is configured with owner= and/or mode= attributes, flush
        must apply those specific values to the managed directory root,
        overriding whatever rsync would have written.

        This test only runs if at least one flush/persist/writeback path has
        an owner= or mode= attribute in the config.  It is skipped (pass) when
        no such paths are configured.
        """
        log_test("\n>>> Test: enforced directory ownership (owner=/mode= attributes)", "INFO")

        import subprocess as _sp

        # Re-parse the raw config to pick up owner= / mode= tokens.
        cfg = load_config()
        paths_block = str(cfg.get("paths", "") or "")
        import pwd as _pwd, grp as _grp

        entries = []  # [(path_str, policy, uid_or_None, gid_or_None, mode_or_None)]
        for raw in paths_block.splitlines():
            line = raw.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            left, rest = line.split("=", 1)
            path_str = left.strip()
            parts = rest.strip().split()
            policy = parts[0].lower() if parts else "drop"
            if policy not in ("flush", "persist", "writeback"):
                continue
            uid = gid = perms = None
            for token in parts[1:]:
                if token.startswith("owner="):
                    val = token[6:]
                    user, sep, group = val.partition(":")
                    if user:
                        try:
                            uid = _pwd.getpwnam(user).pw_uid
                        except KeyError:
                            try:
                                uid = int(user)
                            except ValueError:
                                pass
                    if sep and group:
                        try:
                            gid = _grp.getgrnam(group).gr_gid
                        except KeyError:
                            try:
                                gid = int(group)
                            except ValueError:
                                pass
                elif token.startswith("mode="):
                    try:
                        perms = int(token[5:], 8)
                    except ValueError:
                        pass
            if uid is not None or gid is not None or perms is not None:
                entries.append((path_str, policy, uid, gid, perms))

        if not entries:
            log_test("Skip: no flush paths with owner= or mode= attributes configured", "WARN")
            return True

        ok_all = True
        for path_str, _, want_uid, want_gid, want_mode in entries:
            p = Path(path_str)
            try:
                # Deliberately set the managed directory to wrong attrs via the overlay.
                wrong_uid = 0
                wrong_gid = 0
                wrong_mode = 0o777
                os.chmod(str(p), wrong_mode)
                os.chown(str(p), wrong_uid, wrong_gid)
                log_test(f"  {path_str}: set to root:root 0o777 to simulate bad app behaviour", "INFO")

                ok, _ = run_cmd("flush")
                if not ok:
                    log_test(f"  {path_str}: flush command failed", "FAIL")
                    ok_all = False
                    continue

                after = os.stat(str(p))
                after_mode = stat.S_IMODE(after.st_mode)
                after_uid  = after.st_uid
                after_gid  = after.st_gid

                uid_ok   = (want_uid  is None or after_uid  == want_uid)
                gid_ok   = (want_gid  is None or after_gid  == want_gid)
                mode_ok  = (want_mode is None or after_mode == want_mode)

                if uid_ok and gid_ok and mode_ok:
                    log_test(f"  {path_str}: enforced attrs applied OK "
                             f"(uid={after_uid} gid={after_gid} mode={oct(after_mode)})", "SUCCESS")
                else:
                    msgs = []
                    if not uid_ok:
                        msgs.append(f"uid: want {want_uid} got {after_uid}")
                    if not gid_ok:
                        msgs.append(f"gid: want {want_gid} got {after_gid}")
                    if not mode_ok:
                        msgs.append(f"mode: want {oct(want_mode)} got {oct(after_mode)}")
                    log_test(f"  {path_str}: enforced attrs NOT applied — {'; '.join(msgs)}", "FAIL")
                    ok_all = False

            except PermissionError as e:
                log_test(f"  {path_str}: permission error (requires root): {e}", "WARN")
            except Exception as e:
                log_test(f"  {path_str}: unexpected error: {e}", "FAIL")
                ok_all = False

        return ok_all

    def test_live_overlay_dirs_parsing(self) -> bool:
        """Unit test: live_overlay_dirs() must read a live overlay's own
        lower/upper/work from its mount options, return None when no overlay is
        mounted at the path, and ignore (return None for) multi-lower overlays it
        does not manage. This is what lets flush target the actually-mounted
        workspace instead of re-deriving it from a possibly-drifted workspace_root."""
        log_test("\n>>> Test: live_overlay_dirs parses live overlay mount options", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        if not hasattr(mod, "live_overlay_dirs"):
            log_test("omv-writecache has no live_overlay_dirs()", "FAIL")
            return False

        base = "/srv/disk/writecache/var_log"
        fake_single = [{
            "device": "overlay", "path": "/var/log", "fstype": "overlay",
            "opts": (f"rw,relatime,lowerdir={base}/lower,upperdir={base}/upper,"
                     f"workdir={base}/work,uuid=on,nouserxattr"),
        }]
        fake_multi = [{
            "device": "overlay", "path": "/var/log", "fstype": "overlay",
            "opts": (f"rw,lowerdir={base}/lower:/other/lo,upperdir={base}/upper,"
                     f"workdir={base}/work"),
        }]

        orig_get_mounts = mod.get_mounts
        ok_all = True
        try:
            # 1) Single-lower overlay -> exact lower/upper/work.
            mod.get_mounts = lambda: fake_single
            got = mod.live_overlay_dirs(Path("/var/log"))
            exp = {"lower": f"{base}/lower", "upper": f"{base}/upper", "work": f"{base}/work"}
            single_ok = (got == exp)
            log_test(f"  single-lower parse: {'OK' if single_ok else 'FAIL'} (got={got})",
                     "SUCCESS" if single_ok else "FAIL")
            ok_all &= single_ok

            # 2) No overlay mounted at this path -> None.
            none_ok = mod.live_overlay_dirs(Path("/var/tmp")) is None
            log_test(f"  no-overlay -> None: {'OK' if none_ok else 'FAIL'}",
                     "SUCCESS" if none_ok else "FAIL")
            ok_all &= none_ok

            # 3) Multi (colon-separated) lowerdir is not ours -> None (fall back).
            mod.get_mounts = lambda: fake_multi
            multi_ok = mod.live_overlay_dirs(Path("/var/log")) is None
            log_test(f"  colon lowerdir -> None: {'OK' if multi_ok else 'FAIL'}",
                     "SUCCESS" if multi_ok else "FAIL")
            ok_all &= multi_ok
        finally:
            mod.get_mounts = orig_get_mounts

        log_test("live_overlay_dirs parsing: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return ok_all

    def test_flush_lower_bind_stays_safe(self) -> bool:
        """Regression: a flush must never capture its own overlay as the lower bind.

        The previous code bound the lower from the path while the overlay was still
        mounted, producing an 'overlay'-sourced lower bind and the failure
        'LOWER_BIND_UNSAFE: refusing to flush ... source=overlay'. After a flush every
        managed overlay must be remounted and its lowerdir must resolve to a REAL
        filesystem (not 'overlay')."""
        log_test("\n>>> Test: flush keeps lower bind safe (no overlay-sourced lower)", "INFO")

        flush_paths = [p for p, m in self.config_paths.items()
                       if m in ("flush", "persist", "writeback")]
        if not flush_paths:
            log_test("Skip: no flush/persist/writeback paths configured", "WARN")
            return True

        ok, out = run_cmd("flush")
        if not ok:
            log_test("Flush command failed", "FAIL")
            return False
        if "LOWER_BIND_UNSAFE" in out:
            log_test("Flush logged LOWER_BIND_UNSAFE", "FAIL")
            return False

        ok_all = True
        for path_str in flush_paths:
            if not check_overlay_mounted(path_str):
                log_test(f"  {path_str}: overlay NOT remounted after flush", "FAIL")
                ok_all = False
                continue
            lower = overlay_lowerdir(path_str)
            if not lower:
                log_test(f"  {path_str}: could not determine lowerdir", "FAIL")
                ok_all = False
                continue
            fstype = findmnt_fstype(lower)
            if fstype == "overlay":
                log_test(f"  {path_str}: lower bind {lower} is overlay-sourced (UNSAFE)", "FAIL")
                ok_all = False
            else:
                log_test(f"  {path_str}: lower {lower} fstype={fstype or '?'} OK", "SUCCESS")

        log_test("flush lower-bind safety: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return ok_all

    # ---------- flush failure-path unit tests ----------

    def _fake_flush_env(self, mod, tmp: Path, rsync_ok: bool, rw_ok: bool = True):
        """Build a self-contained fake workspace and patch out everything flush()
        touches except the logic under test. Returns (cfg, upper, calls, restore).

        'calls' records whether rsync was actually attempted, so a test can prove
        the read-only-lower guard short-circuits before the transfer.
        """
        ws = tmp / "ws"
        name = "fake_target"
        upper = ws / name / "upper"
        work = ws / name / "work"
        lower = ws / name / "lower"
        for d in (upper, work, lower):
            d.mkdir(parents=True, exist_ok=True)

        # The sentinel is the "cached write" that must survive a failed flush.
        (upper / "pending.log").write_text("cached-data")

        cfg = {
            "paths": "/fake/target = flush\n",
            "workspace_root": str(ws),
            "workspace_type": "path",
        }

        calls = {"rsync": 0}

        class FakeResult:
            returncode = 0
            stdout = ""

        orig = {k: getattr(mod, k) for k in (
            "is_workspace_ready", "live_overlay_dirs", "is_overlay_mounted",
            "is_mountpoint", "lower_bind_is_safe", "remount_lower",
            "rsync_upper_to_lower", "run", "sync_upper_attrs",
        )}

        def fake_rsync(*a, **kw):
            calls["rsync"] += 1
            return rsync_ok

        mod.is_workspace_ready = lambda cfg: True
        mod.live_overlay_dirs = lambda p: None
        mod.is_overlay_mounted = lambda p: False
        mod.is_mountpoint = lambda p: True
        mod.lower_bind_is_safe = lambda p: True
        mod.remount_lower = lambda lower, rw, check=False: (rw_ok if rw else True)
        mod.rsync_upper_to_lower = fake_rsync
        mod.run = lambda cmd, check=True, quiet=False, **kw: FakeResult()
        mod.sync_upper_attrs = lambda *a, **kw: None

        def restore():
            for k, v in orig.items():
                setattr(mod, k, v)

        return cfg, upper, calls, restore

    def test_flush_preserves_upper_on_failure(self) -> bool:
        """Regression: a FAILED flush must keep the upper and must report failure.

        The upper is the only copy of the cached writes. The old code deleted and
        recreated it unconditionally -- outside the 'if ok:' branch -- so a failed
        rsync discarded every pending write and the overlay was remounted with an
        empty upper, exposing the older lower content ('files reverted to an older
        version'). The failure branch also never incremented 'errors', so flush()
        returned True and cron/systemd recorded a success."""
        log_test("\n>>> Test: failed flush preserves upper and reports failure", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        ok_all = True

        # 1) rsync fails -> upper preserved, flush() reports False.
        with tempfile.TemporaryDirectory() as td:
            cfg, upper, calls, restore = self._fake_flush_env(mod, Path(td), rsync_ok=False)
            try:
                result = mod.flush(cfg)
            finally:
                restore()

            survived = (upper / "pending.log").is_file() and \
                       (upper / "pending.log").read_text() == "cached-data"
            log_test(f"  failed flush keeps cached data: {'OK' if survived else 'FAIL'}",
                     "SUCCESS" if survived else "FAIL")
            reported = (result is False)
            log_test(f"  failed flush returns False: {'OK' if reported else 'FAIL'} (got {result})",
                     "SUCCESS" if reported else "FAIL")
            ok_all = ok_all and survived and reported

        # 2) Control: rsync succeeds -> upper IS cleared and flush() reports True.
        #    Proves the guard is conditional, not a blanket "never clear".
        with tempfile.TemporaryDirectory() as td:
            cfg, upper, calls, restore = self._fake_flush_env(mod, Path(td), rsync_ok=True)
            try:
                result = mod.flush(cfg)
            finally:
                restore()

            cleared = not (upper / "pending.log").exists()
            log_test(f"  successful flush clears upper: {'OK' if cleared else 'FAIL'}",
                     "SUCCESS" if cleared else "FAIL")
            reported = (result is True)
            log_test(f"  successful flush returns True: {'OK' if reported else 'FAIL'} (got {result})",
                     "SUCCESS" if reported else "FAIL")
            ok_all = ok_all and cleared and reported

        log_test("flush upper preservation: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return ok_all

    def test_flush_aborts_on_readonly_lower(self) -> bool:
        """Regression: a lower that cannot be remounted read-write must abort the
        transfer for that path, not run rsync into a read-only target.

        remount_lower() defaults to check=False and only logs REMOUNT_FAILED, so
        the old code fell straight through into rsync, which then failed for every
        file -- and (before the fix above) took the upper down with it."""
        log_test("\n>>> Test: flush aborts when lower cannot be made read-write", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        with tempfile.TemporaryDirectory() as td:
            # rsync would succeed if it ran; the RW failure must stop it first.
            cfg, upper, calls, restore = self._fake_flush_env(
                mod, Path(td), rsync_ok=True, rw_ok=False)
            try:
                result = mod.flush(cfg)
            finally:
                restore()

            # Assert inside the context: leaving it deletes the fake workspace.
            skipped = (calls["rsync"] == 0)
            log_test(f"  rsync not attempted on read-only lower: "
                     f"{'OK' if skipped else 'FAIL'} (attempts={calls['rsync']})",
                     "SUCCESS" if skipped else "FAIL")
            survived = (upper / "pending.log").is_file()
            log_test(f"  upper preserved: {'OK' if survived else 'FAIL'}",
                     "SUCCESS" if survived else "FAIL")
            reported = (result is False)
            log_test(f"  flush returns False: {'OK' if reported else 'FAIL'} (got {result})",
                     "SUCCESS" if reported else "FAIL")

        ok_all = skipped and survived and reported
        log_test("flush read-only lower guard: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return ok_all

    def test_flush_busy_syncs_in_place(self) -> bool:
        """Regression: a routine flush of a busy overlay must not detach it.

        The old code lazily unmounted /var/log every hour because nginx, rsyslog,
        samba, php-fpm... keep their logs open. They then wrote into the detached
        upper, which the flush deletes, so everything they logged afterwards was
        lost. A busy path must now be synced in place from the merged view and
        stay mounted with its upper intact; a teardown keeps the old path."""
        log_test("\n>>> Test: busy flush syncs in place instead of detaching", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        def attempt(td: Path, teardown: bool):
            ws = td / "ws"
            upper, work, lower = (ws / "fake_target" / d for d in ("upper", "work", "lower"))
            for d in (upper, work, lower):
                d.mkdir(parents=True, exist_ok=True)
            (upper / "pending.log").write_text("cached-data")
            cfg = {"paths": "/fake/target = flush\n", "workspace_root": str(ws),
                   "workspace_type": "path"}

            state = {"mounted": True}
            cmds: List[List[str]] = []
            synced = {"merged": [], "upper": 0}

            class Result:
                def __init__(self, rc):
                    self.returncode = rc
                    self.stdout = ""

            def fake_run(cmd, check=True, quiet=False, **kw):
                cmds.append(list(cmd))
                if cmd[:1] == ["umount"]:
                    if "-l" in cmd:
                        state["mounted"] = False
                        return Result(0)
                    return Result(32)  # always busy
                return Result(0)

            def fake_merged(merged, lower_dir, **kw):
                synced["merged"].append(merged)
                return True

            def fake_upper(*a, **kw):
                synced["upper"] += 1
                return True

            names = ("is_workspace_ready", "live_overlay_dirs", "is_overlay_mounted",
                     "is_mountpoint", "lower_bind_is_safe", "remount_lower",
                     "rsync_merged_to_lower", "rsync_upper_to_lower", "run",
                     "sync_upper_attrs", "unmount_if_idle", "orphan_entries")
            orig = {k: getattr(mod, k) for k in names}
            orig_idle = mod.unmount_if_idle
            try:
                target = str(Path("/fake/target").resolve())
                mod.is_workspace_ready = lambda cfg: True
                # The host's own overlays must not reach the fakes below.
                mod.orphan_entries = lambda cfg: []
                mod.live_overlay_dirs = lambda p: (
                    {"lower": str(lower), "upper": str(upper), "work": str(work)}
                    if state["mounted"] else None)
                mod.is_overlay_mounted = lambda p: state["mounted"] and str(p) == target
                mod.is_mountpoint = lambda p: state["mounted"] or str(p) != target
                mod.lower_bind_is_safe = lambda p: True
                mod.remount_lower = lambda lower, rw, check=False: True
                mod.rsync_merged_to_lower = fake_merged
                mod.rsync_upper_to_lower = fake_upper
                mod.run = fake_run
                mod.sync_upper_attrs = lambda *a, **kw: None
                # Same logic, without the real 1s back-off between attempts.
                mod.unmount_if_idle = lambda p: orig_idle(p, delay=0)
                result = mod.flush(cfg, teardown=teardown)
            finally:
                for k, v in orig.items():
                    setattr(mod, k, v)
            kept = (upper / "pending.log").is_file()
            return result, cmds, synced, kept, state["mounted"]

        ok_all = True

        def check(label, cond):
            nonlocal ok_all
            log_test(f"  {label}: {'OK' if cond else 'FAIL'}", "SUCCESS" if cond else "FAIL")
            ok_all &= bool(cond)

        with tempfile.TemporaryDirectory() as td:
            result, cmds, synced, kept, mounted = attempt(Path(td), teardown=False)
        check("routine flush never lazy-unmounts",
              not any(c[:1] == ["umount"] and "-l" in c for c in cmds))
        check("busy overlay stays mounted", mounted)
        check("merged view synced once", synced["merged"] == [str(Path("/fake/target").resolve())])
        check("upper not used as the sync source", synced["upper"] == 0)
        check("upper kept (open files still write to it)", kept)
        check("no overlay remount", not any(c[:2] == ["mount", "-t"] for c in cmds))
        check("flush reports success", result is True)

        # Control: a teardown still unmounts (lazily, if it must) and syncs the upper.
        with tempfile.TemporaryDirectory() as td:
            result, cmds, synced, kept, mounted = attempt(Path(td), teardown=True)
        check("teardown does not sync in place", synced["merged"] == [])
        check("teardown syncs the upper", synced["upper"] == 1)

        log_test("busy flush in place: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return bool(ok_all)

    def test_flush_in_place_keeps_open_files(self) -> bool:
        """Real overlay: flushing while a file is held open must write its data
        to disk, apply deletions, and leave later writes to that fd visible.
        A mount below the path must be skipped, not deleted from the lower."""
        log_test("\n>>> Test: in-place flush with an open file (real overlay)", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False
        if not hasattr(mod, "_flush_in_place"):
            log_test(f"{SCRIPT_PATH} has no in-place flush", "FAIL")
            return False

        ok_all = True

        def check(label, cond):
            nonlocal ok_all
            log_test(f"  {label}: {'OK' if cond else 'FAIL'}", "SUCCESS" if cond else "FAIL")
            ok_all &= bool(cond)

        td = Path(tempfile.mkdtemp(prefix="wc-inplace-"))
        target = (td / "target").resolve()
        ws = td / "ws"
        name = mod.sanitize_name(str(target))
        lower, upper, work = (ws / name / d for d in ("lower", "upper", "work"))
        sub = target / "sub"
        mounted = []
        fh = None
        try:
            for d in (target, lower, upper, work, sub):
                d.mkdir(parents=True, exist_ok=True)
            (target / "old.log").write_text("old\n")
            (target / "gone.log").write_text("gone\n")
            (sub / "keep").write_text("keep\n")

            def sh(*cmd):
                r = subprocess.run(list(cmd), stdout=subprocess.PIPE,
                                   stderr=subprocess.STDOUT, text=True)
                if r.returncode != 0:
                    raise RuntimeError(f"{' '.join(cmd)}: {r.stdout.strip()}")

            sh("mount", "--bind", str(target), str(lower)); mounted.append(lower)
            sh("mount", "--make-private", str(lower))
            sh("mount", "-o", "remount,bind,ro", str(lower))
            sh("mount", "-t", "overlay", "overlay", "-o",
               f"lowerdir={lower},upperdir={upper},workdir={work},"
               f"metacopy=off,redirect_dir=off,index=off", str(target)); mounted.append(target)
            # Hides sub/keep in the merged view; --delete must not take it from lower.
            sh("mount", "-t", "tmpfs", "tmpfs", str(sub)); mounted.append(sub)

            fh = open(target / "app.log", "a")
            fh.write("before\n"); fh.flush()
            os.remove(target / "gone.log")

            entry = mod.PathEntry(str(target), "flush", None, None, None)
            errors = mod._flush_path(entry, ws, teardown=False)

            fh.write("after\n"); fh.flush()

            check("flush reported no errors", errors == 0)
            check("overlay still mounted", check_overlay_mounted(str(target)))
            check("open file's data reached disk",
                  (lower / "app.log").is_file() and (lower / "app.log").read_text() == "before\n")
            check("write after flush not lost",
                  (target / "app.log").read_text() == "before\nafter\n")
            check("deletion applied to disk", not (lower / "gone.log").exists())
            check("untouched file intact", (lower / "old.log").read_text() == "old\n")
            check("mount below path skipped", (lower / "sub" / "keep").is_file())
        except Exception as e:
            log_test(f"  setup/flush error: {e}", "FAIL")
            ok_all = False
        finally:
            if fh:
                fh.close()
            for m in reversed(mounted):
                subprocess.run(["umount", str(m)], stdout=subprocess.DEVNULL,
                               stderr=subprocess.DEVNULL)
            shutil.rmtree(td, ignore_errors=True)

        log_test("in-place flush with open file: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return bool(ok_all)

    def test_unmount_escalation_order(self) -> bool:
        """Regression: force_unmount_with_retry() must try a plain umount before a
        lazy one.

        'umount -l' detaches the mount but keeps it alive for processes holding
        open fds, and it leaves /proc/mounts immediately so it looks successful.
        Those processes keep writing into the detached upper that the caller is
        about to rsync and clear, so their writes are lost. Lazy must remain
        available as the last resort (journald keeps /var/log busy, and refusing
        would break flush entirely) but must never be the first attempt."""
        log_test("\n>>> Test: unmount escalates plain -> kill -> lazy", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        cmds: List[List[str]] = []

        class FakeResult:
            # Every unmount "fails" so the full escalation ladder is exercised
            # and wait_for_unmount() is never entered.
            returncode = 1
            stdout = ""

        orig_run = mod.run
        orig_is_mp = mod.is_mountpoint
        orig_is_ov = mod.is_overlay_mounted
        try:
            def fake_run(cmd, check=True, quiet=False):
                cmds.append(list(cmd))
                return FakeResult()

            mod.run = fake_run
            mod.is_mountpoint = lambda p: True
            mod.is_overlay_mounted = lambda p: True
            mod.force_unmount_with_retry(Path("/fake/busy"), max_retries=2)
        finally:
            mod.run = orig_run
            mod.is_mountpoint = orig_is_mp
            mod.is_overlay_mounted = orig_is_ov

        def idx(pred) -> int:
            for i, c in enumerate(cmds):
                if pred(c):
                    return i
            return -1

        plain = idx(lambda c: c[:1] == ["umount"] and "-l" not in c)
        lazy = idx(lambda c: c[:1] == ["umount"] and "-l" in c)
        fuser = idx(lambda c: c[:1] == ["fuser"])

        ok_all = True

        first_ok = (plain != -1 and (lazy == -1 or plain < lazy))
        log_test(f"  plain umount tried before lazy: {'OK' if first_ok else 'FAIL'} "
                 f"(plain@{plain} lazy@{lazy})", "SUCCESS" if first_ok else "FAIL")
        ok_all &= first_ok

        kill_ok = (fuser == -1 or (plain != -1 and plain < fuser))
        log_test(f"  holders killed only after a plain umount failed: "
                 f"{'OK' if kill_ok else 'FAIL'} (fuser@{fuser})",
                 "SUCCESS" if kill_ok else "FAIL")
        ok_all &= kill_ok

        # Lazy must still be reachable, or busy paths like /var/log never unmount.
        lazy_ok = (lazy != -1)
        log_test(f"  lazy umount still available as last resort: "
                 f"{'OK' if lazy_ok else 'FAIL'}", "SUCCESS" if lazy_ok else "FAIL")
        ok_all &= lazy_ok

        log_test("unmount escalation order: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return bool(ok_all)

    def test_unmount_skips_kill_for_var_log(self) -> bool:
        """Regression: force_unmount_with_retry(/var/log) must never run 'fuser'.

        journald/rsyslog are respawned by systemd and reopen /var/log within
        milliseconds, so the kill escalation can never make a plain umount win --
        it only SIGKILLs journald plus every other process with an fd on that
        filesystem, twice, before falling back to lazy anyway. For paths in
        LAZY_UNMOUNT_PATHS the code must go plain umount -> lazy, with no kill."""
        log_test("\n>>> Test: /var/log unmount skips fuser kill", "INFO")

        try:
            mod = load_writecache_module()
        except Exception as e:
            log_test(f"Could not import omv-writecache: {e}", "FAIL")
            return False

        cmds: List[List[str]] = []

        class FakeResult:
            returncode = 1
            stdout = ""

        orig_run = mod.run
        orig_is_mp = mod.is_mountpoint
        orig_is_ov = mod.is_overlay_mounted
        try:
            def fake_run(cmd, check=True, quiet=False):
                cmds.append(list(cmd))
                return FakeResult()

            mod.run = fake_run
            mod.is_mountpoint = lambda p: True
            mod.is_overlay_mounted = lambda p: True
            mod.force_unmount_with_retry(Path("/var/log"), max_retries=3)
        finally:
            mod.run = orig_run
            mod.is_mountpoint = orig_is_mp
            mod.is_overlay_mounted = orig_is_ov

        ran_fuser = any(c[:1] == ["fuser"] for c in cmds)
        ran_plain = any(c[:1] == ["umount"] and "-l" not in c for c in cmds)
        ran_lazy = any(c[:1] == ["umount"] and "-l" in c for c in cmds)

        ok_all = True

        no_kill = not ran_fuser
        log_test(f"  fuser never invoked for /var/log: {'OK' if no_kill else 'FAIL'}",
                 "SUCCESS" if no_kill else "FAIL")
        ok_all &= no_kill

        plain_first = ran_plain
        log_test(f"  plain umount still attempted first: {'OK' if plain_first else 'FAIL'}",
                 "SUCCESS" if plain_first else "FAIL")
        ok_all &= plain_first

        lazy_last = ran_lazy
        log_test(f"  lazy umount still reached as fallback: {'OK' if lazy_last else 'FAIL'}",
                 "SUCCESS" if lazy_last else "FAIL")
        ok_all &= lazy_last

        log_test("var/log unmount skips kill: " + ("OK" if ok_all else "FAIL"),
                 "SUCCESS" if ok_all else "FAIL")
        return bool(ok_all)

    # ========== MAIN TEST RUNNER ==========

    def run(self) -> Dict[str, bool]:
        results = {
            "mount": False,
            "file_creation": False,
            "flush": False,
            "deletion": False,
            "unmount": False,
            "persistence": False,
            "journald": False,
            "concurrent_ops": False,
            "disk_exhaustion": False,
            "special_permissions": False,
            "symlinks": False,
            "deep_directories": False,
            "whiteout_scenarios": False,
            "status_command": False,
            "large_files": False,
            "rapid_io": False,
            "double_mount": False,
            "empty_directories": False,
            "long_filenames": False,
            "unicode_filenames": False,
            "dropin_file": False,
            "mount_unit_ordering": False,
            "workspace_survives_flush": False,
            "flush_preserves_dir_ownership": False,
            "flush_enforces_dir_ownership": False,
            "live_overlay_dirs_parsing": False,
            "flush_lower_bind_safe": False,
            "flush_preserves_upper_on_failure": False,
            "flush_aborts_on_readonly_lower": False,
            "unmount_escalation_order": False,
            "flush_busy_syncs_in_place": False,
            "flush_in_place_keeps_open_files": False,
            "postfix_baseline": False,
            "postfix_after_mount": False,
            "postfix_after_flush": False,
            "postfix_after_unmount": False,
            "private_tmp_detection": False,
        }

        try:
            # Baseline Postfix health before any mount/flush/unmount
            log_test("\n>>> Test 0: postfix baseline", "INFO")
            results["postfix_baseline"], _ = check_postfix("baseline")

            # Drop-in file (static check, before any mounting)
            results["dropin_file"] = self.test_dropin_file()
            results["setup_unit_kill_mode"] = self.test_setup_unit_kill_mode()

            # Unit test: live overlay dir parsing (no mounting required)
            results["live_overlay_dirs_parsing"] = self.test_live_overlay_dirs_parsing()

            # Unit tests: flush failure paths and unmount escalation. These run
            # against a fake workspace and touch no real mounts, so they are safe
            # to run before anything is mounted.
            results["flush_preserves_upper_on_failure"] = self.test_flush_preserves_upper_on_failure()
            results["flush_aborts_on_readonly_lower"] = self.test_flush_aborts_on_readonly_lower()
            results["flush_busy_syncs_in_place"] = self.test_flush_busy_syncs_in_place()
            # Real mounts, but only of its own temp dirs.
            results["flush_in_place_keeps_open_files"] = self.test_flush_in_place_keeps_open_files()
            results["unmount_escalation_order"] = self.test_unmount_escalation_order()
            results["unmount_skips_kill_for_var_log"] = self.test_unmount_skips_kill_for_var_log()

            # Mount
            log_test("\n>>> Test 1: mount", "INFO")
            ok, _ = run_cmd("mount")
            results["mount"] = ok
            if not ok:
                log_test("Mount failed, aborting.", "FAIL")
                return results

            # Verify overlays report mounted
            for path in self.config_paths.keys():
                if check_overlay_mounted(path):
                    log_test(f"Overlay mounted: {path}", "SUCCESS")
                else:
                    log_test(f"Overlay NOT mounted: {path}", "FAIL")
                    results["mount"] = False

            # Verify drop-in was applied to the auto-generated mount unit
            results["mount_unit_ordering"] = self.test_mount_unit_ordering()

            # Check Postfix after mount
            results["postfix_after_mount"], _ = check_postfix("mount")

            # Create files inside overlays
            log_test("\n>>> Test 2: create files", "INFO")
            self.create_test_files(tag="t1")
            time.sleep(1)
            results["file_creation"] = self.verify_files(tag="t1", should_exist=True)
            log_test("File creation/verify: " + ("OK" if results["file_creation"] else "FAIL"),
                     "SUCCESS" if results["file_creation"] else "FAIL")

            # Journald test
            log_test("\n>>> Test 3: journald receive", "INFO")
            results["journald"] = test_journald_receive()

            # Flush
            log_test("\n>>> Test 4: flush", "INFO")
            ok, _ = run_cmd("flush")
            results["flush"] = ok
            if not ok:
                log_test("Flush failed.", "FAIL")
            else:
                still_ok = self.verify_files(tag="t1", should_exist=True)
                results["flush"] = results["flush"] and still_ok
                log_test("Flush verify: " + ("OK" if still_ok else "FAIL"),
                         "SUCCESS" if still_ok else "FAIL")

            # Verify workspace tmpfs survived the flush
            results["workspace_survives_flush"] = self.test_workspace_survives_flush()

            # Verify flush does not corrupt the managed directory's uid/gid/mode
            results["flush_preserves_dir_ownership"] = self.test_flush_preserves_dir_ownership()

            # Verify flush applies user-specified owner=/mode= attrs when configured
            results["flush_enforces_dir_ownership"] = self.test_enforced_dir_ownership()

            # Regression: flush must not capture its own overlay as the lower bind
            results["flush_lower_bind_safe"] = self.test_flush_lower_bind_stays_safe()

            # Check Postfix after flush
            results["postfix_after_flush"], _ = check_postfix("flush")

            # Delete files (whiteout behavior)
            log_test("\n>>> Test 5: delete files", "INFO")
            self.delete_test_files(tag="t1")
            time.sleep(1)
            results["deletion"] = self.verify_files(tag="t1", should_exist=False)
            log_test("Deletion verify: " + ("OK" if results["deletion"] else "FAIL"),
                     "SUCCESS" if results["deletion"] else "FAIL")

            # Run before unmount
            results["concurrent_ops"] = self.test_concurrent_operations()
            results["special_permissions"] = self.test_special_permissions()
            results["symlinks"] = self.test_symlinks()
            results["deep_directories"] = self.test_deep_directories()
            results["whiteout_scenarios"] = self.test_whiteout_scenarios()
            results["status_command"] = self.test_status_command()
            results["large_files"] = self.test_large_files()
            results["rapid_io"] = self.test_rapid_operations()
            results["double_mount"] = self.test_double_mount()
            results["empty_directories"] = self.test_empty_directories()
            results["long_filenames"] = self.test_long_filenames()
            results["unicode_filenames"] = self.test_unicode_filenames()
            results["disk_exhaustion"] = self.test_disk_space_exhaustion()
            results["parse_services_daily"] = self.test_parse_services_daily_actions()
            results["daily_flag_filtering"] = self.test_daily_flag_service_filtering()
            results["services_running_only"] = self.test_services_only_touch_running_units()
            results["reconfigure_no_kill"] = self.test_reconfigure_unmount_no_kill()
            results["reconfigure_flag"] = self.test_reconfigure_flag_parsing()
            results["private_tmp_detection"] = self.test_private_tmp_detection()

            # Create more files, then unmount
            log_test("\n>>> Prepare for unmount: create more files", "INFO")
            self.create_test_files(tag="t2")
            time.sleep(1)

            log_test("\n>>> Test 6: unmount", "INFO")
            ok, _ = run_cmd("unmount")
            results["unmount"] = ok
            if not ok:
                log_test("Unmount failed.", "FAIL")
            else:
                persisted_ok = self.verify_persistence_per_policy(tag="t2")
                results["persistence"] = persisted_ok
                log_test("Persistence after unmount (per policy): " + ("OK" if persisted_ok else "FAIL"),
                         "SUCCESS" if persisted_ok else "FAIL")

                # Check Postfix after unmount
                results["postfix_after_unmount"], _ = check_postfix("unmount")

            log_test("\n>>> Cleanup", "INFO")
            self.delete_test_files(tag="t1")
            self.delete_test_files(tag="t2")
            self.cleanup_tracker.cleanup_all()

            # Final remount (sanity)
            log_test("\n>>> Remount sanity", "INFO")
            run_cmd("mount")

            # Summary
            log_test("\n===== TEST SUMMARY =====", "INFO")
            all_ok = True
            passed = 0
            failed = 0
            for k, v in results.items():
                all_ok &= v
                if v:
                    passed += 1
                else:
                    failed += 1
                log_test(f"{k.ljust(20)}: " + ("PASSED" if v else "FAILED"),
                         "SUCCESS" if v else "FAIL")

            log_test(f"\nTotal: {passed} passed, {failed} failed",
                     "SUCCESS" if all_ok else "FAIL")

            return results

        finally:
            # Ensure cleanup happens even if tests are interrupted
            log_test("\n>>> Final cleanup", "INFO")
            self.cleanup_tracker.cleanup_all()

def test_journald_receive() -> bool:
    tag = "wc_journald_test"
    msg = f"journald test @{int(time.time())}"
    # If systemd-cat is missing, fall back to logger only
    has_systemd_cat = shutil.which("systemd-cat") is not None
    run_sh(["logger", "-t", tag, msg], timeout=10)
    if has_systemd_cat:
        run_sh(["/bin/sh", "-lc", f'echo "{msg}" | systemd-cat -t {tag}'], timeout=10)
    time.sleep(0.5)
    ok = journal_has(tag, "journald test", since_epoch=time.time()-120)
    log_test(f"journald receive {'OK' if ok else 'FAIL'} (tag={tag})", "SUCCESS" if ok else "FAIL")
    return ok


# -------- Main --------

def main():
    print(colored("=" * 60, Colors.BOLD))
    print(colored("OMV-WRITECACHE TEST", Colors.BOLD))
    print(colored("=" * 60, Colors.BOLD))

    require_root()
    ensure_script()

    ver = subprocess.run([SCRIPT_PATH, "version"], stdout=subprocess.PIPE,
                         stderr=subprocess.DEVNULL, text=True).stdout.strip()
    log_test(f"omv-writecache-test v{TEST_VERSION}, testing {SCRIPT_PATH} v{ver or '?'}", "INFO")

    # Load config once early to fail fast if malformed
    try:
        cfg_paths = load_config_paths()
        log_test(f"Loaded {len(cfg_paths)} configured paths from {CONFIG_PATH}", "INFO")
        for p, m in cfg_paths.items():
            log_test(f"  {p} = {m}", "DEBUG")
    except Exception as e:
        log_test(f"Error reading config: {e}", "FAIL")
        sys.exit(1)

    tester = WriteCacheTest()
    try:
        results = tester.run()
        results_file = Path("/tmp/omv-writecache-test-results.json")
        with open(results_file, "w") as f:
            json.dump(results, f, indent=2)
        print(f"\nResults saved to: {results_file}")
        sys.exit(0 if all(results.values()) else 1)
    except KeyboardInterrupt:
        print(colored("\nInterrupted", Colors.YELLOW))
        log_test("\n>>> Emergency cleanup after interrupt", "WARN")
        tester.cleanup_tracker.cleanup_all()
        sys.exit(130)
    except Exception as e:
        print(colored(f"\nTest failed: {e}", Colors.RED))
        import traceback; traceback.print_exc()
        log_test("\n>>> Emergency cleanup after error", "WARN")
        tester.cleanup_tracker.cleanup_all()
        sys.exit(1)
    finally:
        log_test("Test complete.", "INFO")
        print(f"\nTest log saved to: {TEST_LOG}")

if __name__ == "__main__":
    try:
        os.chdir("/")
    except Exception:
        pass

    main()
