parsedmarc/tests/test_maps.py

"""Tests for the map-maintenance scripts under parsedmarc/resources/maps/.

These scripts are maintainer-only batch tooling — they do not ship in the
wheel — but they still need regression coverage because they enforce the
privacy and integrity rules for the reverse-DNS map data files."""

import contextlib
import io
import os
import shutil
import sys
import tempfile
import unittest
from unittest import mock


class TestMapScriptsIPDetection(unittest.TestCase):
    """Full-IP detection and PSL folding in the map-maintenance scripts."""

    def test_collect_domain_info_detects_full_ips(self):
        import parsedmarc.resources.maps.collect_domain_info as cdi

        # Dotted and dashed four-octet patterns with valid octets: detected.
        self.assertTrue(cdi._has_full_ip("74-208-244-234.cprapid.com"))
        self.assertTrue(cdi._has_full_ip("host.192.168.1.1.example.com"))
        self.assertTrue(cdi._has_full_ip("a-10-20-30-40-brand.com"))
        # Three octets is NOT a full IP — OVH's reverse-DNS pattern stays safe.
        self.assertFalse(cdi._has_full_ip("ip-147-135-108.us"))
        # Out-of-range octet fails the 0-255 sanity check.
        self.assertFalse(cdi._has_full_ip("999-1-2-3-foo.com"))
        # Pure domain, no IP.
        self.assertFalse(cdi._has_full_ip("example.com"))

    def test_find_unknown_detects_full_ips(self):
        import parsedmarc.resources.maps.find_unknown_base_reverse_dns as fu

        self.assertTrue(fu._has_full_ip("170-254-144-204-nobreinternet.com.br"))
        self.assertFalse(fu._has_full_ip("ip-147-135-108.us"))
        self.assertFalse(fu._has_full_ip("cprapid.com"))

    def test_apply_psl_override_dot_prefix(self):
        import parsedmarc.resources.maps.collect_domain_info as cdi

        ov = [".cprapid.com", ".linode.com"]
        self.assertEqual(cdi._apply_psl_override("foo.cprapid.com", ov), "cprapid.com")
        self.assertEqual(cdi._apply_psl_override("a.b.linode.com", ov), "linode.com")

    def test_apply_psl_override_dash_prefix(self):
        import parsedmarc.resources.maps.collect_domain_info as cdi

        ov = ["-nobre.com.br"]
        self.assertEqual(
            cdi._apply_psl_override("1-2-3-4-nobre.com.br", ov), "nobre.com.br"
        )

    def test_apply_psl_override_no_match(self):
        import parsedmarc.resources.maps.collect_domain_info as cdi

        ov = [".cprapid.com"]
        self.assertEqual(cdi._apply_psl_override("example.com", ov), "example.com")


class TestFindUnknownBaseReverseDNS(unittest.TestCase):
    """Missing-file error paths in find_unknown_base_reverse_dns.py's
    ``_main()``.

    Both sites below printed the intended ``"Error: ... does not exist"``
    message but had no ``sys.exit(1)`` after it, so execution fell through
    into the next ``open()`` call on the same missing path and raised an
    unhandled ``FileNotFoundError`` instead of the clean, intended exit.
    """

    def test_missing_known_unknown_list_exits_cleanly(self):
        """A missing known_unknown_base_reverse_dns.txt must print the
        intended error message and exit(1), not fall through into an
        unhandled FileNotFoundError from the subsequent open() call.

        Regression test: the nested ``load_list()`` helper in ``_main()``
        printed ``f"Error: {file_path} does not exist"`` but had no
        ``sys.exit(1)`` after it, so execution fell through into
        ``print(f"Loading {file_path}")`` and then ``open(file_path)`` on
        the same missing path, raising an unhandled ``FileNotFoundError``
        instead of the clean, intended error exit. The sibling
        duplicate-entry check a few lines below (``domain in list_var``)
        already did print-then-``sys.exit(1)``; this verifies the
        missing-file check now matches that style. Capturing stdout and
        asserting on the exact message (rather than only the exit code)
        pins the exit to this specific site, not just any ``sys.exit(1)``
        in the function.
        """
        import parsedmarc.resources.maps.find_unknown_base_reverse_dns as fu

        old_cwd = os.getcwd()
        tmp_dir = tempfile.mkdtemp()
        # Register rmtree first so, under LIFO cleanup ordering, chdir back
        # to old_cwd always runs before rmtree removes tmp_dir -- and, since
        # unittest runs each addCleanup independently, rmtree still runs
        # even if os.chdir were to raise.
        self.addCleanup(shutil.rmtree, tmp_dir, ignore_errors=True)
        self.addCleanup(os.chdir, old_cwd)
        os.chdir(tmp_dir)

        stdout = io.StringIO()
        with mock.patch.object(sys, "argv", ["find_unknown_base_reverse_dns.py"]):
            with contextlib.redirect_stdout(stdout):
                with self.assertRaises(SystemExit) as cm:
                    fu._main()
        self.assertEqual(cm.exception.code, 1)
        self.assertIn(
            "Error: known_unknown_base_reverse_dns.txt does not exist",
            stdout.getvalue(),
        )

    def test_missing_base_reverse_dns_map_exits_cleanly(self):
        """A missing base_reverse_dns_map.csv must print the intended error
        message and exit(1), not fall through into an unhandled
        FileNotFoundError from the subsequent open() call.

        Regression test for the second site fixed alongside the
        ``load_list()`` one above: ``_main()`` printed
        ``f"Error: {base_reverse_dns_map_file_path} does not exist"`` but
        had no ``sys.exit(1)`` after it. Reaching this check requires
        getting past the MMDB load first. ``_load_as_name_index`` does its
        external work entirely through ``maxminddb.open_database()`` --
        the actual SDK boundary per AGENTS.md's "mock at SDK boundaries"
        rule -- so that call is mocked to a context manager over an empty
        iterable instead of loading the real, ~23MB bundled MMDB or
        mocking an internal helper of this codebase.
        """
        import parsedmarc.resources.maps.find_unknown_base_reverse_dns as fu

        old_cwd = os.getcwd()
        tmp_dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, tmp_dir, ignore_errors=True)
        self.addCleanup(os.chdir, old_cwd)

        maps_dir = os.path.join(tmp_dir, "maps")
        ipinfo_dir = os.path.join(tmp_dir, "ipinfo")
        os.makedirs(maps_dir)
        os.makedirs(ipinfo_dir)
        # Both lists are loaded before the MMDB check and must exist, but
        # their content isn't exercised by this test.
        open(os.path.join(maps_dir, "known_unknown_base_reverse_dns.txt"), "w").close()
        open(os.path.join(maps_dir, "psl_overrides.txt"), "w").close()
        # A placeholder for the MMDB: only os.path.exists() touches this
        # path directly, since maxminddb.open_database() itself is mocked
        # below and never actually reads the file.
        open(os.path.join(ipinfo_dir, "ipinfo_lite.mmdb"), "w").close()
        # base_reverse_dns_map.csv is deliberately NOT created -- that's
        # the missing-file condition under test.

        os.chdir(maps_dir)

        class _EmptyMMDBReader:
            def __enter__(self):
                return iter(())

            def __exit__(self, *exc_info):
                return False

        stdout = io.StringIO()
        with mock.patch.object(sys, "argv", ["find_unknown_base_reverse_dns.py"]):
            with mock.patch("maxminddb.open_database", return_value=_EmptyMMDBReader()):
                with contextlib.redirect_stdout(stdout):
                    with self.assertRaises(SystemExit) as cm:
                        fu._main()
        self.assertEqual(cm.exception.code, 1)
        self.assertIn(
            "Error: base_reverse_dns_map.csv does not exist", stdout.getvalue()
        )


class TestDetectPSLOverrides(unittest.TestCase):
    """Cluster detection, brand-tail extraction, and full-pipeline behaviour
    for `detect_psl_overrides.py`."""

    def setUp(self):
        import parsedmarc.resources.maps.detect_psl_overrides as dpo

        self.dpo = dpo

    def test_extract_brand_tail_dot_separator(self):
        self.assertEqual(
            self.dpo.extract_brand_tail("74-208-244-234.cprapid.com"),
            ".cprapid.com",
        )

    def test_extract_brand_tail_dash_separator(self):
        self.assertEqual(
            self.dpo.extract_brand_tail("170-254-144-204-nobre.com.br"),
            "-nobre.com.br",
        )

    def test_extract_brand_tail_no_separator(self):
        self.assertEqual(
            self.dpo.extract_brand_tail("host134-254-143-190tigobusiness.com.ni"),
            "tigobusiness.com.ni",
        )

    def test_extract_brand_tail_no_ip_returns_none(self):
        self.assertIsNone(self.dpo.extract_brand_tail("plain.example.com"))

    def test_extract_brand_tail_rejects_short_tail(self):
        """A tail shorter than MIN_TAIL_LEN is rejected to avoid folding to `.com`."""
        # Four-octet IP followed by only `.br` (2 chars after the dot) — too short.
        self.assertIsNone(self.dpo.extract_brand_tail("1-2-3-4.br"))

    def test_detect_clusters_meets_threshold(self):
        domains = [
            "1-2-3-4.cprapid.com",
            "5-6-7-8.cprapid.com",
            "9-10-11-12.cprapid.com",
            "1-2-3-4-other.com.br",  # not enough of these
        ]
        clusters = self.dpo.detect_clusters(domains, threshold=3, known_overrides=set())
        self.assertIn(".cprapid.com", clusters)
        self.assertEqual(len(clusters[".cprapid.com"]), 3)
        self.assertNotIn("-other.com.br", clusters)

    def test_detect_clusters_honours_threshold(self):
        domains = [
            "1-2-3-4.cprapid.com",
            "5-6-7-8.cprapid.com",
        ]
        clusters = self.dpo.detect_clusters(domains, threshold=3, known_overrides=set())
        self.assertEqual(clusters, {})

    def test_detect_clusters_skips_known_overrides(self):
        """Tails already in psl_overrides.txt must not be re-proposed."""
        domains = [
            "1-2-3-4.cprapid.com",
            "5-6-7-8.cprapid.com",
            "9-10-11-12.cprapid.com",
        ]
        clusters = self.dpo.detect_clusters(
            domains, threshold=3, known_overrides={".cprapid.com"}
        )
        self.assertNotIn(".cprapid.com", clusters)

    def test_apply_override_matches_first(self):
        """apply_override iterates in list order and returns on the first match."""
        ov = [".cprapid.com", "-nobre.com.br"]
        self.assertEqual(
            self.dpo.apply_override("1-2-3-4.cprapid.com", ov), "cprapid.com"
        )
        self.assertEqual(
            self.dpo.apply_override("1-2-3-4-nobre.com.br", ov), "nobre.com.br"
        )
        self.assertEqual(self.dpo.apply_override("unrelated.com", ov), "unrelated.com")

    def test_has_full_ip_shared_with_other_scripts(self):
        """The detect script's IP check must agree with the other map scripts."""
        self.assertTrue(self.dpo.has_full_ip("74-208-244-234.cprapid.com"))
        self.assertFalse(self.dpo.has_full_ip("ip-147-135-108.us"))
        self.assertFalse(self.dpo.has_full_ip("example.com"))


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