|
import asyncio |
|
import ipaddress |
|
import json |
|
import os |
|
import warnings |
|
from unittest import mock as mock |
|
|
|
import httpx |
|
import pytest |
|
|
|
from gradio import analytics, wasm_utils |
|
from gradio.context import Context |
|
|
|
os.environ["GRADIO_ANALYTICS_ENABLED"] = "False" |
|
|
|
|
|
class TestAnalytics: |
|
@mock.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." |
|
) |
|
|
|
@mock.patch("httpx.post") |
|
def test_error_analytics_doesnt_crash_on_connection_error( |
|
self, mock_post, monkeypatch |
|
): |
|
monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "True") |
|
mock_post.side_effect = httpx.ConnectError("Connection error") |
|
analytics._do_normal_analytics_request("placeholder", {}) |
|
mock_post.assert_called() |
|
|
|
@mock.patch("httpx.post") |
|
def test_error_analytics_successful(self, mock_post, monkeypatch): |
|
monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "True") |
|
analytics.error_analytics("placeholder") |
|
mock_post.assert_called() |
|
|
|
@mock.patch.object(wasm_utils, "IS_WASM", True) |
|
@mock.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") |
|
|
|
|
|
|
|
all_tasks = asyncio.all_tasks(loop) |
|
current_task = asyncio.current_task() |
|
all_tasks.remove(current_task) |
|
await asyncio.wait(all_tasks) |
|
|
|
pyodide_pyfetch.assert_called() |
|
|
|
|
|
class TestIPAddress: |
|
@pytest.mark.flaky |
|
def test_get_ip(self): |
|
Context.ip_address = None |
|
ip = analytics.get_local_ip_address() |
|
if ip == "No internet connection" or ip == "Analytics disabled": |
|
return |
|
ipaddress.ip_address(ip) |
|
|
|
@mock.patch("httpx.get") |
|
def test_get_ip_without_internet(self, mock_get, monkeypatch): |
|
mock_get.side_effect = httpx.ConnectError("Connection error") |
|
monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "True") |
|
Context.ip_address = None |
|
ip = analytics.get_local_ip_address() |
|
assert ip == "No internet connection" |
|
|
|
monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "False") |
|
Context.ip_address = None |
|
ip = analytics.get_local_ip_address() |
|
assert ip == "Analytics disabled" |
|
|
