mirror of
https://github.com/NewFuture/DDNS.git
synced 2026-10-19 16:15:45 +00:00
445 lines
22 KiB
Python
445 lines
22 KiB
Python
# -*- 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()
|