"""Test configure_tunnel_dns_over_tls — per-link DoT off for VPN tunnels.

The resolved drop-in pins DNSOverTLS=yes globally. Proton VPN (proton0) and
the static Proton WireGuard profiles (wgpvpn) push an in-tunnel resolver,
10.2.0.1, that answers plain port 53 and never completes TLS on 853, so
every lookup through the tunnel hangs. The Proton client recreates its NM
profile on each connect, so a per-profile setting can't stick; a
NetworkManager [connection-*] default matched on the interface names is
what NM pushes to resolved on every activation. Diagnosed 2026-09-09/10,
verified live on ratio and velox.

Method: sed-extract configure_tunnel_dns_over_tls from the real
`archsetup`, point it at a temp conf.d, and assert on the file it writes.

    python3 -m unittest tests.installer-steps.test_configure_tunnel_dns_over_tls
"""

import os
import re
import stat
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")
FILENAME = "tunnel-dns-over-tls.conf"


def run(confdir):
    script = textwrap.dedent(f"""\
        logfile=/dev/null
        action=""
        display() {{ :; }}
        error_warn() {{ echo "WARN: $1"; return 1; }}
        source <(sed -n '/^configure_tunnel_dns_over_tls() {{/,/^}}/p' "{ARCHSETUP}")
        configure_tunnel_dns_over_tls "{confdir}"
        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))


def body(confdir):
    with open(os.path.join(confdir, FILENAME)) as f:
        return f.read()


def directives(text):
    """The non-comment, non-blank lines — what NM actually parses."""
    return [l.strip() for l in text.splitlines()
            if l.strip() and not l.lstrip().startswith("#")]


class ConfigureTunnelDnsOverTls(unittest.TestCase):
    # ------------------------------------------------------------ normal ----
    def test_writes_a_connection_default_that_turns_dot_off(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d)
            self.assertEqual(rc_of(r), 0)
            lines = directives(body(d))
            self.assertIn("[connection-tunnel-dot]", lines)
            self.assertIn("connection.dns-over-tls=0", lines)

    def test_matches_both_proton_interfaces_and_nothing_else(self):
        # proton0 is the Proton client; wgpvpn is the static WireGuard
        # profiles. A match-device line without both leaves one tunnel
        # broken; one with a wildcard would turn DoT off for wifi too.
        with tempfile.TemporaryDirectory() as d:
            run(d)
            match = [l for l in directives(body(d)) if l.startswith("match-device=")]
            self.assertEqual(len(match), 1)
            devices = match[0].split("=", 1)[1].split(",")
            self.assertEqual(sorted(devices),
                             ["interface-name:proton0", "interface-name:wgpvpn"])

    def test_explains_why_the_global_setting_is_overridden_per_link(self):
        # The file contradicts the resolved drop-in next to it, so it has
        # to carry the reason: the in-tunnel resolver that never completes
        # TLS, and why the Proton client can't hold a per-profile setting.
        with tempfile.TemporaryDirectory() as d:
            run(d)
            text = body(d).lower()
            self.assertIn("10.2.0.1", text)
            self.assertIn("853", text)
            self.assertIn("recreates", text)

    def test_drop_in_is_world_readable_not_writable(self):
        with tempfile.TemporaryDirectory() as d:
            run(d)
            mode = stat.S_IMODE(os.stat(os.path.join(d, FILENAME)).st_mode)
            self.assertEqual(mode, 0o644)

    # ---------------------------------------------------------- boundary ----
    def test_running_twice_leaves_one_identical_file(self):
        with tempfile.TemporaryDirectory() as d:
            run(d)
            first = body(d)
            r = run(d)
            self.assertEqual(rc_of(r), 0)
            self.assertEqual(first, body(d))
            self.assertEqual(os.listdir(d), [FILENAME])

    def test_absent_directory_is_created(self):
        with tempfile.TemporaryDirectory() as d:
            nested = os.path.join(d, "etc", "NetworkManager", "conf.d")
            r = run(nested)
            self.assertEqual(rc_of(r), 0)
            self.assertIn("connection.dns-over-tls=0", directives(body(nested)))

    def test_leaves_sibling_drop_ins_alone(self):
        # dns.conf, wifi-privacy.conf and wifi-powersave-off.conf share the
        # directory; this step must add a file, never rewrite the dir.
        with tempfile.TemporaryDirectory() as d:
            sibling = os.path.join(d, "dns.conf")
            with open(sibling, "w") as f:
                f.write("[main]\ndns=systemd-resolved\n")
            run(d)
            with open(sibling) as f:
                self.assertEqual(f.read(), "[main]\ndns=systemd-resolved\n")

    # ------------------------------------------------------------- error ----
    @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits")
    def test_unwritable_directory_warns_and_does_not_crash(self):
        with tempfile.TemporaryDirectory() as d:
            confdir = os.path.join(d, "ro")
            os.mkdir(confdir)
            os.chmod(confdir, 0o500)
            try:
                r = run(confdir)
                self.assertIn("WARN:", r.stdout)
                self.assertNotEqual(rc_of(r), 0)
            finally:
                os.chmod(confdir, 0o700)


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