dify-1.8.1/api/tests/unit_tests/controllers/console/app/test_p0_statistics.py

"""
Unit tests for P0 Statistics API endpoints:
- Response Time Percentiles (P99/P95/P50)
- Token Usage Detail (Input/Output breakdown)
- Error Rate

These tests verify the SQL query logic and data transformation
rather than testing the full HTTP endpoint with authentication.
"""

from datetime import datetime

import pytest


class TestResponseTimePercentilesStatistic:
    """Test suite for Response Time Percentiles SQL and data transformation logic"""

    def test_percentile_calculation_logic(self):
        """Test that percentile values are correctly converted from seconds to milliseconds"""
        # Simulate raw database values (in seconds)
        p50_seconds = 0.123
        p95_seconds = 0.456
        p99_seconds = 0.789
        avg_seconds = 0.234

        # Apply the transformation logic from the API
        p50_ms = round(p50_seconds * 1000, 4)
        p95_ms = round(p95_seconds * 1000, 4)
        p99_ms = round(p99_seconds * 1000, 4)
        avg_ms = round(avg_seconds * 1000, 4)

        # Verify conversion
        assert p50_ms == 123.0
        assert p95_ms == 456.0
        assert p99_ms == 789.0
        assert avg_ms == 234.0

    def test_null_percentile_handling(self):
        """Test handling of null percentile values"""
        p50_seconds = None

        # Apply null handling logic from API
        p50_ms = round(p50_seconds * 1000, 4) if p50_seconds else 0

        assert p50_ms == 0

    def test_sql_query_contains_percentile_functions(self):
        """Verify the SQL query includes correct percentile functions"""
        from controllers.console.app.statistic import ResponseTimePercentilesStatistic

        # Get the source code of the get method
        import inspect

        source = inspect.getsource(ResponseTimePercentilesStatistic.get)

        # Verify SQL contains required percentile functions
        assert "PERCENTILE_CONT(0.50)" in source
        assert "PERCENTILE_CONT(0.95)" in source
        assert "PERCENTILE_CONT(0.99)" in source
        assert "provider_response_latency" in source


class TestTokenUsageDetailStatistic:
    """Test suite for Token Usage Detail SQL and data transformation logic"""

    def test_token_breakdown_logic(self):
        """Test that tokens are correctly split into input and output"""
        # Simulate database values
        input_tokens = 1000
        output_tokens = 2000
        total_tokens = 3000

        # Verify the logic
        assert input_tokens + output_tokens == total_tokens
        assert input_tokens == 1000
        assert output_tokens == 2000

    def test_null_token_handling(self):
        """Test handling of null token values"""
        input_tokens = None
        output_tokens = None

        # Apply null handling logic from API
        input_result = input_tokens or 0
        output_result = output_tokens or 0

        assert input_result == 0
        assert output_result == 0

    def test_sql_query_separates_input_output_tokens(self):
        """Verify the SQL query correctly separates input and output tokens"""
        from controllers.console.app.statistic import TokenUsageDetailStatistic

        import inspect

        source = inspect.getsource(TokenUsageDetailStatistic.get)

        # Verify SQL separates input and output tokens
        assert "message_tokens" in source  # input tokens
        assert "answer_tokens" in source  # output tokens
        assert "SUM" in source  # aggregation function


class TestErrorRateStatistic:
    """Test suite for Error Rate calculation logic"""

    def test_error_rate_calculation(self):
        """Test error rate calculation"""
        total_count = 100
        error_count = 5

        # Apply calculation logic from API
        error_rate = (error_count * 100 / total_count) if total_count > 0 else 0

        assert error_rate == 5.0  # 5%

    def test_zero_requests_error_rate(self):
        """Test error rate calculation when total count is zero"""
        total_count = 0
        error_count = 0

        # Should not raise division by zero
        error_rate = (error_count * 100 / total_count) if total_count > 0 else 0

        assert error_rate == 0.0

    def test_100_percent_error_rate(self):
        """Test error rate when all requests failed"""
        total_count = 50
        error_count = 50

        error_rate = (error_count * 100 / total_count) if total_count > 0 else 0

        assert error_rate == 100.0

    def test_partial_error_rate(self):
        """Test various partial error rates"""
        test_cases = [
            (1000, 1, 0.1),  # 0.1%
            (1000, 10, 1.0),  # 1%
            (1000, 250, 25.0),  # 25%
            (100, 33, 33.0),  # 33%
        ]

        for total, errors, expected in test_cases:
            error_rate = round((errors * 100 / total) if total > 0 else 0, 2)
            assert error_rate == expected

    def test_sql_query_identifies_errors(self):
        """Verify the SQL query correctly identifies error status"""
        from controllers.console.app.statistic import ErrorRateStatistic

        import inspect

        source = inspect.getsource(ErrorRateStatistic.get)

        # Verify SQL checks for error status
        assert "status" in source
        assert "!=" in source or "<>" in source  # SQL not equal operator
        assert "normal" in source  # status != 'normal' means error