mirror of
https://github.com/gradio-app/gradio.git
synced 2024-12-21 02:19:59 +08:00
7cbf96e0bd
* issue 5245 commit 1 * formatted * add changeset * fixes * fixes * make changes in client too * remove requests from client * add changeset * add changeset * Fixes to test_utils.py in client * fixes in client utils.py and test_utils.py --------- Co-authored-by: gradio-pr-bot <gradio-pr-bot@users.noreply.github.com> Co-authored-by: Abubakar Abid <abubakar@huggingface.co>
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
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"
|