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'test diagram' 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'cached diagram')) # 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'diagram' 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'test' 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'lifecycle test' 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))