diff --git a/diagramm_proxy/test_diagram_cache.py b/diagramm_proxy/test_diagram_cache.py
new file mode 100644
index 0000000..eed7f42
--- /dev/null
+++ b/diagramm_proxy/test_diagram_cache.py
@@ -0,0 +1,267 @@
+import os
+import hashlib
+import tempfile
+from unittest.mock import patch, Mock, MagicMock
+from django.test import TestCase, override_settings
+from django.core.files.storage import default_storage
+from django.core.files.base import ContentFile
+from diagramm_proxy.diagram_cache import (
+ get_cache_path,
+ compute_hash,
+ get_cached_diagram,
+ clear_cache,
+ KROKI_UPSTREAM
+)
+
+
+class DiagramCacheTestCase(TestCase):
+ """Test cases for diagram caching functionality."""
+
+ def setUp(self):
+ """Set up test fixtures."""
+ self.test_diagram_content = """
+ @startuml
+ Alice -> Bob: Hello
+ Bob -> Alice: Hi
+ @enduml
+ """
+ self.diagram_type = "plantuml"
+ self.temp_media_root = tempfile.mkdtemp()
+
+ def tearDown(self):
+ """Clean up after tests."""
+ # Clean up temporary files
+ import shutil
+ if os.path.exists(self.temp_media_root):
+ shutil.rmtree(self.temp_media_root)
+
+ def test_compute_hash(self):
+ """Test that hash computation is consistent."""
+ content1 = "test content"
+ content2 = "test content"
+ content3 = "different content"
+
+ hash1 = compute_hash(content1)
+ hash2 = compute_hash(content2)
+ hash3 = compute_hash(content3)
+
+ # Same content should produce same hash
+ self.assertEqual(hash1, hash2)
+
+ # Different content should produce different hash
+ self.assertNotEqual(hash1, hash3)
+
+ # Hash should be valid SHA256 (64 hex characters)
+ self.assertEqual(len(hash1), 64)
+ self.assertTrue(all(c in '0123456789abcdef' for c in hash1))
+
+ def test_get_cache_path(self):
+ """Test cache path generation."""
+ content_hash = "abc123def456"
+ diagram_type = "mermaid"
+
+ path = get_cache_path(diagram_type, content_hash)
+
+ # Path should contain diagram type and hash
+ self.assertIn("diagram_cache", path)
+ self.assertIn(diagram_type, path)
+ self.assertIn(content_hash, path)
+ self.assertTrue(path.endswith(".svg"))
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ @patch('diagramm_proxy.diagram_cache.requests.post')
+ def test_get_cached_diagram_cache_miss(self, mock_post):
+ """Test diagram generation on cache miss."""
+ # Mock successful Kroki response
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.content = b''
+ mock_response.raise_for_status = Mock()
+ mock_post.return_value = mock_response
+
+ # Call function
+ result = get_cached_diagram(self.diagram_type, self.test_diagram_content)
+
+ # Verify POST was called with correct parameters
+ expected_url = f"{KROKI_UPSTREAM}/{self.diagram_type}/svg"
+ mock_post.assert_called_once()
+ call_args = mock_post.call_args
+ self.assertEqual(call_args[0][0], expected_url)
+ self.assertEqual(call_args[1]['data'], self.test_diagram_content.encode('utf-8'))
+ self.assertEqual(call_args[1]['headers']['Content-Type'], 'text/plain')
+
+ # Verify result is a valid path
+ self.assertIsNotNone(result)
+ self.assertIn('diagram_cache', result)
+
+ # Verify file was cached
+ self.assertTrue(default_storage.exists(result))
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ def test_get_cached_diagram_cache_hit(self):
+ """Test that cached diagrams are retrieved without HTTP call."""
+ # Pre-populate cache
+ content_hash = compute_hash(self.test_diagram_content)
+ cache_path = get_cache_path(self.diagram_type, content_hash)
+
+ # Create cache directory and file
+ full_path = default_storage.path(cache_path)
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+ default_storage.save(cache_path, ContentFile(b''))
+
+ # Mock requests.post to ensure it's NOT called
+ with patch('diagramm_proxy.diagram_cache.requests.post') as mock_post:
+ result = get_cached_diagram(self.diagram_type, self.test_diagram_content)
+
+ # Verify POST was NOT called (cache hit)
+ mock_post.assert_not_called()
+
+ # Verify correct path returned
+ self.assertEqual(result, cache_path)
+
+ # Verify file exists
+ self.assertTrue(default_storage.exists(result))
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ @patch('diagramm_proxy.diagram_cache.requests.post')
+ def test_get_cached_diagram_http_error(self, mock_post):
+ """Test error handling when Kroki server fails."""
+ # Mock failed HTTP response
+ mock_post.side_effect = Exception("Connection failed")
+
+ # Should raise exception
+ with self.assertRaises(Exception):
+ get_cached_diagram(self.diagram_type, self.test_diagram_content)
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ def test_clear_cache_all(self):
+ """Test clearing all cached diagrams."""
+ # Create some test cache files
+ test_files = [
+ ('diagram_cache/plantuml/hash1.svg', b'diagram1'),
+ ('diagram_cache/plantuml/hash2.svg', b'diagram2'),
+ ('diagram_cache/mermaid/hash3.svg', b'diagram3'),
+ ]
+
+ for path, content in test_files:
+ full_path = default_storage.path(path)
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+ default_storage.save(path, ContentFile(content))
+
+ # Verify files exist
+ for path, _ in test_files:
+ self.assertTrue(default_storage.exists(path))
+
+ # Clear all cache
+ clear_cache()
+
+ # Verify files are deleted
+ for path, _ in test_files:
+ self.assertFalse(default_storage.exists(path))
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ def test_clear_cache_by_type(self):
+ """Test clearing cache for specific diagram type."""
+ # Create test cache files
+ plantuml_file = 'diagram_cache/plantuml/hash1.svg'
+ mermaid_file = 'diagram_cache/mermaid/hash2.svg'
+
+ for path in [plantuml_file, mermaid_file]:
+ full_path = default_storage.path(path)
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+ default_storage.save(path, ContentFile(b'test'))
+
+ # Clear only plantuml cache
+ clear_cache('plantuml')
+
+ # Verify plantuml file is deleted but mermaid remains
+ self.assertFalse(default_storage.exists(plantuml_file))
+ self.assertTrue(default_storage.exists(mermaid_file))
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ @patch('diagramm_proxy.diagram_cache.requests.post')
+ def test_same_content_different_type(self, mock_post):
+ """Test that same content but different type creates different cache entries."""
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.content = b''
+ mock_response.raise_for_status = Mock()
+ mock_post.return_value = mock_response
+
+ # Generate for two different types
+ path1 = get_cached_diagram('plantuml', self.test_diagram_content)
+ path2 = get_cached_diagram('mermaid', self.test_diagram_content)
+
+ # Paths should be different
+ self.assertNotEqual(path1, path2)
+ self.assertIn('plantuml', path1)
+ self.assertIn('mermaid', path2)
+
+ # Both should exist
+ self.assertTrue(default_storage.exists(path1))
+ self.assertTrue(default_storage.exists(path2))
+
+ def test_compute_hash_unicode(self):
+ """Test hash computation with unicode characters."""
+ content_with_unicode = "Test with émojis 🎨 and ümlauts"
+
+ # Should not raise exception
+ result = compute_hash(content_with_unicode)
+
+ # Should produce valid hash
+ self.assertEqual(len(result), 64)
+ self.assertTrue(all(c in '0123456789abcdef' for c in result))
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ @patch('diagramm_proxy.diagram_cache.requests.post')
+ def test_timeout_configuration(self, mock_post):
+ """Test that POST request includes timeout."""
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.content = b''
+ mock_response.raise_for_status = Mock()
+ mock_post.return_value = mock_response
+
+ get_cached_diagram(self.diagram_type, self.test_diagram_content)
+
+ # Verify timeout parameter was passed
+ call_kwargs = mock_post.call_args[1]
+ self.assertIn('timeout', call_kwargs)
+ self.assertEqual(call_kwargs['timeout'], 30)
+
+
+class DiagramCacheIntegrationTestCase(TestCase):
+ """Integration tests for diagram caching with file system."""
+
+ @override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+ @patch('diagramm_proxy.diagram_cache.requests.post')
+ def test_full_cache_lifecycle(self, mock_post):
+ """Test complete lifecycle: cache miss, cache hit, clear."""
+ diagram_content = "@startuml\nA -> B\n@enduml"
+ diagram_type = "plantuml"
+
+ # Mock Kroki response
+ mock_response = Mock()
+ mock_response.status_code = 200
+ mock_response.content = b''
+ mock_response.raise_for_status = Mock()
+ mock_post.return_value = mock_response
+
+ # First call - cache miss
+ path1 = get_cached_diagram(diagram_type, diagram_content)
+ self.assertEqual(mock_post.call_count, 1)
+ self.assertTrue(default_storage.exists(path1))
+
+ # Second call - cache hit
+ path2 = get_cached_diagram(diagram_type, diagram_content)
+ self.assertEqual(mock_post.call_count, 1) # Still 1, not called again
+ self.assertEqual(path1, path2)
+
+ # Clear cache
+ clear_cache()
+ self.assertFalse(default_storage.exists(path1))
+
+ # Third call - cache miss again
+ path3 = get_cached_diagram(diagram_type, diagram_content)
+ self.assertEqual(mock_post.call_count, 2) # Called again
+ self.assertTrue(default_storage.exists(path3))