|
"""Contains tests for networking.py and app.py""" |
|
|
|
import os |
|
import urllib |
|
import warnings |
|
|
|
import pytest |
|
from fastapi.testclient import TestClient |
|
|
|
import gradio as gr |
|
from gradio import Interface, networking |
|
|
|
os.environ["GRADIO_ANALYTICS_ENABLED"] = "False" |
|
|
|
|
|
class TestPort: |
|
def test_port_is_in_range(self): |
|
start = 7860 |
|
end = 7960 |
|
try: |
|
port = networking.get_first_available_port(start, end) |
|
assert start <= port <= end |
|
except OSError: |
|
warnings.warn("Unable to test, no ports available") |
|
|
|
def test_same_port_is_returned(self): |
|
start = 7860 |
|
end = 7960 |
|
try: |
|
port1 = networking.get_first_available_port(start, end) |
|
port2 = networking.get_first_available_port(start, end) |
|
assert port1 == port2 |
|
except OSError: |
|
warnings.warn("Unable to test, no ports available") |
|
|
|
|
|
class TestInterfaceErrors: |
|
def test_processing_error(self): |
|
io = Interface(lambda x: 1 / x, "number", "number") |
|
app, _, _ = io.launch(show_error=True, prevent_thread_lock=True) |
|
client = TestClient(app) |
|
response = client.post("/api/predict/", json={"data": [0], "fn_index": 0}) |
|
assert response.status_code == 500 |
|
assert "error" in response.json() |
|
io.close() |
|
|
|
def test_validation_error(self): |
|
io = Interface(lambda x: 1 / x, "number", "number") |
|
app, _, _ = io.launch(show_error=True, prevent_thread_lock=True) |
|
client = TestClient(app) |
|
response = client.post("/api/predict/", json={"fn_index": [0]}) |
|
assert response.status_code == 422 |
|
io.close() |
|
|
|
|
|
class TestStartServer: |
|
|
|
@pytest.mark.parametrize("host", ["127.0.0.1", "[::1]"]) |
|
def test_start_server(self, host): |
|
io = Interface(lambda x: x, "number", "number") |
|
io.favicon_path = None |
|
io.config = io.get_config_file() |
|
io.show_error = True |
|
io.flagging_callback.setup(gr.Number(), io.flagging_dir) |
|
io.auth = None |
|
io.host = host |
|
|
|
port = networking.get_first_available_port( |
|
networking.INITIAL_PORT_VALUE, |
|
networking.INITIAL_PORT_VALUE + networking.TRY_NUM_PORTS, |
|
) |
|
_, _, local_path, _, server = networking.start_server(io, server_port=port) |
|
url = urllib.parse.urlparse(local_path) |
|
assert url.scheme == "http" |
|
assert url.port == port |
|
server.close() |
|
|
|
|
|
class TestURLs: |
|
def test_url_ok(self): |
|
res = networking.url_ok("https://www.gradio.app") |
|
assert res |
|
|
|
|
|
def test_start_server_app_kwargs(): |
|
""" |
|
Test that start_server accepts app_kwargs and they're propagated to FastAPI. |
|
""" |
|
io = Interface(lambda x: x, "number", "number") |
|
app, _, _ = io.launch( |
|
show_error=True, |
|
prevent_thread_lock=True, |
|
app_kwargs={ |
|
"docs_url": "/docs", |
|
}, |
|
) |
|
client = TestClient(app) |
|
assert client.get("/docs").status_code == 200 |
|
io.close() |
|
|
