Adrian A. Baumann
bff887428e
- Test hash computation and consistency - Test cache path generation - Test cache miss (generates diagram) - Test cache hit (uses cached diagram) - Test error handling - Test cache clearing (all and by type) - Test unicode handling - Test timeout configuration - Integration tests for full lifecycle
268 lines
10 KiB
Python
268 lines
10 KiB
Python
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'<svg>test diagram</svg>'
|
|
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'<svg>cached diagram</svg>'))
|
|
|
|
# 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'<svg>diagram</svg>'
|
|
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'<svg>test</svg>'
|
|
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'<svg>lifecycle test</svg>'
|
|
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))
|