gradio/test/test_networking.py
Abubakar Abid 53005ab88a
Switch from SSH tunneling to FRP (#2509)
* FRP Poc (#2396)

* FRP Poc

* Gracefully handle exceptions in thread tunneling

* comments

* Fix share error message when files are built locally (#2502)

* fix share error message

* changelog

* formatting

* tunneling rename

* version

* formatting

* remove test

* changelog

* version

Co-authored-by: Abubakar Abid <abubakar@huggingface.co>
Co-authored-by: Wauplin <lucainp@gmail.com>

* 2509

* updated url to testing.gradiodash.com

* gradiotesting

* format, version

* gradio.live

* temp fix for https

* remove unnecessary tests

* version

* updated tunnel logic

* formatting and tests

* load testing

* changes

* Make private method + generate privilege key (#2519)

* rm load test

* frp

* formatting

* Update run.py

* Update run.py

* updated message

* share=True

* [DO NOT MERGE] Add pymux for FRP (#2747)

* Add pymux for FRP

* Cleaning pyamux

* Cleaning pyamux + make it work

* Forgot the thread

* Reformat

* some logs to be removed afterwards

* added share to hello world

* Transform into object

* I guess it's cleaner now

* Handle 404 + Transform to object

* Fix params names

* Add debug

* windows fix

Co-authored-by: Wauplin <lucainp@gmail.com>
Co-authored-by: Abubakar Abid <abubakar@huggingface.co>

* removed share=True

* formatting

* hello world notebook

* version

* fixes

* formatting

* testing tunneling exists

* tests

* formatting

* lint

* Remove asyncio + kill proc on exit

* version

* version

* update changelog

* explicit message about reporting

Co-authored-by: Adrien <adrien@xcid.fr>
Co-authored-by: Wauplin <lucainp@gmail.com>
Co-authored-by: Ali Abid <aabid94@gmail.com>
2022-12-14 08:10:45 -06:00

80 lines
2.5 KiB
Python

"""Contains tests for networking.py and app.py"""
import os
import urllib
import warnings
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:
def test_start_server(self):
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
port = networking.get_first_available_port(
networking.INITIAL_PORT_VALUE,
networking.INITIAL_PORT_VALUE + networking.TRY_NUM_PORTS,
)
io.enable_queue = False
_, _, 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