Back to Repositories

Validating Analytics System Integration in Gradio

This test suite validates the analytics functionality in the Gradio application, focusing on error handling, telemetry, and version checking. It ensures robust analytics behavior across different environments including WASM mode.

Test Coverage Overview

The test suite provides comprehensive coverage of analytics-related functionality in Gradio.

Key areas tested include:
  • Version check error handling
  • Telemetry request processing
  • Error analytics in both standard and WASM environments
  • Connection error handling
Edge cases covered include JSON parsing errors and network connection failures.

Implementation Analysis

The testing approach utilizes pytest’s powerful mocking capabilities to simulate various scenarios and environments.

Notable patterns include:
  • Use of unittest.mock.patch for HTTP request mocking
  • Environment variable manipulation via monkeypatch
  • Async testing for WASM-specific functionality
  • Warning capture and verification

Technical Details

Testing tools and configuration:
  • pytest framework with asyncio support
  • httpx mocking for HTTP requests
  • Environment variable control through monkeypatch
  • Warning management via warnings module
  • Mock objects for analytics service interaction

Best Practices Demonstrated

The test suite exemplifies several testing best practices.

Notable examples include:
  • Isolation of external dependencies through mocking
  • Proper async/await pattern usage
  • Comprehensive error case handling
  • Clean separation of test scenarios
  • Environment-specific testing considerations

gradio-app/gradio

test/test_analytics.py

            
import asyncio
import json
import os
import warnings
from unittest.mock import patch

import pytest

from gradio import analytics, wasm_utils

os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"


class TestAnalytics:
    @patch("httpx.get")
    def test_should_warn_with_unable_to_parse(self, mock_get, monkeypatch):
        monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "True")
        mock_get.side_effect = json.decoder.JSONDecodeError("Expecting value", "", 0)

        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter("always")
            analytics.version_check()
            assert (
                str(w[-1].message)
                == "unable to parse version details from package URL."
            )

    @patch("gradio.analytics._send_telemetry_in_thread")
    def test_error_analytics_doesnt_crash_on_connection_error(
        self, mock_send, monkeypatch
    ):
        monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "True")
        mock_send.side_effect = Exception("Connection error")
        analytics._do_normal_analytics_request("placeholder", {})
        mock_send.assert_called()

    @patch("gradio.analytics._send_telemetry_in_thread")
    def test_error_analytics_successful(self, mock_post, monkeypatch):
        monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "True")
        analytics.error_analytics("placeholder")

    @patch.object(wasm_utils, "IS_WASM", True)
    @patch("gradio.analytics.pyodide_pyfetch")
    @pytest.mark.asyncio
    async def test_error_analytics_successful_in_wasm_mode(
        self, pyodide_pyfetch, monkeypatch
    ):
        loop = asyncio.get_event_loop()
        monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "True")

        analytics.error_analytics("placeholder")

        # Await all background tasks.
        # Ref: https://superfastpython.com/asyncio-wait-for-tasks/#How_to_Wait_for_All_Background_Tasks
        all_tasks = asyncio.all_tasks(loop)
        current_task = asyncio.current_task()
        assert current_task
        all_tasks.remove(current_task)
        await asyncio.wait(all_tasks)

        pyodide_pyfetch.assert_called()