File size: 3,001 Bytes
a03b3ba |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 |
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")
# 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()
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"
|