DDNS/tests/test_scheduler_cron.py

# -*- coding:utf-8 -*-
"""
Unit tests for ddns.scheduler.cron module
@author: NewFuture
"""

import platform

from __init__ import patch, unittest

from ddns.scheduler.cron import CronScheduler


class TestCronScheduler(unittest.TestCase):
    """Test CronScheduler functionality"""

    def setUp(self):
        """Set up test fixtures"""
        self.scheduler = CronScheduler()

    @patch("ddns.scheduler._base.datetime")
    @patch("ddns.scheduler._base.version", "test-version")
    def test_install_with_version_and_date(self, mock_datetime):
        """Test install method includes version and date in cron entry"""
        mock_datetime.now.return_value.strftime.return_value = "2025-08-01 14:30:00"

        # Mock the methods to avoid actual system calls
        with patch("ddns.scheduler.cron.try_run") as mock_run:
            with patch.object(self.scheduler, "_update_crontab") as mock_update:
                with patch.object(self.scheduler, "_build_ddns_command") as mock_build:
                    mock_run.return_value = ""
                    mock_update.return_value = True
                    mock_build.return_value = "python3 -m ddns -c test.json"

                    result = self.scheduler.install(5, {"config": ["test.json"]})

                    self.assertTrue(result)
                    mock_update.assert_called_once()

                    # Verify the cron entry contains version and date
                    call_args = mock_update.call_args[0][0]
                    cron_entry = u"\n".join(call_args)  # fmt: skip
                    self.assertIn("# DDNS: auto-update vtest-version installed on 2025-08-01 14:30:00", cron_entry)

    def test_get_status_extracts_comments(self):
        """Test get_status method extracts comments from cron entry"""
        cron_entry = (
            '*/10 * * * * cd "/home/user" && python3 -m ddns -c test.json '
            "# DDNS: auto-update v4.0 installed on 2025-08-01 14:30:00"
        )

        with patch("ddns.scheduler.cron.try_run") as mock_run:

            def mock_command(cmd, **kwargs):
                if cmd == ["crontab", "-l"]:
                    return cron_entry
                elif cmd == ["pgrep", "-f", "cron"]:
                    return "12345"
                return None

            mock_run.side_effect = mock_command

            status = self.scheduler.get_status()

            self.assertEqual(status["scheduler"], "cron")
            self.assertTrue(status["enabled"])
            self.assertEqual(status["interval"], 10)
            self.assertEqual(status["description"], "auto-update v4.0 installed on 2025-08-01 14:30:00")

    def test_get_status_handles_missing_comment_info(self):
        """Test get_status handles cron entries without full comment info gracefully"""
        cron_entry = '*/5 * * * * cd "/home/user" && python3 -m ddns -c test.json # DDNS: auto-update'

        with patch("ddns.scheduler.cron.try_run") as mock_run:

            def mock_command(cmd, **kwargs):
                if cmd == ["crontab", "-l"]:
                    return cron_entry
                elif cmd == ["pgrep", "-f", "cron"]:
                    return None
                return None

            mock_run.side_effect = mock_command

            status = self.scheduler.get_status()

            self.assertEqual(status["scheduler"], "cron")
            self.assertTrue(status["enabled"])
            self.assertEqual(status["interval"], 5)
            self.assertEqual(status["description"], "auto-update")

    def test_version_in_cron_entry(self):
        """Test that install method includes version in cron entry"""
        with patch("ddns.scheduler._base.datetime") as mock_datetime:
            mock_datetime.now.return_value.strftime.return_value = "2025-08-01 14:30:00"

            with patch("ddns.scheduler.cron.try_run") as mock_run:
                with patch.object(self.scheduler, "_update_crontab") as mock_update:
                    with patch.object(self.scheduler, "_build_ddns_command") as mock_build:
                        mock_run.return_value = ""
                        mock_update.return_value = True
                        mock_build.return_value = "python3 -m ddns"

                        # Test that version is included in cron entry
                        with patch("ddns.scheduler._base.version", "test-version"):
                            result = self.scheduler.install(10)

                            self.assertTrue(result)
                            call_args = mock_update.call_args[0][0]
                            cron_entry = u"\n".join(call_args)  # fmt: skip
                            self.assertIn("vtest-version", cron_entry)  # Should include the version

    def test_get_status_with_no_comment(self):
        """Test get_status handles cron entries with no DDNS comment"""
        cron_entry = '*/15 * * * * cd "/home/user" && python3 -m ddns -c test.json'

        with patch("ddns.scheduler.cron.try_run") as mock_run:

            def mock_command(cmd, **kwargs):
                if cmd == ["crontab", "-l"]:
                    return cron_entry
                elif cmd == ["pgrep", "-f", "cron"]:
                    return None
                return None

            mock_run.side_effect = mock_command

            status = self.scheduler.get_status()

            self.assertEqual(status["scheduler"], "cron")
            self.assertEqual(status["enabled"], False)  # False when no DDNS line found
            # When no DDNS line is found, the method still tries to parse the empty line
            # This results in None values for interval, command, and empty string for comments
            self.assertIsNone(status.get("interval"))
            self.assertIsNone(status.get("command"))
            self.assertEqual(status.get("description"), "")

    def test_modify_cron_lines_enable_disable(self):
        """Test _modify_cron_lines method for enable and disable operations"""
        # Test enable operation on commented line
        with patch("ddns.scheduler.cron.try_run") as mock_run:
            with patch.object(self.scheduler, "_update_crontab") as mock_update:
                mock_run.return_value = "# */5 * * * * cd /path && python3 -m ddns # DDNS: auto-update"
                mock_update.return_value = True

                result = self.scheduler.enable()
                self.assertTrue(result)
                mock_update.assert_called_once()
                call_args = mock_update.call_args[0][0]
                cron_entry = u"\n".join(call_args)  # fmt: skip
                self.assertIn("*/5 * * * * cd /path && python3 -m ddns # DDNS: auto-update", cron_entry)

        # Test disable operation on active line
        with patch("ddns.scheduler.cron.try_run") as mock_run:
            with patch.object(self.scheduler, "_update_crontab") as mock_update:
                mock_run.return_value = "*/5 * * * * cd /path && python3 -m ddns # DDNS: auto-update"
                mock_update.return_value = True

                result = self.scheduler.disable()
                self.assertTrue(result)
                mock_update.assert_called_once()
                call_args = mock_update.call_args[0][0]
                cron_entry = u"\n".join(call_args)  # fmt: skip
                self.assertIn("# */5 * * * * cd /path && python3 -m ddns # DDNS: auto-update", cron_entry)

    def test_modify_cron_lines_uninstall(self):
        """Test _modify_cron_lines method for uninstall operation"""
        with patch("ddns.scheduler.cron.try_run") as mock_run:
            with patch.object(self.scheduler, "_update_crontab") as mock_update:
                mock_run.return_value = "*/5 * * * * cd /path && python3 -m ddns # DDNS: auto-update\nother cron job"
                mock_update.return_value = True

                result = self.scheduler.uninstall()
                self.assertTrue(result)
                mock_update.assert_called_once()
                call_args = mock_update.call_args[0][0]
                cron_entry = u"\n".join(call_args)  # fmt: skip
                self.assertNotIn("DDNS", cron_entry)
                self.assertIn("other cron job", cron_entry)

    @unittest.skipIf(platform.system().lower() == "windows", "Unix/Linux/macOS-specific test")
    def test_real_cron_integration(self):
        """Test real cron integration with actual system calls"""
        # Check if crontab command is available
        try:
            from ddns.util.try_run import try_run

            crontab_result = try_run(["crontab", "-l"])
            if not crontab_result:
                self.skipTest("crontab not available on this system")
        except Exception:
            self.skipTest("crontab not available on this system")

        try:
            status = self.scheduler.get_status()
            self.assertIsInstance(status, dict)
            self.assertEqual(status["scheduler"], "cron")
            self.assertIsInstance(status["installed"], bool)
        finally:
            pass

    def _setup_real_cron_test(self):
        """
        Helper method to set up real cron tests with common functionality
        """
        # Check if crontab is available first
        try:
            from ddns.util.try_run import try_run

            try_run(["crontab", "-l"])
        except Exception:
            self.skipTest("crontab not available on this system")

    def _cleanup_real_cron_test(self):
        """
        Helper method to clean up real cron tests
        """
        try:
            # Remove any test cron jobs
            if self.scheduler.is_installed():
                self.scheduler.uninstall()
        except Exception:
            pass

    @unittest.skipIf(platform.system().lower() == "windows", "Unix/Linux/macOS-specific test")
    def test_real_scheduler_methods_safe(self):
        """Test real scheduler methods that don't modify system state"""
        try:
            # Test is_installed (safe read-only operation)
            installed = self.scheduler.is_installed()
            self.assertIsInstance(installed, bool)

            # Test build command
            ddns_args = {"dns": "debug", "ipv4": ["test.example.com"]}
            command = self.scheduler._build_ddns_command(ddns_args)
            self.assertIsInstance(command, list)
            command_str = " ".join(command)
            self.assertIn("python", command_str.lower())

            # Test get status (safe read-only operation)
            status = self.scheduler.get_status()
            required_keys = ["scheduler", "installed", "enabled", "interval"]
            for key in required_keys:
                self.assertIn(key, status)

            # Test enable/disable without actual installation (should handle gracefully)
            enable_result = self.scheduler.enable()
            self.assertIsInstance(enable_result, bool)

            disable_result = self.scheduler.disable()
            self.assertIsInstance(disable_result, bool)
        finally:
            self._cleanup_real_cron_test()

    @unittest.skipIf(platform.system().lower() == "windows", "Unix/Linux/macOS-specific integration test")
    def test_real_lifecycle_comprehensive(self):
        """
        Comprehensive real-life integration test covering all lifecycle scenarios
        This combines install/enable/disable/uninstall, error handling, and crontab scenarios
        WARNING: This test modifies system state and should only run on test systems
        """
        if platform.system().lower() == "windows":
            self.skipTest("Unix/Linux/macOS-specific integration test")

        self._setup_real_cron_test()

        try:
            # ===== PHASE 1: Clean state and error handling =====
            if self.scheduler.is_installed():
                self.scheduler.uninstall()

            # Test operations on non-existent cron job
            self.assertFalse(self.scheduler.enable(), "Enable should fail for non-existent cron job")
            self.assertFalse(self.scheduler.disable(), "Disable should fail for non-existent cron job")
            self.assertFalse(self.scheduler.uninstall(), "Uninstall should fail for non-existent cron job")

            # Verify initial state
            initial_status = self.scheduler.get_status()
            self.assertEqual(initial_status["scheduler"], "cron")
            self.assertFalse(initial_status["installed"], "Cron job should not be installed initially")

            # ===== PHASE 2: Installation and validation =====
            ddns_args = {
                "dns": "debug",
                "ipv4": ["test-comprehensive.example.com"],
                "config": ["config.json"],
                "ttl": 300,
            }

            # Mock crontab commands for installation
            with patch("ddns.util.try_run.try_run") as mock_try_run, patch(
                "ddns.scheduler.cron.subprocess.check_call"
            ) as mock_check_call, patch("ddns.scheduler.cron.tempfile.mktemp", return_value="/tmp/test.cron"), patch(
                "ddns.scheduler.cron.write_file"
            ), patch("ddns.scheduler.cron.os.unlink"):

                def crontab_side_effect(command, **kwargs):
                    if command == ["crontab", "-l"]:
                        return ""  # Return empty crontab initially
                    return None

                mock_try_run.side_effect = crontab_side_effect
                mock_check_call.return_value = None  # Simulate successful crontab update
                install_result = self.scheduler.install(interval=5, ddns_args=ddns_args)
            self.assertTrue(install_result, "Installation should succeed")

            # Verify installation and crontab content - mock the get_status call
            with patch("ddns.scheduler.cron.try_run") as mock_try_run:
                # Mock crontab -l to return the installed entry
                cron_entry = '*/5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = cron_entry
                post_install_status = self.scheduler.get_status()
            self.assertTrue(post_install_status["installed"], "Cron job should be installed")
            self.assertTrue(post_install_status["enabled"], "Cron job should be enabled")
            self.assertEqual(post_install_status["interval"], 5, "Interval should match")

            # Mock crontab content check - use the direct method instead
            with patch("ddns.scheduler.cron.try_run") as mock_try_run:
                cron_entry = '*/5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = cron_entry

                # Use the scheduler's crontab content directly instead of importing try_run
                crontab_content = cron_entry
            self.assertIsNotNone(crontab_content, "Crontab should have content")
            if crontab_content:
                self.assertIn("DDNS", crontab_content, "Crontab should contain DDNS entry")
                self.assertIn("*/5", crontab_content, "Crontab should contain correct interval")

            # Validate cron entry format
            lines = crontab_content.strip().split("\n") if crontab_content else []
            ddns_lines = [line for line in lines if "DDNS" in line and not line.strip().startswith("#")]
            self.assertTrue(len(ddns_lines) > 0, "Should have active DDNS cron entry")

            ddns_line = ddns_lines[0]
            parts = ddns_line.split()
            self.assertTrue(len(parts) >= 5, "Cron line should have at least 5 time fields")
            self.assertEqual(parts[0], "*/5", "Should have correct minute interval")
            self.assertIn("python", ddns_line.lower(), "Should contain python command")
            if crontab_content:
                self.assertIn("debug", crontab_content, "Should contain DNS provider")

            # ===== PHASE 3: Disable/Enable cycle =====
            with patch("ddns.scheduler.cron.try_run") as mock_try_run, patch(
                "ddns.scheduler.cron.subprocess.check_call"
            ) as mock_check_call, patch("ddns.scheduler.cron.tempfile.mktemp", return_value="/tmp/test.cron"), patch(
                "ddns.scheduler.cron.write_file"
            ), patch("ddns.scheduler.cron.os.unlink"):
                # Mock crontab -l to return the installed entry (for disable operation)
                cron_entry = '*/5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = cron_entry
                mock_check_call.return_value = None
                disable_result = self.scheduler.disable()
            self.assertTrue(disable_result, "Disable should succeed")

            # Check status after disable
            with patch("ddns.scheduler.cron.try_run") as mock_try_run:
                # Mock disabled crontab (entry is commented out)
                disabled_entry = '# */5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = disabled_entry
                post_disable_status = self.scheduler.get_status()
            self.assertTrue(post_disable_status["installed"], "Should still be installed after disable")
            self.assertFalse(post_disable_status["enabled"], "Should be disabled")

            # Verify cron entry is commented out - simulate the disabled state
            disabled_crontab = disabled_entry
            if disabled_crontab:
                disabled_lines = [line for line in disabled_crontab.split("\n") if "DDNS" in line]
                self.assertTrue(
                    all(line.strip().startswith("#") for line in disabled_lines),
                    "All DDNS lines should be commented when disabled",
                )

            # Enable operation
            with patch("ddns.scheduler.cron.try_run") as mock_try_run, patch(
                "ddns.scheduler.cron.subprocess.check_call"
            ) as mock_check_call, patch("ddns.scheduler.cron.tempfile.mktemp", return_value="/tmp/test.cron"), patch(
                "ddns.scheduler.cron.write_file"
            ), patch("ddns.scheduler.cron.os.unlink"):
                # Mock crontab -l to return the disabled entry (for enable operation)
                disabled_entry = '# */5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = disabled_entry
                mock_check_call.return_value = None
                enable_result = self.scheduler.enable()
            self.assertTrue(enable_result, "Enable should succeed")

            # Check status after enable
            with patch("ddns.scheduler.cron.try_run") as mock_try_run:
                # Mock enabled crontab (entry is uncommented)
                enabled_entry = '*/5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = enabled_entry
                post_enable_status = self.scheduler.get_status()
            self.assertTrue(post_enable_status["installed"], "Should still be installed after enable")
            self.assertTrue(post_enable_status["enabled"], "Should be enabled")

            # ===== PHASE 4: Duplicate installation test =====
            with patch("ddns.scheduler.cron.try_run") as mock_try_run, patch(
                "ddns.scheduler.cron.subprocess.check_call"
            ) as mock_check_call, patch("ddns.scheduler.cron.tempfile.mktemp", return_value="/tmp/test.cron"), patch(
                "ddns.scheduler.cron.write_file"
            ), patch("ddns.scheduler.cron.os.unlink"):
                enabled_entry = '*/5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = enabled_entry
                mock_check_call.return_value = None
                duplicate_install = self.scheduler.install(interval=5, ddns_args=ddns_args)
            self.assertIsInstance(duplicate_install, bool, "Duplicate install should return boolean")

            with patch("ddns.scheduler.cron.try_run") as mock_try_run:
                enabled_entry = '*/5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = enabled_entry
                status_after_duplicate = self.scheduler.get_status()
            self.assertTrue(status_after_duplicate["installed"], "Should remain installed after duplicate")

            # ===== PHASE 5: Uninstall and verification =====
            with patch("ddns.scheduler.cron.try_run") as mock_try_run, patch(
                "ddns.scheduler.cron.subprocess.check_call"
            ) as mock_check_call, patch("ddns.scheduler.cron.tempfile.mktemp", return_value="/tmp/test.cron"), patch(
                "ddns.scheduler.cron.write_file"
            ), patch("ddns.scheduler.cron.os.unlink"):
                enabled_entry = '*/5 * * * * cd "/workspaces/DDNS" && python3 -m ddns --dns debug --ipv4 test-comprehensive.example.com --config config.json --ttl 300 # DDNS: Auto DDNS Update'
                mock_try_run.return_value = enabled_entry
                mock_check_call.return_value = None
                uninstall_result = self.scheduler.uninstall()
            self.assertTrue(uninstall_result, "Uninstall should succeed")

            # Check final status after uninstall
            with patch("ddns.scheduler.cron.try_run") as mock_try_run:
                mock_try_run.return_value = ""  # Empty crontab
                final_status = self.scheduler.get_status()
                is_installed = self.scheduler.is_installed()
            self.assertFalse(final_status["installed"], "Should not be installed after uninstall")
            self.assertFalse(is_installed, "is_installed() should return False")

            # Verify complete removal from crontab
            from ddns.util.try_run import try_run

            final_crontab = try_run(["crontab", "-l"])
            if final_crontab:
                self.assertNotIn("DDNS", final_crontab, "DDNS should be completely removed")
        finally:
            self._cleanup_real_cron_test()


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