DDNS/tests/test_main.py

# -*- coding: utf-8 -*-
"""
Unit tests for the DDNS main run path.
"""

import io
import sys
import threading

from __init__ import MagicMock, patch, unittest

from ddns import __main__
from ddns.config.config import Config


class TestMain(unittest.TestCase):
    """Test the main DDNS run path."""

    @patch.object(__main__, "update_ip", return_value=True)
    @patch.object(__main__.Cache, "new")
    @patch.object(__main__, "get_provider_class")
    def test_run_passes_cache_max_age_to_cache(self, mock_provider_class, mock_cache_new, mock_update_ip):
        """Test run passes cache_max_age as the fourth Cache.new argument."""
        provider = MagicMock()
        mock_provider_class.return_value = lambda *args, **kwargs: provider
        config = Config(
            cli_config={"dns": "debug", "cache": True, "cache_max_age": 86400, "index4": False, "index6": False}
        )

        self.assertTrue(__main__.run(config))
        mock_cache_new.assert_called_once_with(True, config.md5(), __main__.logger, 86400)
        self.assertEqual(mock_update_ip.call_count, 2)

    @patch.object(__main__, "get_ip", side_effect=[None, "2001:db8::1"])
    @patch.object(__main__.Cache, "new", return_value=None)
    @patch.object(__main__, "get_provider_class")
    def test_run_attempts_ipv6_after_ipv4_failure(self, mock_provider_class, mock_cache_new, mock_get_ip):
        """An unavailable IPv4 address must not prevent an IPv6 update."""
        provider = MagicMock()
        provider.set_record.return_value = True
        mock_provider_class.return_value = lambda *args, **kwargs: provider
        config = Config(
            cli_config={
                "dns": "debug",
                "ipv4": ["v4.example.com"],
                "ipv6": ["v6.example.com"],
                "index4": ["default"],
                "index6": ["default"],
            }
        )

        self.assertFalse(__main__.run(config))

        self.assertEqual([call[0][0] for call in mock_get_ip.call_args_list], ["4", "6"])
        provider.set_record.assert_called_once()
        self.assertEqual(provider.set_record.call_args[0], ("v6.example.com", "2001:db8::1"))
        self.assertEqual(provider.set_record.call_args[1]["record_type"], "AAAA")

    @patch.object(__main__, "_get_ip_from_rule", return_value="2001:db8::1")
    @patch.object(__main__.Cache, "new", return_value=None)
    @patch.object(__main__, "get_provider_class")
    def test_run_disabled_ipv4_does_not_block_ipv6(self, mock_provider_class, mock_cache_new, mock_get_rule):
        """Keeping disabled IPv4 domains must not disable the configured IPv6 family."""
        provider = MagicMock()
        provider.set_record.return_value = True
        mock_provider_class.return_value = lambda *args, **kwargs: provider
        config = Config(
            cli_config={
                "dns": "debug",
                "ipv4": ["v4.example.com"],
                "ipv6": ["v6.example.com"],
                "index4": False,
                "index6": ["default"],
            }
        )

        self.assertTrue(__main__.run(config))

        mock_get_rule.assert_called_once_with("6", "default")
        provider.set_record.assert_called_once()
        self.assertEqual(provider.set_record.call_args[1]["record_type"], "AAAA")

    @patch.object(__main__, "get_ip", return_value=False)
    def test_update_ip_skips_disabled_family(self, mock_get_ip):
        """An explicitly disabled family is skipped without discovery or writes."""
        provider = MagicMock()

        result = __main__.update_ip(
            provider, None, False, ["disabled.example.com"], "AAAA", Config(cli_config={"dns": "debug"})
        )

        self.assertIsNone(result)
        mock_get_ip.assert_not_called()
        provider.set_record.assert_not_called()

    @patch.object(__main__, "get_ip", return_value="192.0.2.1")
    def test_update_ip_reports_partial_failures_and_continues(self, mock_get_ip):
        """Later success must not hide an earlier failure, or stop subsequent domains."""
        config = Config(cli_config={"dns": "debug"})
        for outcomes in (
            [True, False, True],
            [False, True],
            [True, RuntimeError("update rejected"), True],
            [RuntimeError("update rejected"), True],
            [None, True],
        ):
            provider = MagicMock()
            provider.set_record.side_effect = outcomes
            domains = ["record{}.example.com".format(index) for index in range(len(outcomes))]
            cache = {}

            result = __main__.update_ip(provider, cache, ["default"], domains, "A", config)

            self.assertFalse(result, repr(outcomes))
            self.assertEqual(provider.set_record.call_count, len(domains))
            self.assertEqual(
                cache,
                {"{}:A".format(domain): "192.0.2.1" for domain, outcome in zip(domains, outcomes) if outcome is True},
            )

    @patch.object(__main__, "get_ip", return_value="192.0.2.1")
    def test_cached_success_does_not_mask_failed_update(self, mock_get_ip):
        """A cached record remains valid without making a different failed record successful."""
        provider = MagicMock()
        provider.set_record.return_value = False
        cache = {"cached.example.com:A": "192.0.2.1"}

        result = __main__.update_ip(
            provider,
            cache,
            ["default"],
            ["cached.example.com", "failed.example.com"],
            "A",
            Config(cli_config={"dns": "debug"}),
        )

        self.assertFalse(result)
        provider.set_record.assert_called_once()
        self.assertEqual(cache, {"cached.example.com:A": "192.0.2.1"})

    @patch.object(__main__, "get_ip", return_value="192.0.2.1")
    def test_update_ip_all_successes_populate_cache(self, mock_get_ip):
        """All successful records retain the existing cache behavior."""
        provider = MagicMock()
        provider.set_record.return_value = True
        cache = {}
        domains = ["first.example.com", "second.example.com"]

        self.assertTrue(
            __main__.update_ip(provider, cache, ["default"], domains, "A", Config(cli_config={"dns": "debug"}))
        )
        self.assertEqual(provider.set_record.call_count, 2)
        self.assertEqual(cache, {domain + ":A": "192.0.2.1" for domain in domains})

    @patch.object(__main__, "get_ip", return_value="192.0.2.1")
    @patch.object(__main__.Cache, "new", return_value=None)
    @patch.object(__main__, "get_provider_class")
    def test_main_reports_partial_domain_failure(self, mock_provider_class, mock_cache_new, mock_get_ip):
        """The process result must report any rejected domain, not just total failures."""
        provider = MagicMock()
        provider.set_record.side_effect = [True, False]
        mock_provider_class.return_value = lambda *args, **kwargs: provider
        config = Config(cli_config={"dns": "debug", "ipv4": ["first.example.com", "second.example.com"]})

        with patch.object(__main__, "load_configs", return_value=[config]):
            with patch.object(sys, "stdout", io.StringIO()):
                with self.assertRaises(SystemExit) as context:
                    __main__.main()

        self.assertEqual(context.exception.code, 1)
        self.assertEqual(provider.set_record.call_count, 2)

    @patch.object(__main__, "get_ip", return_value="192.0.2.1")
    def test_update_ip_stops_before_next_domain_when_cancelled(self, mock_get_ip):
        """Stop cooperative updates between configured DNS records."""
        cancelled = threading.Event()
        provider = MagicMock()
        provider.set_record.side_effect = lambda *args, **kwargs: cancelled.set() or True
        config = Config(cli_config={"dns": "debug"})

        with self.assertRaises(__main__.UpdateCancelled):
            __main__.update_ip(
                provider,
                None,
                ["public"],
                ["first.example.com", "second.example.com"],
                "A",
                config,
                cancelled=cancelled.is_set,
            )

        provider.set_record.assert_called_once()
        mock_get_ip.assert_called_once()

    @patch.object(__main__, "_get_ip_from_rule")
    def test_get_ip_stops_before_next_rule_when_cancelled(self, mock_get_rule):
        """Stop cooperative address discovery between configured rules."""
        cancelled = threading.Event()
        mock_get_rule.side_effect = lambda *args: cancelled.set()

        with self.assertRaises(__main__.UpdateCancelled):
            __main__.get_ip("4", ["first", "second"], cancelled=cancelled.is_set)

        mock_get_rule.assert_called_once_with("4", "first")

    def test_mcp_mode_does_not_write_windows_leading_line(self):
        """Keep stdout clean before the stdio protocol handler starts."""
        output = io.StringIO()

        with patch.object(sys, "argv", ["ddns", "mcp"]):
            with patch.object(sys, "platform", "win32"):
                with patch.object(sys, "stdout", output):
                    with patch.object(__main__, "load_configs", side_effect=SystemExit(0)):
                        with self.assertRaises(SystemExit):
                            __main__.main()

        self.assertEqual(output.getvalue(), "")


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