# -*- coding: utf-8 -*- """ Test cases for cache module @author: GitHub Copilot """ from __init__ import patch, unittest import os import tempfile from time import sleep from ddns.cache import Cache # noqa: E402 class TestCache(unittest.TestCase): """Test cases for Cache class""" def setUp(self): """Set up test fixtures""" # Create a temporary directory for test cache files self.cache_file = tempfile.mktemp(prefix="ddns_test_cache_", suffix="pk1") # self.cache_file = os.path.join(self.test_dir, "test_cache.%d.pkl" % random.randint(1, 10000)) def tearDown(self): """Clean up test fixtures""" # Remove the temporary directory and all its contents if os.path.exists(self.cache_file): os.remove(self.cache_file) def test_init_new_cache(self): """Test cache initialization with new cache file""" cache = Cache(self.cache_file) # Verify initialization self.assertEqual(len(cache), 0) self.assertIsInstance(cache.time, float) self.assertFalse(os.path.exists(self.cache_file)) # File not created until sync def test_init_with_logger(self): """Test cache initialization with custom logger""" import logging logger = logging.getLogger("test_logger") cache = Cache(self.cache_file, logger=logger) self.assertEqual(len(cache), 0) def test_init_with_sync(self): """Test cache initialization with sync enabled""" cache = Cache(self.cache_file, sync=True) self.assertEqual(len(cache), 0) def test_setitem_and_getitem(self): """Test setting and getting cache items""" cache = Cache(self.cache_file) # Test setting items cache["key1"] = "value1" cache["key2"] = "value2" # Test getting items self.assertEqual(cache["key1"], "value1") self.assertEqual(cache["key2"], "value2") self.assertEqual(len(cache), 2) def test_setitem_duplicate_value(self): """Test setting the same value twice doesn't trigger update""" cache = Cache(self.cache_file) with patch.object(cache, "_Cache__update") as mock_update: cache["key1"] = "value1" mock_update.assert_called_once() # Setting the same value should not trigger update mock_update.reset_mock() cache["key1"] = "value1" mock_update.assert_not_called() def test_delitem(self): """Test deleting cache items""" cache = Cache(self.cache_file) cache["key1"] = "value1" cache["key2"] = "value2" # Delete an item del cache["key1"] self.assertEqual(len(cache), 1) self.assertNotIn("key1", cache) self.assertIn("key2", cache) def test_delitem_nonexistent_key(self): """Test deleting non-existent key doesn't raise error (silent handling)""" cache = Cache(self.cache_file) cache["key1"] = "value1" # Should not raise any exception del cache["nonexistent"] # Original data should remain unchanged self.assertEqual(len(cache), 1) self.assertIn("key1", cache) def test_delitem_idempotent(self): """Test that multiple deletions of the same key are safe""" cache = Cache(self.cache_file) cache["key1"] = "value1" # First deletion should work del cache["key1"] self.assertEqual(len(cache), 0) self.assertNotIn("key1", cache) # Second deletion should be safe (no error) del cache["key1"] self.assertEqual(len(cache), 0) # Third deletion should also be safe del cache["key1"] self.assertEqual(len(cache), 0) def test_contains(self): """Test membership testing""" cache = Cache(self.cache_file) cache["key1"] = "value1" self.assertIn("key1", cache) self.assertNotIn("key2", cache) def test_clear(self): """Test clearing cache""" cache = Cache(self.cache_file) cache["key1"] = "value1" cache["key2"] = "value2" cache.clear() self.assertEqual(len(cache), 0) self.assertNotIn("key1", cache) self.assertNotIn("key2", cache) def test_clear_empty_cache(self): """Test clearing empty cache doesn't trigger update""" cache = Cache(self.cache_file) with patch.object(cache, "_Cache__update") as mock_update: cache.clear() mock_update.assert_not_called() def test_clear_preserves_private_fields(self): """Test that clear only removes non-private fields""" cache = Cache(self.cache_file) cache["normal1"] = "value1" cache["normal2"] = "value2" cache["__private"] = "private_value" # Check initial state self.assertEqual(len(cache), 2) # Only counts non-private fields # Clear should only remove non-private fields cache.clear() # Private field should still exist in underlying dict self.assertEqual(len(cache), 0) self.assertNotIn("normal1", cache) self.assertNotIn("normal2", cache) # Private field still exists but not counted/visible self.assertTrue("__private" in dict(cache)) def test_iteration(self): """Test iterating over cache keys""" cache = Cache(self.cache_file) cache["key1"] = "value1" cache["key2"] = "value2" keys = list(cache) self.assertEqual(set(keys), {"key1", "key2"}) def test_private_fields_excluded(self): """Test that private fields (starting with __) are excluded from operations""" cache = Cache(self.cache_file) cache["normal_key"] = "normal_value" # Manually add a private field to the underlying dict (for testing purposes) super(Cache, cache).__setitem__("__private_field", "private_value") # len() should exclude private fields self.assertEqual(len(cache), 1) # iteration should exclude private fields keys = list(cache) self.assertEqual(keys, ["normal_key"]) # data() should exclude private fields data = cache.get(None) self.assertEqual(data, {"normal_key": "normal_value"}) def test_private_field_operations_no_sync(self): """Test that private field operations don't trigger sync""" cache = Cache(self.cache_file) with patch.object(cache, "_Cache__update") as mock_update: # Setting private field should not trigger sync cache["__private"] = "private_value" mock_update.assert_not_called() # Modifying private field should not trigger sync cache["__private"] = "new_private_value" mock_update.assert_not_called() # Deleting private field should not trigger sync del cache["__private"] mock_update.assert_not_called() # Normal field operations should trigger sync cache["normal"] = "value" mock_update.assert_called_once() def test_private_fields_not_saved_to_file(self): """Test that private fields are not saved to file""" cache = Cache(self.cache_file) cache["normal_key"] = "normal_value" cache["__private_key"] = "private_value" # Sync to file cache.sync() # Load new cache instance cache2 = Cache(self.cache_file) # Only normal fields should be loaded self.assertEqual(len(cache2), 1) self.assertIn("normal_key", cache2) self.assertNotIn("__private_key", cache2) self.assertEqual(cache2["normal_key"], "normal_value") def test_data_method(self): """Test data method for getting cache contents""" cache = Cache(self.cache_file) cache["key1"] = "value1" cache["key2"] = "value2" # Test getting all data # data = cache.get() # self.assertEqual(data, {"key1": "value1", "key2": "value2"}) # Test getting specific key self.assertEqual(cache.get("key1"), "value1") self.assertEqual(cache.get("nonexistent", "default"), "default") def test_data_method_with_sync(self): """Test data method with sync enabled calls load""" cache = Cache(self.cache_file, sync=True) with patch.object(cache, "load") as mock_load: cache.load() mock_load.assert_called_once() def test_sync_method(self): """Test sync method saves data to file""" cache = Cache(self.cache_file) cache["key1"] = "value1" cache["key2"] = "value2" # Sync should save to file result = cache.sync() self.assertIs(result, cache) # Should return self self.assertTrue(os.path.exists(self.cache_file)) # Load another cache instance to verify data was saved cache2 = Cache(self.cache_file) self.assertEqual(len(cache2), 2) self.assertEqual(cache2["key1"], "value1") self.assertEqual(cache2["key2"], "value2") def test_sync_no_changes(self): """Test sync when no changes have been made after load""" # Create and save initial cache cache = Cache(self.cache_file) cache["key1"] = "value1" cache.sync() # This clears the __changed flag with patch("ddns.cache.dump") as mock_dump: cache.sync() # This should not call dump since no changes mock_dump.assert_not_called() def test_load_existing_file(self): """Test loading from existing cache file""" # Create initial cache and save data cache1 = Cache(self.cache_file) cache1["key1"] = "value1" cache1["key2"] = 2 cache1.sync() # Load new cache instance cache2 = Cache(self.cache_file) self.assertEqual(len(cache2), 2) self.assertEqual(cache2["key1"], "value1") self.assertEqual(cache2["key2"], 2) def test_load_corrupted_file(self): """Test loading from corrupted cache file""" # Create a corrupted cache file with open(self.cache_file, "w") as f: f.write("corrupted data") # Should handle corruption gracefully cache = Cache(self.cache_file) self.assertEqual(len(cache), 0) def test_load_with_exception(self): """Test load method handles exceptions properly""" # Create a file first with open(self.cache_file, "wb") as f: f.write(b"invalid pickle data") cache = Cache(self.cache_file) with patch("ddns.cache.load", side_effect=Exception("Test error")): with patch.object(cache, "_Cache__logger") as mock_logger: cache.load() mock_logger.warning.assert_called_once() def test_time_property(self): """Test time property returns modification time""" cache = Cache(self.cache_file) initial_time = cache.time # type: float # type: ignore[assignment] self.assertIsInstance(initial_time, float) self.assertGreater(initial_time, 0) def test_close_method(self): """Test close method syncs and cleans up""" cache = Cache(self.cache_file) cache["key1"] = "value1" with patch.object(cache, "sync") as mock_sync: cache.close() mock_sync.assert_called_once() def test_str_representation(self): """Test string representation of cache""" cache = Cache(self.cache_file) cache["key1"] = "value1" str_repr = str(cache) self.assertIn("key1", str_repr) self.assertIn("value1", str_repr) def test_auto_sync_behavior(self): """Test auto-sync behavior when sync=True""" cache = Cache(self.cache_file, sync=True) with patch.object(cache, "sync") as mock_sync: cache["key1"] = "value1" mock_sync.assert_called() def test_no_auto_sync_behavior(self): """Test no auto-sync behavior when sync=False (default)""" cache = Cache(self.cache_file, sync=False) with patch.object(cache, "sync") as mock_sync: cache["key1"] = "value1" mock_sync.assert_not_called() def test_context_manager_like_usage(self): """Test using cache in a context-manager-like pattern""" cache = Cache(self.cache_file) cache["key1"] = "value1" cache["key2"] = "value2" # Manually call close (simulating __del__) cache.close() # Verify data was persisted cache2 = Cache(self.cache_file) self.assertEqual(len(cache2), 2) self.assertEqual(cache2["key1"], "value1") def test_update_time_on_changes(self): """Test that modification time is updated on changes""" cache = Cache(self.cache_file) initial_time = cache.time # Small delay to ensure time difference sleep(0.01) cache["key1"] = "value1" new_time = cache.time # type: float # type: ignore[assignment] self.assertGreater(new_time, initial_time) # type: ignore[comparison-overlap] def test_integration_multiple_operations(self): """Integration test with multiple operations""" cache = Cache(self.cache_file) # Add some data cache["user1"] = {"name": "Alice", "age": 30} cache["user2"] = {"name": "Bob", "age": 25} cache["config"] = {"debug": True, "timeout": 30} self.assertEqual(len(cache), 3) # Modify data cache["user1"]["age"] = 31 # This won't trigger update automatically cache["user1"] = {"name": "Alice", "age": 31} # This will # Delete data del cache["user2"] self.assertEqual(len(cache), 2) self.assertEqual(cache["user1"]["age"], 31) self.assertNotIn("user2", cache) # Persist and reload cache.sync() cache2 = Cache(self.cache_file) self.assertEqual(len(cache2), 2) self.assertEqual(cache2["user1"]["age"], 31) self.assertEqual(cache2["config"]["debug"], True) def test_mixed_public_private_operations(self): """Test mixed operations with public and private fields""" cache = Cache(self.cache_file) # Add mixed data cache["public1"] = "public_value1" cache["__private1"] = "private_value1" cache["public2"] = "public_value2" cache["__private2"] = "private_value2" # Only public fields should be counted self.assertEqual(len(cache), 2) # Only public fields should be iterable public_keys = list(cache) self.assertEqual(set(public_keys), {"public1", "public2"}) # data() should only return public fields data = cache.get(None) self.assertEqual(data, {"public1": "public_value1", "public2": "public_value2"}) # Delete operations del cache["public1"] # Should work del cache["__private1"] # Should work but not trigger sync del cache["nonexistent"] # Should be safe # Only one public field should remain self.assertEqual(len(cache), 1) self.assertEqual(list(cache), ["public2"]) # Sync and reload cache.sync() cache2 = Cache(self.cache_file) # Only public field should be persisted self.assertEqual(len(cache2), 1) self.assertEqual(list(cache2), ["public2"]) self.assertEqual(cache2["public2"], "public_value2") def test_str_representation_excludes_private(self): """Test that string representation only shows public fields""" cache = Cache(self.cache_file) cache["public"] = "public_value" cache["__private"] = "private_value" str_repr = str(cache) self.assertIn("public", str_repr) self.assertIn("public_value", str_repr) # Note: private fields might still appear in str() since it calls super().__str__() # This is acceptable as str() shows the raw dict content def test_json_format_verification(self): """Test that cache files are saved in JSON format""" import json cache = Cache(self.cache_file) cache["string_key"] = "string_value" cache["number_key"] = 42 cache["list_key"] = [1, 2, 3] cache["dict_key"] = {"nested": "value"} # Save to file cache.sync() # Verify file exists and is valid JSON self.assertTrue(os.path.exists(self.cache_file)) # Read the file and verify it's valid JSON with open(self.cache_file, "r") as f: file_content = f.read() parsed_json = json.loads(file_content) # Verify the content matches what we saved self.assertEqual(parsed_json["string_key"], "string_value") self.assertEqual(parsed_json["number_key"], 42) self.assertEqual(parsed_json["list_key"], [1, 2, 3]) self.assertEqual(parsed_json["dict_key"], {"nested": "value"}) # Verify the file contains readable JSON (not binary pickle data) self.assertIn('"string_key"', file_content) self.assertIn('"string_value"', file_content) def test_cache_new_disabled(self): """Test Cache.new with cache disabled (config_cache=False)""" import logging logger = logging.getLogger("test_logger") cache = Cache.new(False, "test_hash", logger) # Should return None when cache is disabled self.assertIsNone(cache) def test_cache_new_temp_file(self): """Test Cache.new with temp file creation (config_cache=True)""" import logging import tempfile logger = logging.getLogger("test_logger") test_hash = "test_hash_123" cache = Cache.new(True, test_hash, logger) # Should create a cache instance self.assertIsNotNone(cache) self.assertIsInstance(cache, Cache) # Should use temp directory with correct naming pattern expected_pattern = "ddns.%s.cache" % test_hash self.assertIn(expected_pattern, cache._Cache__filename) self.assertIn(tempfile.gettempdir(), cache._Cache__filename) # Clean up cache.close() def test_cache_new_custom_path(self): """Test Cache.new with custom cache file path""" import logging logger = logging.getLogger("test_logger") custom_path = self.cache_file cache = Cache.new(custom_path, "unused_hash", logger) # Should create a cache instance with custom path self.assertIsNotNone(cache) self.assertIsInstance(cache, Cache) self.assertEqual(cache._Cache__filename, custom_path) # Clean up cache.close() @patch('ddns.cache.time') def test_cache_new_outdated_cache(self, mock_time): """Test Cache.new with outdated cache file (>72 hours old)""" import logging import json logger = logging.getLogger("test_logger") # Create a cache file with some data test_data = {"test_key": "test_value"} with open(self.cache_file, "w") as f: json.dump(test_data, f) # Mock current time to make cache appear outdated # Cache file mtime will be recent, but we'll mock current time to be 73 hours later current_time = 1000000 mock_time.return_value = current_time # Mock the file modification time to be 73 hours ago old_mtime = current_time - (73 * 3600) # 73 hours ago with patch('ddns.cache.stat') as mock_stat: mock_stat.return_value.st_mtime = old_mtime cache = Cache.new(self.cache_file, "test_hash", logger) # Should create cache instance but clear outdated data self.assertIsNotNone(cache) self.assertIsInstance(cache, Cache) self.assertEqual(len(cache), 0) # Should be empty due to age # Clean up cache.close() def test_cache_new_empty_cache(self): """Test Cache.new with empty cache file""" import logging import json logger = logging.getLogger("test_logger") # Create an empty cache file with open(self.cache_file, "w") as f: json.dump({}, f) cache = Cache.new(self.cache_file, "test_hash", logger) # Should create cache instance self.assertIsNotNone(cache) self.assertIsInstance(cache, Cache) self.assertEqual(len(cache), 0) # Clean up cache.close() @patch('ddns.cache.time') def test_cache_new_valid_cache(self, mock_time): """Test Cache.new with valid cache file with data""" import logging import json logger = logging.getLogger("test_logger") # Create a cache file with test data test_data = { "domain1.com": {"ip": "1.2.3.4", "timestamp": 1234567890}, "domain2.com": {"ip": "5.6.7.8", "timestamp": 1234567891} } with open(self.cache_file, "w") as f: json.dump(test_data, f) # Mock current time to be within 72 hours of cache creation current_time = 1000000 mock_time.return_value = current_time # Mock file modification time to be recent (within 72 hours) recent_mtime = current_time - (24 * 3600) # 24 hours ago with patch('ddns.cache.stat') as mock_stat: mock_stat.return_value.st_mtime = recent_mtime cache = Cache.new(self.cache_file, "test_hash", logger) # Should create cache instance with loaded data self.assertIsNotNone(cache) self.assertIsInstance(cache, Cache) self.assertEqual(len(cache), 2) self.assertEqual(cache["domain1.com"]["ip"], "1.2.3.4") self.assertEqual(cache["domain2.com"]["ip"], "5.6.7.8") # Clean up cache.close() def test_cache_new_nonexistent_file(self): """Test Cache.new with nonexistent cache file""" import logging logger = logging.getLogger("test_logger") nonexistent_path = tempfile.mktemp(prefix="ddns_test_nonexistent_", suffix=".cache") # Ensure file doesn't exist if os.path.exists(nonexistent_path): os.remove(nonexistent_path) cache = Cache.new(nonexistent_path, "test_hash", logger) # Should create cache instance with empty data self.assertIsNotNone(cache) self.assertIsInstance(cache, Cache) self.assertEqual(len(cache), 0) # Clean up cache.close() if __name__ == "__main__": unittest.main()