mirror of
https://github.com/gradio-app/gradio.git
synced 2024-11-21 01:01:05 +08:00
273 lines
10 KiB
Python
273 lines
10 KiB
Python
import io
|
|
import sys
|
|
import unittest
|
|
import unittest.mock as mock
|
|
from contextlib import contextmanager
|
|
from functools import partial
|
|
from string import capwords
|
|
|
|
import mlflow
|
|
import pytest
|
|
import requests
|
|
import wandb
|
|
from fastapi.testclient import TestClient
|
|
|
|
from gradio.blocks import Blocks
|
|
from gradio.interface import Interface, TabbedInterface, close_all, os
|
|
from gradio.layouts import TabItem, Tabs
|
|
from gradio.utils import assert_configs_are_equivalent_besides_ids
|
|
|
|
os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
|
|
|
|
|
|
@contextmanager
|
|
def captured_output():
|
|
new_out, new_err = io.StringIO(), io.StringIO()
|
|
old_out, old_err = sys.stdout, sys.stderr
|
|
try:
|
|
sys.stdout, sys.stderr = new_out, new_err
|
|
yield sys.stdout, sys.stderr
|
|
finally:
|
|
sys.stdout, sys.stderr = old_out, old_err
|
|
|
|
|
|
class TestInterface(unittest.TestCase):
|
|
def test_close(self):
|
|
io = Interface(lambda input: None, "textbox", "label")
|
|
_, local_url, _ = io.launch(prevent_thread_lock=True)
|
|
response = requests.get(local_url)
|
|
self.assertEqual(response.status_code, 200)
|
|
io.close()
|
|
with self.assertRaises(Exception):
|
|
response = requests.get(local_url)
|
|
|
|
def test_close_all(self):
|
|
interface = Interface(lambda input: None, "textbox", "label")
|
|
interface.close = mock.MagicMock()
|
|
close_all()
|
|
interface.close.assert_called()
|
|
|
|
def test_no_input_or_output(self):
|
|
with self.assertRaises(TypeError):
|
|
Interface(lambda x: x, examples=1234)
|
|
|
|
def test_partial_functions(self):
|
|
def greet(name, formatter):
|
|
return formatter("Hello " + name + "!")
|
|
|
|
greet_upper_case = partial(greet, formatter=capwords)
|
|
demo = Interface(fn=greet_upper_case, inputs="text", outputs="text")
|
|
assert demo("abubakar") == "Hello Abubakar!"
|
|
|
|
def test_examples_valid_path(self):
|
|
path = os.path.join(
|
|
os.path.dirname(__file__), "../gradio/test_data/flagged_with_log"
|
|
)
|
|
interface = Interface(lambda x: 3 * x, "number", "number", examples=path)
|
|
dataset_check = any(
|
|
[c["type"] == "dataset" for c in interface.get_config_file()["components"]]
|
|
)
|
|
self.assertTrue(dataset_check)
|
|
|
|
def test_test_launch(self):
|
|
with captured_output() as (out, err):
|
|
prediction_fn = lambda x: x
|
|
prediction_fn.__name__ = "prediction_fn"
|
|
interface = Interface(prediction_fn, "textbox", "label")
|
|
interface.test_launch()
|
|
output = out.getvalue().strip()
|
|
self.assertEqual(output, "Test launch: prediction_fn()... PASSED")
|
|
|
|
@mock.patch("time.sleep")
|
|
def test_block_thread(self, mock_sleep):
|
|
with self.assertRaises(KeyboardInterrupt):
|
|
with captured_output() as (out, _):
|
|
mock_sleep.side_effect = KeyboardInterrupt()
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
interface.launch(prevent_thread_lock=False)
|
|
output = out.getvalue().strip()
|
|
self.assertEqual(
|
|
output, "Keyboard interruption in main thread... closing server."
|
|
)
|
|
|
|
@mock.patch("gradio.utils.colab_check")
|
|
def test_launch_colab_share(self, mock_colab_check):
|
|
mock_colab_check.return_value = True
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
_, _, share_url = interface.launch(prevent_thread_lock=True)
|
|
self.assertIsNotNone(share_url)
|
|
interface.close()
|
|
|
|
@mock.patch("gradio.utils.colab_check")
|
|
@mock.patch("gradio.networking.setup_tunnel")
|
|
def test_launch_colab_share_error(self, mock_setup_tunnel, mock_colab_check):
|
|
mock_setup_tunnel.side_effect = RuntimeError()
|
|
mock_colab_check.return_value = True
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
_, _, share_url = interface.launch(prevent_thread_lock=True)
|
|
self.assertIsNone(share_url)
|
|
interface.close()
|
|
|
|
def test_interface_representation(self):
|
|
prediction_fn = lambda x: x
|
|
prediction_fn.__name__ = "prediction_fn"
|
|
repr = str(Interface(prediction_fn, "textbox", "label")).split("\n")
|
|
self.assertTrue(prediction_fn.__name__ in repr[0])
|
|
self.assertEqual(len(repr[0]), len(repr[1]))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_interface_none_interp(self):
|
|
interface = Interface(lambda x: x, "textbox", "label", interpretation=[None])
|
|
scores = (await interface.interpret(["quickest brown fox"]))[0][
|
|
"interpretation"
|
|
]
|
|
self.assertIsNone(scores)
|
|
|
|
@mock.patch("webbrowser.open")
|
|
def test_interface_browser(self, mock_browser):
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
interface.launch(inbrowser=True, prevent_thread_lock=True)
|
|
mock_browser.assert_called_once()
|
|
interface.close()
|
|
|
|
def test_examples_list(self):
|
|
examples = ["test1", "test2"]
|
|
interface = Interface(lambda x: x, "textbox", "label", examples=examples)
|
|
interface.launch(prevent_thread_lock=True)
|
|
self.assertEqual(len(interface.examples_handler.examples), 2)
|
|
self.assertEqual(len(interface.examples_handler.examples[0]), 1)
|
|
interface.close()
|
|
|
|
@mock.patch("IPython.display.display")
|
|
def test_inline_display(self, mock_display):
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
interface.launch(inline=True, prevent_thread_lock=True)
|
|
mock_display.assert_called_once()
|
|
interface.launch(inline=True, prevent_thread_lock=True)
|
|
self.assertEqual(mock_display.call_count, 2)
|
|
interface.close()
|
|
|
|
@mock.patch("comet_ml.Experiment")
|
|
def test_integration_comet(self, mock_experiment):
|
|
experiment = mock_experiment()
|
|
experiment.log_text = mock.MagicMock()
|
|
experiment.log_other = mock.MagicMock()
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
interface.launch(prevent_thread_lock=True)
|
|
interface.integrate(comet_ml=experiment)
|
|
experiment.log_text.assert_called_with("gradio: " + interface.local_url)
|
|
interface.share_url = "tmp" # used to avoid creating real share links.
|
|
interface.integrate(comet_ml=experiment)
|
|
experiment.log_text.assert_called_with("gradio: " + interface.share_url)
|
|
self.assertEqual(experiment.log_other.call_count, 2)
|
|
interface.share_url = None
|
|
interface.close()
|
|
|
|
def test_integration_mlflow(self):
|
|
mlflow.log_param = mock.MagicMock()
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
interface.launch(prevent_thread_lock=True)
|
|
interface.integrate(mlflow=mlflow)
|
|
mlflow.log_param.assert_called_with(
|
|
"Gradio Interface Local Link", interface.local_url
|
|
)
|
|
interface.share_url = "tmp" # used to avoid creating real share links.
|
|
interface.integrate(mlflow=mlflow)
|
|
mlflow.log_param.assert_called_with(
|
|
"Gradio Interface Share Link", interface.share_url
|
|
)
|
|
interface.share_url = None
|
|
interface.close()
|
|
|
|
def test_integration_wandb(self):
|
|
with captured_output() as (out, err):
|
|
wandb.log = mock.MagicMock()
|
|
wandb.Html = mock.MagicMock()
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
interface.width = 500
|
|
interface.height = 500
|
|
interface.integrate(wandb=wandb)
|
|
|
|
self.assertEqual(
|
|
out.getvalue().strip(),
|
|
"The WandB integration requires you to `launch(share=True)` first.",
|
|
)
|
|
interface.share_url = "tmp"
|
|
interface.integrate(wandb=wandb)
|
|
wandb.log.assert_called_once()
|
|
|
|
@mock.patch("requests.post")
|
|
def test_integration_analytics(self, mock_post):
|
|
mlflow.log_param = mock.MagicMock()
|
|
interface = Interface(lambda x: x, "textbox", "label")
|
|
interface.analytics_enabled = True
|
|
interface.integrate(mlflow=mlflow)
|
|
mock_post.assert_called_once()
|
|
|
|
|
|
class TestTabbedInterface(unittest.TestCase):
|
|
def test_tabbed_interface_config_matches_manual_tab(self):
|
|
interface1 = Interface(lambda x: x, "textbox", "textbox")
|
|
interface2 = Interface(lambda x: x, "image", "image")
|
|
|
|
with Blocks(mode="tabbed_interface") as demo:
|
|
with Tabs():
|
|
with TabItem(label="tab1"):
|
|
interface1.render()
|
|
with TabItem(label="tab2"):
|
|
interface2.render()
|
|
|
|
interface3 = Interface(lambda x: x, "textbox", "textbox")
|
|
interface4 = Interface(lambda x: x, "image", "image")
|
|
tabbed_interface = TabbedInterface([interface3, interface4], ["tab1", "tab2"])
|
|
|
|
self.assertTrue(
|
|
assert_configs_are_equivalent_besides_ids(
|
|
demo.get_config_file(), tabbed_interface.get_config_file()
|
|
)
|
|
)
|
|
|
|
|
|
class TestDeprecatedInterface(unittest.TestCase):
|
|
def test_deprecation_notice(self):
|
|
with self.assertWarns(Warning):
|
|
_ = Interface(lambda x: x, "textbox", "textbox", verbose=True)
|
|
|
|
|
|
class TestInterfaceInterpretation(unittest.TestCase):
|
|
def test_interpretation_from_interface(self):
|
|
def quadratic(num1: float, num2: float) -> float:
|
|
return 3 * num1**2 + num2
|
|
|
|
iface = Interface(
|
|
fn=quadratic,
|
|
inputs=["number", "number"],
|
|
outputs="number",
|
|
interpretation="default",
|
|
)
|
|
|
|
app, _, _ = iface.launch(prevent_thread_lock=True)
|
|
client = TestClient(app)
|
|
|
|
btn = next(
|
|
c["id"]
|
|
for c in iface.config["components"]
|
|
if c["props"].get("value") == "Interpret"
|
|
)
|
|
fn_index = next(
|
|
i
|
|
for i, d in enumerate(iface.config["dependencies"])
|
|
if d["targets"] == [btn]
|
|
)
|
|
|
|
response = client.post(
|
|
"/api/predict/", json={"fn_index": fn_index, "data": [10, 50, 350]}
|
|
)
|
|
self.assertTrue(response.json()["data"][0]["interpretation"] is not None)
|
|
iface.close()
|
|
close_all()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|