ivatar/ivatar/test_opentelemetry.py

# -*- coding: utf-8 -*-
"""
Tests for OpenTelemetry integration in ivatar.

This module contains comprehensive tests for OpenTelemetry functionality,
including configuration, middleware, metrics, and tracing.
"""

import os
import unittest
from unittest.mock import patch, MagicMock
import pytest
from django.test import TestCase, RequestFactory
from django.http import HttpResponse

from ivatar.opentelemetry_config import (
    OpenTelemetryConfig,
    is_enabled,
)
from ivatar.opentelemetry_middleware import (
    OpenTelemetryMiddleware,
    trace_avatar_operation,
    trace_file_upload,
    trace_authentication,
    AvatarMetrics,
    get_avatar_metrics,
    reset_avatar_metrics,
)


@pytest.mark.opentelemetry
class OpenTelemetryConfigTest(TestCase):
    """Test OpenTelemetry configuration."""

    def setUp(self):
        """Set up test environment."""
        self.original_env = os.environ.copy()

    def tearDown(self):
        """Clean up test environment."""
        os.environ.clear()
        os.environ.update(self.original_env)

    def test_config_disabled_by_default(self):
        """Test that OpenTelemetry is disabled by default."""
        # Clear environment variables to test default behavior
        original_env = os.environ.copy()
        os.environ.pop("ENABLE_OPENTELEMETRY", None)
        os.environ.pop("OTEL_ENABLED", None)

        try:
            config = OpenTelemetryConfig()
            self.assertFalse(config.enabled)
        finally:
            os.environ.clear()
            os.environ.update(original_env)

    def test_config_enabled_with_env_var(self):
        """Test that OpenTelemetry can be enabled with environment variable."""
        os.environ["OTEL_ENABLED"] = "true"
        config = OpenTelemetryConfig()
        self.assertTrue(config.enabled)

    def test_service_name_default(self):
        """Test default service name."""
        # Clear environment variables to test default behavior
        original_env = os.environ.copy()
        os.environ.pop("OTEL_SERVICE_NAME", None)

        try:
            config = OpenTelemetryConfig()
            self.assertEqual(config.service_name, "ivatar")
        finally:
            os.environ.clear()
            os.environ.update(original_env)

    def test_service_name_custom(self):
        """Test custom service name."""
        os.environ["OTEL_SERVICE_NAME"] = "custom-service"
        config = OpenTelemetryConfig()
        self.assertEqual(config.service_name, "custom-service")

    def test_environment_default(self):
        """Test default environment."""
        # Clear environment variables to test default behavior
        original_env = os.environ.copy()
        os.environ.pop("OTEL_ENVIRONMENT", None)

        try:
            config = OpenTelemetryConfig()
            self.assertEqual(config.environment, "development")
        finally:
            os.environ.clear()
            os.environ.update(original_env)

    def test_environment_custom(self):
        """Test custom environment."""
        os.environ["OTEL_ENVIRONMENT"] = "production"
        config = OpenTelemetryConfig()
        self.assertEqual(config.environment, "production")

    def test_resource_creation(self):
        """Test resource creation with service information."""
        os.environ["OTEL_SERVICE_NAME"] = "test-service"
        os.environ["OTEL_ENVIRONMENT"] = "test"
        os.environ["IVATAR_VERSION"] = "1.0.0"
        os.environ["HOSTNAME"] = "test-host"

        config = OpenTelemetryConfig()
        resource = config.resource

        self.assertEqual(resource.attributes["service.name"], "test-service")
        self.assertEqual(resource.attributes["service.version"], "1.0.0")
        self.assertEqual(resource.attributes["deployment.environment"], "test")
        self.assertEqual(resource.attributes["service.instance.id"], "test-host")

    @patch("ivatar.opentelemetry_config.OTLPSpanExporter")
    @patch("ivatar.opentelemetry_config.BatchSpanProcessor")
    @patch("ivatar.opentelemetry_config.trace")
    def test_setup_tracing_with_otlp(self, mock_trace, mock_processor, mock_exporter):
        """Test tracing setup with OTLP endpoint."""
        os.environ["OTEL_ENABLED"] = "true"
        os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:4317"

        config = OpenTelemetryConfig()
        config.setup_tracing()

        mock_exporter.assert_called_once_with(endpoint="http://localhost:4317")
        mock_processor.assert_called_once()
        mock_trace.get_tracer_provider().add_span_processor.assert_called_once()

    @patch("ivatar.opentelemetry_config.PrometheusMetricReader")
    @patch("ivatar.opentelemetry_config.PeriodicExportingMetricReader")
    @patch("ivatar.opentelemetry_config.OTLPMetricExporter")
    @patch("ivatar.opentelemetry_config.metrics")
    def test_setup_metrics_with_prometheus_and_otlp(
        self,
        mock_metrics,
        mock_otlp_exporter,
        mock_periodic_reader,
        mock_prometheus_reader,
    ):
        """Test metrics setup with Prometheus and OTLP."""
        os.environ["OTEL_ENABLED"] = "true"
        os.environ["OTEL_PROMETHEUS_ENDPOINT"] = "0.0.0.0:9464"
        os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:4317"

        config = OpenTelemetryConfig()
        config.setup_metrics()

        mock_prometheus_reader.assert_called_once()
        mock_otlp_exporter.assert_called_once_with(endpoint="http://localhost:4317")
        mock_periodic_reader.assert_called_once()
        mock_metrics.set_meter_provider.assert_called_once()

    @patch("ivatar.opentelemetry_config.DjangoInstrumentor")
    @patch("ivatar.opentelemetry_config.Psycopg2Instrumentor")
    @patch("ivatar.opentelemetry_config.PyMySQLInstrumentor")
    @patch("ivatar.opentelemetry_config.RequestsInstrumentor")
    @patch("ivatar.opentelemetry_config.URLLib3Instrumentor")
    def test_setup_instrumentation(
        self,
        mock_urllib3,
        mock_requests,
        mock_pymysql,
        mock_psycopg2,
        mock_django,
    ):
        """Test instrumentation setup."""
        os.environ["OTEL_ENABLED"] = "true"

        config = OpenTelemetryConfig()
        config.setup_instrumentation()

        mock_django().instrument.assert_called_once()
        mock_psycopg2().instrument.assert_called_once()
        mock_pymysql().instrument.assert_called_once()
        mock_requests().instrument.assert_called_once()
        mock_urllib3().instrument.assert_called_once()


@pytest.mark.opentelemetry
class OpenTelemetryMiddlewareTest(TestCase):
    """Test OpenTelemetry middleware."""

    def setUp(self):
        """Set up test environment."""
        self.factory = RequestFactory()
        reset_avatar_metrics()  # Reset global metrics instance
        self.middleware = OpenTelemetryMiddleware(lambda r: HttpResponse("test"))

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    def test_middleware_disabled(self, mock_enabled):
        """Test middleware when OpenTelemetry is disabled."""
        mock_enabled.return_value = False

        request = self.factory.get("/avatar/test@example.com")
        response = self.middleware(request)

        self.assertEqual(response.status_code, 200)
        self.assertFalse(hasattr(request, "_ot_span"))

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    @patch("ivatar.opentelemetry_middleware.get_tracer")
    def test_middleware_enabled(self, mock_get_tracer, mock_enabled):
        """Test middleware when OpenTelemetry is enabled."""
        mock_enabled.return_value = True
        mock_tracer = MagicMock()
        mock_span = MagicMock()
        mock_tracer.start_span.return_value = mock_span
        mock_get_tracer.return_value = mock_tracer

        request = self.factory.get("/avatar/test@example.com")
        response = self.middleware(request)

        self.assertEqual(response.status_code, 200)
        self.assertTrue(hasattr(request, "_ot_span"))
        mock_tracer.start_span.assert_called_once()
        mock_span.set_attributes.assert_called()
        mock_span.end.assert_called_once()

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    @patch("ivatar.opentelemetry_middleware.get_tracer")
    def test_avatar_request_attributes(self, mock_get_tracer, mock_enabled):
        """Test that avatar requests get proper attributes."""
        mock_enabled.return_value = True
        mock_tracer = MagicMock()
        mock_span = MagicMock()
        mock_tracer.start_span.return_value = mock_span
        mock_get_tracer.return_value = mock_tracer

        request = self.factory.get("/avatar/test@example.com?s=128&d=png")
        # Reset metrics to ensure we get a fresh instance
        reset_avatar_metrics()
        self.middleware.process_request(request)

        # Check that avatar-specific attributes were set
        calls = mock_span.set_attributes.call_args_list
        avatar_attrs = any(
            call[0][0].get("ivatar.request_type") == "avatar" for call in calls
        )
        # Also check for individual set_attribute calls
        set_attribute_calls = mock_span.set_attribute.call_args_list
        individual_avatar_attrs = any(
            call[0][0] == "ivatar.request_type" and call[0][1] == "avatar"
            for call in set_attribute_calls
        )
        self.assertTrue(avatar_attrs or individual_avatar_attrs)

    def test_is_avatar_request(self):
        """Test avatar request detection."""
        avatar_request = self.factory.get("/avatar/test@example.com")
        non_avatar_request = self.factory.get("/stats/")

        self.assertTrue(self.middleware._is_avatar_request(avatar_request))
        self.assertFalse(self.middleware._is_avatar_request(non_avatar_request))

    def test_get_avatar_size(self):
        """Test avatar size extraction."""
        request = self.factory.get("/avatar/test@example.com?s=256")
        size = self.middleware._get_avatar_size(request)
        self.assertEqual(size, "256")

    def test_get_avatar_format(self):
        """Test avatar format extraction."""
        request = self.factory.get("/avatar/test@example.com?d=jpg")
        format_type = self.middleware._get_avatar_format(request)
        self.assertEqual(format_type, "jpg")

    def test_get_avatar_email(self):
        """Test email extraction from avatar request."""
        request = self.factory.get("/avatar/test@example.com")
        email = self.middleware._get_avatar_email(request)
        self.assertEqual(email, "test@example.com")


@pytest.mark.opentelemetry
class AvatarMetricsTest(TestCase):
    """Test AvatarMetrics class."""

    def setUp(self):
        """Set up test environment."""
        self.metrics = AvatarMetrics()

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    def test_metrics_disabled(self, mock_enabled):
        """Test metrics when OpenTelemetry is disabled."""
        mock_enabled.return_value = False

        # Should not raise any exceptions
        self.metrics.record_avatar_generated("128", "png", "generated")
        self.metrics.record_cache_hit("128", "png")
        self.metrics.record_cache_miss("128", "png")
        self.metrics.record_external_request("gravatar", 200)
        self.metrics.record_file_upload(1024, "image/png", True)

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    @patch("ivatar.opentelemetry_middleware.get_meter")
    def test_metrics_enabled(self, mock_get_meter, mock_enabled):
        """Test metrics when OpenTelemetry is enabled."""
        mock_enabled.return_value = True
        mock_meter = MagicMock()
        mock_counter = MagicMock()
        mock_histogram = MagicMock()

        mock_meter.create_counter.return_value = mock_counter
        mock_meter.create_histogram.return_value = mock_histogram
        mock_get_meter.return_value = mock_meter

        avatar_metrics = AvatarMetrics()

        # Test avatar generation recording
        avatar_metrics.record_avatar_generated("128", "png", "generated")
        mock_counter.add.assert_called_with(
            1, {"size": "128", "format": "png", "source": "generated"}
        )

        # Test cache hit recording
        avatar_metrics.record_cache_hit("128", "png")
        mock_counter.add.assert_called_with(1, {"size": "128", "format": "png"})

        # Test file upload recording
        avatar_metrics.record_file_upload(1024, "image/png", True)
        mock_histogram.record.assert_called_with(
            1024, {"content_type": "image/png", "success": "True"}
        )


@pytest.mark.opentelemetry
class TracingDecoratorsTest(TestCase):
    """Test tracing decorators."""

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    @patch("ivatar.opentelemetry_middleware.get_tracer")
    def test_trace_avatar_operation(self, mock_get_tracer, mock_enabled):
        """Test trace_avatar_operation decorator."""
        mock_enabled.return_value = True
        mock_tracer = MagicMock()
        mock_span = MagicMock()
        mock_tracer.start_as_current_span.return_value.__enter__.return_value = (
            mock_span
        )
        mock_get_tracer.return_value = mock_tracer

        @trace_avatar_operation("test_operation")
        def test_function():
            return "success"

        result = test_function()

        self.assertEqual(result, "success")
        mock_tracer.start_as_current_span.assert_called_once_with(
            "avatar.test_operation"
        )
        mock_span.set_status.assert_called_once()

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    @patch("ivatar.opentelemetry_middleware.get_tracer")
    def test_trace_avatar_operation_exception(self, mock_get_tracer, mock_enabled):
        """Test trace_avatar_operation decorator with exception."""
        mock_enabled.return_value = True
        mock_tracer = MagicMock()
        mock_span = MagicMock()
        mock_tracer.start_as_current_span.return_value.__enter__.return_value = (
            mock_span
        )
        mock_get_tracer.return_value = mock_tracer

        @trace_avatar_operation("test_operation")
        def test_function():
            raise ValueError("test error")

        with self.assertRaises(ValueError):
            test_function()

        mock_span.set_status.assert_called_once()
        mock_span.set_attribute.assert_called_with("error.message", "test error")

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    def test_trace_file_upload(self, mock_enabled):
        """Test trace_file_upload decorator."""
        mock_enabled.return_value = True

        @trace_file_upload("test_upload")
        def test_function():
            return "success"

        result = test_function()
        self.assertEqual(result, "success")

    @patch("ivatar.opentelemetry_middleware.is_enabled")
    def test_trace_authentication(self, mock_enabled):
        """Test trace_authentication decorator."""
        mock_enabled.return_value = True

        @trace_authentication("test_auth")
        def test_function():
            return "success"

        result = test_function()
        self.assertEqual(result, "success")


@pytest.mark.opentelemetry
class IntegrationTest(TestCase):
    """Integration tests for OpenTelemetry."""

    def setUp(self):
        """Set up test environment."""
        self.original_env = os.environ.copy()

    def tearDown(self):
        """Clean up test environment."""
        os.environ.clear()
        os.environ.update(self.original_env)

    @patch("ivatar.opentelemetry_config.setup_opentelemetry")
    def test_setup_opentelemetry_called(self, mock_setup):
        """Test that setup_opentelemetry is called during Django startup."""
        # This would be called during Django settings import
        from ivatar.opentelemetry_config import setup_opentelemetry as setup_func

        setup_func()
        mock_setup.assert_called_once()

    def test_is_enabled_function(self):
        """Test is_enabled function."""
        # Clear environment variables to test default behavior
        original_env = os.environ.copy()
        os.environ.pop("ENABLE_OPENTELEMETRY", None)
        os.environ.pop("OTEL_ENABLED", None)

        try:
            # Test disabled by default
            self.assertFalse(is_enabled())
        finally:
            os.environ.clear()
            os.environ.update(original_env)

        # Test enabled with environment variable
        os.environ["OTEL_ENABLED"] = "true"
        config = OpenTelemetryConfig()
        self.assertTrue(config.enabled)


@pytest.mark.no_opentelemetry
class OpenTelemetryDisabledTest(TestCase):
    """Test OpenTelemetry behavior when disabled (no-op mode)."""

    def setUp(self):
        """Set up test environment."""
        self.original_env = os.environ.copy()
        # Ensure OpenTelemetry is disabled
        os.environ.pop("ENABLE_OPENTELEMETRY", None)
        os.environ.pop("OTEL_ENABLED", None)

    def tearDown(self):
        """Clean up test environment."""
        os.environ.clear()
        os.environ.update(self.original_env)

    def test_opentelemetry_disabled_by_default(self):
        """Test that OpenTelemetry is disabled by default."""
        # Clear environment variables to test default behavior
        original_env = os.environ.copy()
        os.environ.pop("ENABLE_OPENTELEMETRY", None)
        os.environ.pop("OTEL_ENABLED", None)

        try:
            self.assertFalse(is_enabled())
        finally:
            os.environ.clear()
            os.environ.update(original_env)

    def test_no_op_decorators_work(self):
        """Test that no-op decorators work when OpenTelemetry is disabled."""

        @trace_avatar_operation("test_operation")
        def test_function():
            return "success"

        result = test_function()
        self.assertEqual(result, "success")

    def test_no_op_metrics_work(self):
        """Test that no-op metrics work when OpenTelemetry is disabled."""
        avatar_metrics = get_avatar_metrics()

        # These should not raise exceptions
        avatar_metrics.record_avatar_generated("80", "png", "uploaded")
        avatar_metrics.record_cache_hit("80", "png")
        avatar_metrics.record_cache_miss("80", "png")
        avatar_metrics.record_external_request("gravatar", "success")
        avatar_metrics.record_file_upload("success", "image/png", True)

    def test_middleware_disabled(self):
        """Test that middleware works when OpenTelemetry is disabled."""
        factory = RequestFactory()
        middleware = OpenTelemetryMiddleware(lambda r: HttpResponse("test"))

        request = factory.get("/avatar/test@example.com")
        response = middleware(request)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.content.decode(), "test")


if __name__ == "__main__":
    unittest.main()