mirror of
https://git.linux-kernel.at/oliver/ivatar.git
synced 2025-11-11 18:56:23 +00:00
355 lines
13 KiB
Python
355 lines
13 KiB
Python
"""
|
|
Tests to verify the application works properly without OpenTelemetry packages installed.
|
|
|
|
This test simulates the ImportError scenario by mocking the import failure
|
|
and ensures all functionality continues to work normally.
|
|
"""
|
|
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from io import BytesIO
|
|
|
|
from django.test import TestCase, RequestFactory, Client
|
|
from django.contrib.auth.models import User
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.urls import reverse
|
|
from PIL import Image
|
|
|
|
|
|
class NoOpenTelemetryTestCase(TestCase):
|
|
"""Test application functionality when OpenTelemetry is not available"""
|
|
|
|
def setUp(self):
|
|
self.factory = RequestFactory()
|
|
self.client = Client()
|
|
|
|
# Create test user
|
|
self.user = User.objects.create_user(
|
|
username="testuser", email="test@example.com", password="testpass123"
|
|
)
|
|
|
|
# Store original modules for restoration
|
|
self.original_modules = {}
|
|
for module_name in list(sys.modules.keys()):
|
|
if "telemetry" in module_name:
|
|
self.original_modules[module_name] = sys.modules[module_name]
|
|
|
|
def tearDown(self):
|
|
"""Restore original module state to prevent test isolation issues"""
|
|
# Remove any modules that were added during testing
|
|
modules_to_remove = [k for k in sys.modules.keys() if "telemetry" in k]
|
|
for module in modules_to_remove:
|
|
if module in sys.modules:
|
|
del sys.modules[module]
|
|
|
|
# Restore original modules
|
|
for module_name, module in self.original_modules.items():
|
|
sys.modules[module_name] = module
|
|
|
|
# Force reload of telemetry_utils to restore proper state
|
|
if "ivatar.telemetry_utils" in self.original_modules:
|
|
import importlib
|
|
|
|
importlib.reload(self.original_modules["ivatar.telemetry_utils"])
|
|
|
|
def _mock_import_error(self, name, *args, **kwargs):
|
|
"""Mock function to simulate ImportError for OpenTelemetry packages"""
|
|
if "opentelemetry" in name:
|
|
raise ImportError(f"No module named '{name}'")
|
|
return self._original_import(name, *args, **kwargs)
|
|
|
|
def _create_test_image(self, format="PNG", size=(100, 100)):
|
|
"""Create a test image for upload testing"""
|
|
image = Image.new("RGB", size, color="red")
|
|
image_io = BytesIO()
|
|
image.save(image_io, format=format)
|
|
image_io.seek(0)
|
|
return image_io
|
|
|
|
def test_telemetry_utils_without_opentelemetry(self):
|
|
"""Test that telemetry_utils works when OpenTelemetry is not installed"""
|
|
# Create a mock module that simulates ImportError for OpenTelemetry
|
|
original_import = __builtins__["__import__"]
|
|
|
|
def mock_import(name, *args, **kwargs):
|
|
if "opentelemetry" in name:
|
|
raise ImportError(f"No module named '{name}'")
|
|
return original_import(name, *args, **kwargs)
|
|
|
|
# Patch the import and force module reload
|
|
with patch("builtins.__import__", side_effect=mock_import):
|
|
# Remove from cache to force reimport
|
|
modules_to_remove = [k for k in sys.modules.keys() if "telemetry" in k]
|
|
for module in modules_to_remove:
|
|
if module in sys.modules:
|
|
del sys.modules[module]
|
|
|
|
# Import should trigger the ImportError path
|
|
import importlib
|
|
import ivatar.telemetry_utils
|
|
|
|
importlib.reload(ivatar.telemetry_utils)
|
|
|
|
from ivatar.telemetry_utils import (
|
|
get_telemetry_decorators,
|
|
get_telemetry_metrics,
|
|
is_telemetry_available,
|
|
)
|
|
|
|
# Should indicate telemetry is not available
|
|
self.assertFalse(is_telemetry_available())
|
|
|
|
# Should get no-op decorators
|
|
trace_avatar, trace_file, trace_auth = get_telemetry_decorators()
|
|
|
|
# Test decorators work as no-op
|
|
@trace_avatar("test")
|
|
def test_func():
|
|
return "success"
|
|
|
|
self.assertEqual(test_func(), "success")
|
|
|
|
# Should get no-op metrics
|
|
metrics = get_telemetry_metrics()
|
|
|
|
# These should not raise exceptions
|
|
metrics.record_avatar_generated(size="80", format_type="png", source="test")
|
|
metrics.record_cache_hit(size="80", format_type="png")
|
|
metrics.record_external_request("test", 200)
|
|
metrics.record_file_upload(1024, "image/png", True)
|
|
|
|
@patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"opentelemetry": None,
|
|
"opentelemetry.trace": None,
|
|
"opentelemetry.metrics": None,
|
|
},
|
|
)
|
|
def test_views_work_without_opentelemetry(self):
|
|
"""Test that views work when OpenTelemetry is not installed"""
|
|
# Force reimport to trigger ImportError path
|
|
modules_to_reload = [
|
|
"ivatar.telemetry_utils",
|
|
"ivatar.views",
|
|
"ivatar.ivataraccount.views",
|
|
]
|
|
|
|
for module in modules_to_reload:
|
|
if module in sys.modules:
|
|
del sys.modules[module]
|
|
|
|
# Import views - this should work without OpenTelemetry
|
|
from ivatar.views import AvatarImageView
|
|
from ivatar.ivataraccount.views import UploadPhotoView
|
|
|
|
# Create instances - should not raise exceptions
|
|
avatar_view = AvatarImageView()
|
|
upload_view = UploadPhotoView()
|
|
|
|
# Views should have the no-op metrics
|
|
self.assertTrue(hasattr(avatar_view, "__class__"))
|
|
self.assertTrue(hasattr(upload_view, "__class__"))
|
|
|
|
def test_avatar_generation_without_opentelemetry(self):
|
|
"""Test avatar generation works without OpenTelemetry"""
|
|
# Test default avatar generation (should work without telemetry)
|
|
response = self.client.get("/avatar/nonexistent@example.com?d=identicon&s=80")
|
|
|
|
# Should get a redirect or image response, not an error
|
|
self.assertIn(response.status_code, [200, 302, 404])
|
|
|
|
def test_file_upload_without_opentelemetry(self):
|
|
"""Test file upload works without OpenTelemetry"""
|
|
# Login user
|
|
self.client.login(username="testuser", password="testpass123")
|
|
|
|
# Create test image
|
|
test_image = self._create_test_image()
|
|
uploaded_file = SimpleUploadedFile(
|
|
"test.png", test_image.getvalue(), content_type="image/png"
|
|
)
|
|
|
|
# Test upload (should work without telemetry)
|
|
response = self.client.post(
|
|
reverse("upload_photo"), {"photo": uploaded_file}, follow=True
|
|
)
|
|
|
|
# Should not get a server error
|
|
self.assertNotEqual(response.status_code, 500)
|
|
|
|
def test_authentication_without_opentelemetry(self):
|
|
"""Test authentication works without OpenTelemetry"""
|
|
# Test login page access
|
|
response = self.client.get(reverse("login"))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# Test login submission
|
|
response = self.client.post(
|
|
reverse("login"), {"username": "testuser", "password": "testpass123"}
|
|
)
|
|
|
|
# Should not get a server error
|
|
self.assertNotEqual(response.status_code, 500)
|
|
|
|
def test_user_registration_without_opentelemetry(self):
|
|
"""Test user registration works without OpenTelemetry"""
|
|
# Check if the 'new' URL exists, skip if not
|
|
try:
|
|
url = reverse("new")
|
|
except Exception:
|
|
# URL doesn't exist, skip this test
|
|
self.skipTest("User registration URL 'new' not found")
|
|
|
|
response = self.client.post(
|
|
url,
|
|
{
|
|
"username": "newuser",
|
|
"password1": "newpass123",
|
|
"password2": "newpass123",
|
|
},
|
|
)
|
|
|
|
# Should not get a server error
|
|
self.assertNotEqual(response.status_code, 500)
|
|
|
|
@patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"opentelemetry": None,
|
|
"opentelemetry.trace": None,
|
|
"opentelemetry.metrics": None,
|
|
},
|
|
)
|
|
def test_decorated_functions_work_without_opentelemetry(self):
|
|
"""Test that decorated functions work when OpenTelemetry is not available"""
|
|
# Force reimport to get no-op decorators
|
|
if "ivatar.telemetry_utils" in sys.modules:
|
|
del sys.modules["ivatar.telemetry_utils"]
|
|
|
|
from ivatar.telemetry_utils import (
|
|
trace_avatar_operation,
|
|
trace_file_upload,
|
|
trace_authentication,
|
|
)
|
|
|
|
# Test each decorator type
|
|
@trace_avatar_operation("test_avatar")
|
|
def avatar_function():
|
|
return "avatar_success"
|
|
|
|
@trace_file_upload("test_upload")
|
|
def upload_function():
|
|
return "upload_success"
|
|
|
|
@trace_authentication("test_auth")
|
|
def auth_function():
|
|
return "auth_success"
|
|
|
|
# All should work normally
|
|
self.assertEqual(avatar_function(), "avatar_success")
|
|
self.assertEqual(upload_function(), "upload_success")
|
|
self.assertEqual(auth_function(), "auth_success")
|
|
|
|
def test_metrics_recording_without_opentelemetry(self):
|
|
"""Test that metrics recording works (as no-op) without OpenTelemetry"""
|
|
# Test the no-op metrics class directly
|
|
from ivatar.telemetry_utils import NoOpMetrics
|
|
|
|
metrics = NoOpMetrics()
|
|
|
|
# These should all work without exceptions
|
|
metrics.record_avatar_generated(size="80", format_type="png", source="test")
|
|
metrics.record_avatar_request(size="80", format_type="png")
|
|
metrics.record_cache_hit(size="80", format_type="png")
|
|
metrics.record_cache_miss(size="80", format_type="png")
|
|
metrics.record_external_request("gravatar", 200)
|
|
metrics.record_file_upload(1024, "image/png", True)
|
|
|
|
# Verify it's the no-op implementation
|
|
self.assertEqual(metrics.__class__.__name__, "NoOpMetrics")
|
|
|
|
def test_application_startup_without_opentelemetry(self):
|
|
"""Test that Django can start without OpenTelemetry packages"""
|
|
# This test verifies that the settings.py OpenTelemetry setup
|
|
# handles ImportError gracefully
|
|
|
|
# The fact that this test runs means Django started successfully
|
|
# even if OpenTelemetry packages were missing during import
|
|
|
|
from django.conf import settings
|
|
|
|
# Django should be configured
|
|
self.assertTrue(settings.configured)
|
|
|
|
# Middleware should be loaded (even if OpenTelemetry middleware failed to load)
|
|
self.assertIsInstance(settings.MIDDLEWARE, list)
|
|
|
|
def test_views_import_safely_without_opentelemetry(self):
|
|
"""Test that all views can be imported without OpenTelemetry"""
|
|
# These imports should not raise ImportError even without OpenTelemetry
|
|
try:
|
|
from ivatar import views
|
|
from ivatar.ivataraccount import views as account_views
|
|
|
|
# Should be able to access the views
|
|
self.assertTrue(hasattr(views, "AvatarImageView"))
|
|
self.assertTrue(hasattr(account_views, "UploadPhotoView"))
|
|
|
|
except ImportError as e:
|
|
self.fail(f"Views failed to import without OpenTelemetry: {e}")
|
|
|
|
def test_middleware_handles_missing_opentelemetry(self):
|
|
"""Test that middleware handles missing OpenTelemetry gracefully"""
|
|
# Test a simple request to ensure middleware doesn't break
|
|
response = self.client.get("/")
|
|
|
|
# Should get some response, not a server error
|
|
self.assertNotEqual(response.status_code, 500)
|
|
|
|
|
|
class OpenTelemetryFallbackIntegrationTest(TestCase):
|
|
"""Integration tests for OpenTelemetry fallback behavior"""
|
|
|
|
def setUp(self):
|
|
self.client = Client()
|
|
|
|
def test_full_avatar_workflow_without_opentelemetry(self):
|
|
"""Test complete avatar workflow works without OpenTelemetry"""
|
|
# Test various avatar generation methods
|
|
test_cases = [
|
|
"/avatar/test@example.com?d=identicon&s=80",
|
|
"/avatar/test@example.com?d=monsterid&s=80",
|
|
"/avatar/test@example.com?d=robohash&s=80",
|
|
"/avatar/test@example.com?d=retro&s=80",
|
|
"/avatar/test@example.com?d=pagan&s=80",
|
|
"/avatar/test@example.com?d=mm&s=80",
|
|
]
|
|
|
|
for url in test_cases:
|
|
with self.subTest(url=url):
|
|
response = self.client.get(url)
|
|
# Should not get server error
|
|
self.assertNotEqual(
|
|
response.status_code, 500, f"Server error for {url}"
|
|
)
|
|
|
|
def test_stats_endpoint_without_opentelemetry(self):
|
|
"""Test stats endpoint works without OpenTelemetry"""
|
|
response = self.client.get("/stats/")
|
|
|
|
# Should not get server error
|
|
self.assertNotEqual(response.status_code, 500)
|
|
|
|
def test_version_endpoint_without_opentelemetry(self):
|
|
"""Test version endpoint works without OpenTelemetry"""
|
|
response = self.client.get("/version/")
|
|
|
|
# Should not get server error
|
|
self.assertNotEqual(response.status_code, 500)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|