DDNS/tests/test_scheduler_systemd.py

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

import os
import platform

from __init__ import patch, unittest

from ddns.scheduler.systemd import SystemdScheduler
from ddns.util.try_run import try_run


class TestSystemdScheduler(unittest.TestCase):
    """Test cases for SystemdScheduler class"""

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

    def test_service_name_property(self):
        """Test service name constant"""
        self.assertEqual(self.scheduler.SERVICE_NAME, "ddns.service")

    def test_timer_name_property(self):
        """Test timer name constant"""
        self.assertEqual(self.scheduler.TIMER_NAME, "ddns.timer")

    @patch("os.path.exists")
    def test_is_installed_true(self, mock_exists):
        """Test is_installed returns True when service exists"""
        mock_exists.return_value = True
        result = self.scheduler.is_installed()
        self.assertTrue(result)

    @patch("os.path.exists")
    def test_is_installed_false(self, mock_exists):
        """Test is_installed returns False when service doesn't exist"""
        mock_exists.return_value = False
        result = self.scheduler.is_installed()
        self.assertFalse(result)

    @patch("subprocess.check_output")
    @patch("ddns.scheduler.systemd.read_file_safely")
    @patch("os.path.exists")
    def test_get_status_success(self, mock_exists, mock_read_file, mock_check_output):
        """Test get_status with proper file reading"""
        mock_exists.return_value = True
        # Mock read_file_safely to return content for timer file and service file

        def mock_read_side_effect(file_path):
            if "ddns.timer" in file_path:
                return "OnUnitActiveSec=5m\n"
            elif "ddns.service" in file_path:
                return "ExecStart=/usr/bin/python3 -m ddns\n"
            return ""

        mock_read_file.side_effect = mock_read_side_effect
        # Mock subprocess.check_output to return "enabled" status
        mock_check_output.return_value = "enabled"

        status = self.scheduler.get_status()

        self.assertEqual(status["scheduler"], "systemd")
        self.assertTrue(status["installed"])
        self.assertTrue(status["enabled"])
        self.assertEqual(status["interval"], 5)

    @patch("ddns.scheduler.systemd.write_file")
    @patch.object(SystemdScheduler, "_systemctl")
    def test_install_with_sudo_fallback(self, mock_systemctl, mock_write_file):
        """Test install with sudo fallback for permission issues"""
        # Mock successful file writing and systemctl calls
        mock_write_file.return_value = None  # write_file doesn't return anything
        mock_systemctl.side_effect = [True, True, True]  # daemon-reload, enable, start all succeed

        ddns_args = {"dns": "debug", "ipv4": ["test.com"]}
        result = self.scheduler.install(5, ddns_args)
        self.assertTrue(result)

        # Verify that write_file was called twice (service and timer files)
        self.assertEqual(mock_write_file.call_count, 2)
        # Verify systemctl was called 3 times (daemon-reload, enable, start)
        self.assertEqual(mock_systemctl.call_count, 3)

    def test_systemctl_basic_functionality(self):
        """Test systemctl command basic functionality"""
        # Test that systemctl calls try_run and returns appropriate boolean
        with patch("ddns.scheduler.systemd.try_run") as mock_run_cmd:
            # Test success case
            mock_run_cmd.return_value = "success"
            result = self.scheduler._systemctl("enable", "ddns.timer")
            self.assertTrue(result)
            mock_run_cmd.assert_called_with(["systemctl", "enable", "ddns.timer"], logger=self.scheduler.logger)

            # Test failure case
            mock_run_cmd.return_value = None
            result = self.scheduler._systemctl("enable", "ddns.timer")
            self.assertFalse(result)

    @patch("os.remove")
    @patch.object(SystemdScheduler, "_systemctl")
    def test_uninstall_with_permission_handling(self, mock_systemctl, mock_remove):
        """Test uninstall with proper permission handling"""
        mock_systemctl.return_value = True  # disable() succeeds

        # Mock successful file removal
        mock_remove.return_value = None

        result = self.scheduler.uninstall()
        self.assertTrue(result)

        # Verify both service and timer files are removed
        self.assertEqual(mock_remove.call_count, 2)

    def test_build_ddns_command(self):
        """Test _build_ddns_command functionality"""
        ddns_args = {"dns": "debug", "ipv4": ["test.example.com"], "debug": True}
        command = self.scheduler._build_ddns_command(ddns_args)
        self.assertIsInstance(command, list)
        command_str = " ".join(command)
        self.assertIn("debug", command_str)
        self.assertIn("test.example.com", command_str)

    @unittest.skipUnless(platform.system().lower() == "linux", "Linux-specific test")
    def test_real_systemctl_availability(self):
        """Test if systemctl is available on Linux systems"""
        # Check if systemctl command is available
        try:
            from ddns.util.try_run import try_run

            systemctl_result = try_run(["systemctl", "--version"])
            if not systemctl_result:
                self.skipTest("systemctl not available on this system")
        except Exception:
            self.skipTest("systemctl not available on this system")

        # Test both regular and sudo access
        self.scheduler._systemctl("--version")

        # Test if we have sudo access (don't actually run sudo commands in tests)
        try:
            sudo_result = try_run(["sudo", "--version"])
            if sudo_result:
                # Just verify sudo is available for fallback
                self.assertIsNotNone(sudo_result)
        except Exception:
            # sudo not available, skip test
            self.skipTest("sudo not available for elevated permissions")

    @unittest.skipUnless(platform.system().lower() == "linux", "Linux-specific test")
    def test_permission_check_methods(self):
        """Test permission checking for systemd operations"""
        # Test if we can write to systemd directory
        systemd_dir = "/etc/systemd/system"
        can_write = os.access(systemd_dir, os.W_OK) if os.path.exists(systemd_dir) else False

        # If we can't write directly, we should be able to use sudo
        if not can_write:
            try:
                sudo_result = try_run(["sudo", "--version"])
                self.assertIsNotNone(sudo_result, "sudo should be available for elevated permissions")
            except Exception:
                self.skipTest("sudo not available for elevated permissions")

    @unittest.skipUnless(platform.system().lower() == "linux", "Linux-specific test")
    def test_real_systemd_integration(self):
        """Test real systemd integration with actual system calls"""
        # Check if systemctl command is available
        try:
            systemctl_result = try_run(["systemctl", "--version"])
            if not systemctl_result:
                self.skipTest("systemctl not available on this system")
        except Exception:
            self.skipTest("systemctl not available on this system")

        # Test real systemctl version call
        version_result = self.scheduler._systemctl("--version")
        # On a real Linux system with systemd, this should work
        # We don't assert the result since it may vary based on permissions
        self.assertIsInstance(version_result, bool)

        # Test real status check for a non-existent service
        status = self.scheduler.get_status()
        self.assertIsInstance(status, dict)
        self.assertEqual(status["scheduler"], "systemd")
        self.assertIsInstance(status["installed"], bool)

        # Test if daemon-reload works (read-only operation)
        daemon_reload_result = self.scheduler._systemctl("daemon-reload")
        # This might fail due to permissions, but shouldn't crash
        self.assertIsInstance(daemon_reload_result, bool)

    @unittest.skipUnless(platform.system().lower() == "linux", "Linux-specific test")
    def test_real_scheduler_methods_safe(self):
        """Test real scheduler methods that don't modify system state"""
        # Check if systemctl command is available
        try:
            systemctl_result = try_run(["systemctl", "--version"])
            if not systemctl_result:
                self.skipTest("systemctl not available on this system")
        except Exception:
            self.skipTest("systemctl not available on this system")

        # 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()
        # Basic keys should always be present
        basic_required_keys = ["scheduler", "installed"]
        for key in basic_required_keys:
            self.assertIn(key, status)

        # If service is installed, additional keys should be present
        if status.get("installed", False):
            additional_keys = ["enabled", "interval"]
            for key in additional_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)

    @unittest.skipUnless(platform.system().lower() == "linux", "Linux-specific test")
    def test_real_systemd_lifecycle_operations(self):
        """Test real systemd lifecycle operations: install -> enable -> disable -> uninstall"""
        # Check if systemctl command is available
        try:
            systemctl_result = try_run(["systemctl", "--version"])
            if not systemctl_result:
                self.skipTest("systemctl not available on this system")
        except Exception:
            self.skipTest("systemctl not available on this system")

        # Test arguments for DDNS
        ddns_args = {"dns": "debug", "ipv4": ["test.example.com"], "interval": 10}

        # Store original state
        original_installed = self.scheduler.is_installed()
        self.scheduler.get_status() if original_installed else None

        try:
            # Test 1: Install operation
            install_result = self.scheduler.install(10, ddns_args)
            self.assertIsInstance(install_result, bool)

            # After install, service should be installed (regardless of permissions)
            post_install_status = self.scheduler.get_status()
            self.assertIsInstance(post_install_status, dict)
            self.assertEqual(post_install_status["scheduler"], "systemd")
            self.assertIsInstance(post_install_status["installed"], bool)

            # If installation succeeded, test enable/disable
            if install_result and post_install_status.get("installed", False):
                # Test 2: Enable operation
                enable_result = self.scheduler.enable()
                self.assertIsInstance(enable_result, bool)

                # Check status after enable attempt
                post_enable_status = self.scheduler.get_status()
                self.assertIsInstance(post_enable_status, dict)
                self.assertIn("enabled", post_enable_status)

                # Test 3: Disable operation
                disable_result = self.scheduler.disable()
                self.assertIsInstance(disable_result, bool)

                # Check status after disable attempt
                post_disable_status = self.scheduler.get_status()
                self.assertIsInstance(post_disable_status, dict)
                self.assertIn("enabled", post_disable_status)

                # Test 4: Uninstall operation
                uninstall_result = self.scheduler.uninstall()
                self.assertIsInstance(uninstall_result, bool)

                # Check status after uninstall attempt
                post_uninstall_status = self.scheduler.get_status()
                self.assertIsInstance(post_uninstall_status, dict)
                # After uninstall, installed should be False (if uninstall succeeded)
                if uninstall_result:
                    self.assertFalse(post_uninstall_status.get("installed", True))
            else:
                self.skipTest("Install failed due to permissions - cannot test lifecycle")

        except Exception as e:
            # If we get permission errors, that's expected in test environment
            if "Permission denied" in str(e) or "Interactive authentication required" in str(e):
                self.skipTest("Insufficient permissions for systemd operations")
            else:
                # Re-raise unexpected exceptions
                raise

        finally:
            # Cleanup: Try to restore original state
            try:
                if original_installed:
                    # If it was originally installed, try to restore
                    if not self.scheduler.is_installed():
                        # Try to reinstall with original settings if we have them
                        self.scheduler.install(10, ddns_args)
                else:
                    # If it wasn't originally installed, try to uninstall
                    if self.scheduler.is_installed():
                        self.scheduler.uninstall()
            except Exception:
                # Cleanup failures are not critical for tests
                pass

    @unittest.skipUnless(platform.system().lower() == "linux", "Linux-specific test")
    def test_real_systemd_status_consistency(self):
        """Test that systemd status reporting is consistent across operations"""
        # Check if systemctl command is available
        try:
            systemctl_result = try_run(["systemctl", "--version"])
            if not systemctl_result:
                self.skipTest("systemctl not available on this system")
        except Exception:
            self.skipTest("systemctl not available on this system")

        # Get initial status
        initial_status = self.scheduler.get_status()
        self.assertIsInstance(initial_status, dict)
        self.assertEqual(initial_status["scheduler"], "systemd")
        self.assertIn("installed", initial_status)

        # Test is_installed consistency
        installed_check = self.scheduler.is_installed()
        self.assertEqual(installed_check, initial_status["installed"])

        # If installed, check that additional status fields are present
        if initial_status.get("installed", False):
            required_keys = ["enabled", "interval"]
            for key in required_keys:
                self.assertIn(key, initial_status, "Key '{}' should be present when service is installed".format(key))

        # Test that repeated status calls are consistent
        second_status = self.scheduler.get_status()
        self.assertEqual(initial_status["scheduler"], second_status["scheduler"])
        self.assertEqual(initial_status["installed"], second_status["installed"])

        # If both report as installed, other fields should also match
        if initial_status.get("installed", False) and second_status.get("installed", False):
            for key in ["enabled", "interval"]:
                if key in initial_status and key in second_status:
                    self.assertEqual(
                        initial_status[key],
                        second_status[key],
                        "Status field '{}' should be consistent between calls".format(key),
                    )


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