"""Test mask_tmp_mount_for_zfs — the tmpfs-over-ZFS /tmp guard.

When the pool carries a /tmp dataset, systemd's tmp.mount (tmpfs) races it at
boot; when tmpfs wins, the dataset is shadowed and systemd-tmpfiles-clean
fails repeatedly with "Protocol driver not attached" (velox, 2026-04-10). The
fix is masking tmp.mount so the dataset owns /tmp — but only when such a
dataset actually exists, and never on a machine without zfs at all.

Method: sed-extract mask_tmp_mount_for_zfs from the real `archsetup`; fake
zfs / run_task / systemctl. The zfs-binary-absent case runs with a stripped
PATH containing only the tools the function itself needs.

Run from repo root:
    python3 -m unittest tests.installer-steps.test_mask_tmp_mount_for_zfs
"""

import os
import re
import subprocess
import tempfile
import textwrap
import unittest

REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
ARCHSETUP = os.path.join(REPO_ROOT, "archsetup")


def run(zfs_body=None, strip_zfs_from_path=False):
    """zfs_body: bash body for a fake zfs function, or None for no fake.
    strip_zfs_from_path: run with a minimal PATH that has grep/sed but no zfs.
    """
    with tempfile.TemporaryDirectory() as d:
        path_setup = ""
        if strip_zfs_from_path:
            fakebin = os.path.join(d, "bin")
            os.makedirs(fakebin)
            for tool in ("grep", "sed", "cat"):
                src = subprocess.run(["bash", "-lc", f"command -v {tool}"],
                                     capture_output=True, text=True).stdout.strip()
                if src:
                    os.symlink(src, os.path.join(fakebin, tool))
            path_setup = f'PATH="{fakebin}"'
        zfs_fake = f"zfs() {{ {zfs_body} }}" if zfs_body is not None else ""
        script = textwrap.dedent(f"""\
            logfile=/dev/null
            action=""
            display() {{ :; }}
            run_task() {{ echo "TASK: $1"; shift; "$@"; }}
            systemctl() {{ echo "SYSTEMCTL: $*"; }}
            error_warn() {{ echo "WARN: $1"; return 1; }}
            {zfs_fake}
            source <(sed -n '/^mask_tmp_mount_for_zfs() {{/,/^}}/p' "{ARCHSETUP}")
            {path_setup}
            mask_tmp_mount_for_zfs
            echo "RC=$?"
            exit 0
        """)
        return subprocess.run(
            ["bash", "-c", script], capture_output=True, text=True, timeout=10,
        )


def rc_of(r):
    m = re.search(r"^RC=(\d+)$", r.stdout, re.M)
    assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr)
    return int(m.group(1))


class MaskTmpMountForZfs(unittest.TestCase):
    # ------------------------------------------------------------ normal ----
    def test_tmp_dataset_present_masks_tmp_mount(self):
        r = run(zfs_body='printf "/\\n/home\\n/tmp\\n";')
        self.assertIn("SYSTEMCTL: mask tmp.mount", r.stdout)
        self.assertEqual(rc_of(r), 0)

    # ---------------------------------------------------------- boundary ----
    def test_no_tmp_dataset_is_a_no_op(self):
        r = run(zfs_body='printf "/\\n/home\\n/var\\n";')
        self.assertNotIn("SYSTEMCTL:", r.stdout)
        self.assertEqual(rc_of(r), 0)

    def test_tmp_prefix_dataset_does_not_match(self):
        # /tmp/scratch or /tmpfoo must not trigger the mask — exact match only.
        r = run(zfs_body='printf "/\\n/tmp/scratch\\n/tmpfoo\\n";')
        self.assertNotIn("SYSTEMCTL:", r.stdout)
        self.assertEqual(rc_of(r), 0)

    def test_legacy_and_none_mountpoints_do_not_match(self):
        r = run(zfs_body='printf "legacy\\nnone\\n-\\n";')
        self.assertNotIn("SYSTEMCTL:", r.stdout)
        self.assertEqual(rc_of(r), 0)

    # ------------------------------------------------------------- error ----
    def test_zfs_binary_absent_is_a_silent_no_op(self):
        r = run(zfs_body=None, strip_zfs_from_path=True)
        self.assertNotIn("SYSTEMCTL:", r.stdout)
        self.assertNotIn("WARN:", r.stdout)
        self.assertEqual(rc_of(r), 0)

    def test_zfs_list_failure_is_a_no_op_not_a_crash(self):
        r = run(zfs_body='echo "no pools available" >&2; return 1;')
        self.assertNotIn("SYSTEMCTL:", r.stdout)
        self.assertEqual(rc_of(r), 0)
        self.assertNotIn("no pools available", r.stdout,
                         "zfs stderr noise must not leak into output")


if __name__ == "__main__":
    unittest.main()