mirror/gradio
2
0
Fork 0
mirror of https://github.com/gradio-app/gradio.git synced 2024-11-27 01:40:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed tests

Browse Source
This commit is contained in:
Abubakar Abid 2022-01-04 12:58:37 -05:00
parent 81d711d079
commit c3cb06a17c
6 changed files with 100 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
gradio/app.py
View File

@ -39,6 +39,7 @@ app.add_middleware(
templates = Jinja2Templates(directory=STATIC_TEMPLATE_LIB)
###########
# Auth
###########
@ -102,6 +103,11 @@ def main(
{"request": request, "config": config}
)
@app.get("/config", dependencies=[Depends(login_check)])
def get_config():
return app.interface.config
@app.get("/static/{path:path}", dependencies=[Depends(login_check)])
def static_resource(path: str):
@ -109,6 +115,7 @@ def static_resource(path: str):
return RedirectResponse(GRADIO_STATIC_ROOT + path)
else:
static_file = safe_join(STATIC_PATH_LIB, path)
print('static_file', static_file)
if static_file is not None:
return FileResponse(static_file)
raise HTTPException(status_code=404, detail="Static file not found")

1
gradio/interface.py
View File

@ -278,6 +278,7 @@ class Interface:
self.flagging_callback = flagging_callback
self.flagging_dir = flagging_dir
self.save_to = None # Used for selenium tests
self.share = None
self.share_url = None
self.local_url = None

2
gradio/networking.py
View File

@ -140,7 +140,7 @@ def start_server(
queueing.init()
app.queue_thread = threading.Thread(target=queue_thread, args=(path_to_local_server,))
app.queue_thread.start()
if interface.save_to is not None:
if interface.save_to is not None: # Used for selenium tests
interface.save_to["port"] = port
app.tokens = {}
config = uvicorn.Config(app=app, port=port, host=server_name,

10
test/test_interfaces.py
View File

@ -69,19 +69,15 @@ class TestInterface(unittest.TestCase):
self.assertEqual(output, 'Test launch: prediction_fn()... PASSED')
@mock.patch("time.sleep")
def test_run_until_interupted(self, mock_sleep):
def test_block_thread(self, mock_sleep):
with self.assertRaises(KeyboardInterrupt):
with captured_output() as (out, err):
with captured_output() as (out, _):
mock_sleep.side_effect = KeyboardInterrupt()
interface = Interface(lambda x: x, "textbox", "label")
interface.enable_queue = False
thread = threading.Thread()
thread.keep_running = mock.MagicMock()
interface.run_until_interrupted(thread, None)
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

183
test/test_networking.py
View File

@ -1,40 +1,17 @@
from gradio import networking, Interface, reset_all, flagging
from fastapi.testclient import TestClient
import os
import requests
import unittest
import unittest.mock as mock
import ipaddress
import requests
import warnings
from unittest.mock import ANY, MagicMock
import urllib.request
import os
import warnings
from gradio import flagging, Interface, networking, reset_all, utils
os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
class TestUser(unittest.TestCase):
def test_id(self):
user = networking.User("test")
self.assertEqual(user.get_id(), "test")
def test_load_user(self):
user = networking.load_user("test")
self.assertEqual(user.get_id(), "test")
class TestIPAddress(unittest.TestCase):
def test_get_ip(self):
ip = networking.get_local_ip_address()
try: # check whether ip is valid
ipaddress.ip_address(ip)
except ValueError:
self.fail("Invalid IP address")
@mock.patch("requests.get")
def test_get_ip_without_internet(self, mock_get):
mock_get.side_effect = requests.ConnectionError()
ip = networking.get_local_ip_address()
self.assertEqual(ip, "No internet connection")
class TestPort(unittest.TestCase):
def test_port_is_in_range(self):
start = 7860
@ -56,11 +33,11 @@ class TestPort(unittest.TestCase):
warnings.warn("Unable to test, no ports available")
class TestFlaskRoutes(unittest.TestCase):
class TestRoutes(unittest.TestCase):
def setUp(self) -> None:
self.io = Interface(lambda x: x, "text", "text")
self.app, _, _ = self.io.launch(prevent_thread_lock=True)
self.client = self.app.test_client()
self.client = TestClient(self.app)
def test_get_main_route(self):
response = self.client.get('/')
@ -72,57 +49,49 @@ class TestFlaskRoutes(unittest.TestCase):
def test_static_files_served_safely(self):
# Make sure things outside the static folder are not accessible
response = self.client.get(r'/static/..%2findex.html')
self.assertEqual(response.status_code, 404)
response = self.client.get(r'/static/..%2f..%2fapi_docs.html')
self.assertEqual(response.status_code, 500)
self.assertEqual(response.status_code, 404)
def test_get_config_route(self):
response = self.client.get('/config/')
self.assertEqual(response.status_code, 200)
def test_enable_sharing_route(self):
path = "www.gradio.app"
response = self.client.get('/enable_sharing/www.gradio.app')
self.assertEqual(response.status_code, 200)
self.assertEqual(self.io.config["share_url"], path)
def test_predict_route(self):
response = self.client.post('/api/predict/', json={"data": ["test"]})
self.assertEqual(response.status_code, 200)
output = dict(response.get_json())
output = dict(response.json())
self.assertEqual(output["data"], ["test"])
self.assertTrue("durations" in output)
self.assertTrue("avg_durations" in output)
def test_queue_push_route(self):
networking.queue.push = mock.MagicMock(return_value=(None, None))
response = self.client.post('/api/queue/push/', json={"data": "test", "action": "test"})
self.assertEqual(response.status_code, 200)
# def test_queue_push_route(self):
# networking.queue.push = mock.MagicMock(return_value=(None, None))
# response = self.client.post('/api/queue/push/', json={"data": "test", "action": "test"})
# self.assertEqual(response.status_code, 200)
def test_queue_push_route(self):
networking.queue.get_status = mock.MagicMock(return_value=(None, None))
response = self.client.post('/api/queue/status/', json={"hash": "test"})
self.assertEqual(response.status_code, 200)
# def test_queue_push_route(self):
# networking.queue.get_status = mock.MagicMock(return_value=(None, None))
# response = self.client.post('/api/queue/status/', json={"hash": "test"})
# self.assertEqual(response.status_code, 200)
def tearDown(self) -> None:
self.io.close()
reset_all()
class TestAuthenticatedFlaskRoutes(unittest.TestCase):
class TestAuthenticatedRoutes(unittest.TestCase):
def setUp(self) -> None:
self.io = Interface(lambda x: x, "text", "text")
self.app, _, _ = self.io.launch(auth=("test", "correct_password"), prevent_thread_lock=True)
self.client = self.app.test_client()
def test_get_login_route(self):
response = self.client.get('/login')
self.assertEqual(response.status_code, 200)
self.client = TestClient(self.app)
def test_post_login(self):
response = self.client.post('/login', data=dict(username="test", password="correct_password"))
self.assertEqual(response.status_code, 302)
self.assertEqual(response.status_code, 200)
response = self.client.post('/login', data=dict(username="test", password="incorrect_password"))
self.assertEqual(response.status_code, 401)
self.assertEqual(response.status_code, 400)
def tearDown(self) -> None:
self.io.close()
@ -133,25 +102,21 @@ class TestInterfaceCustomParameters(unittest.TestCase):
def test_show_error(self):
io = Interface(lambda x: 1/x, "number", "number")
app, _, _ = io.launch(show_error=True, prevent_thread_lock=True)
client = app.test_client()
client = TestClient(app)
response = client.post('/api/predict/', json={"data": [0]})
self.assertEqual(response.status_code, 500)
self.assertTrue("error" in response.get_json())
self.assertTrue("error" in response.json())
io.close()
def test_feature_logging(self):
with mock.patch('requests.post') as mock_post:
io = Interface(lambda x: 1/x, "number", "number", analytics_enabled=True)
io = Interface(
lambda x: 1/x, "number", "number", analytics_enabled=True)
io.launch(show_error=True, prevent_thread_lock=True)
networking.log_feature_analytics("test_feature")
mock_post.assert_called_with(networking.GRADIO_FEATURE_ANALYTICS_URL, data=ANY, timeout=ANY)
io.close()
io = Interface(lambda x: 1/x, "number", "number")
io.launch(show_error=True, prevent_thread_lock=True)
with mock.patch('requests.post') as mock_post:
networking.log_feature_analytics("test_feature")
mock_post.assert_not_called()
utils.log_feature_analytics("none", "test_feature")
mock_post.assert_called_with(
utils.analytics_url + 'gradio-feature-analytics/',
data=mock.ANY, timeout=mock.ANY)
io.close()
@ -160,11 +125,18 @@ class TestFlagging(unittest.TestCase):
def test_flagging_analytics(self, mock_post):
callback = flagging.CSVLogger()
callback.flag = mock.MagicMock()
io = Interface(lambda x: x, "text", "text", analytics_enabled=True, flagging_callback=callback)
io = Interface(
lambda x: x, "text", "text",
analytics_enabled=True, flagging_callback=callback)
app, _, _ = io.launch(show_error=True, prevent_thread_lock=True)
client = app.test_client()
response = client.post('/api/flag/', json={"data": {"input_data": ["test"], "output_data": ["test"]}})
mock_post.assert_any_call(networking.GRADIO_FEATURE_ANALYTICS_URL, data=ANY, timeout=ANY)
client = TestClient(app)
response = client.post(
'/api/flag/',
json={"data": {"input_data": ["test"], "output_data": ["test"]}})
mock_post.assert_any_call(
utils.analytics_url + 'gradio-feature-analytics/',
data=mock.ANY,
timeout=mock.ANY)
callback.flag.assert_called_once()
self.assertEqual(response.status_code, 200)
io.close()
@ -173,30 +145,20 @@ class TestFlagging(unittest.TestCase):
@mock.patch("requests.post")
class TestInterpretation(unittest.TestCase):
def test_interpretation(self, mock_post):
io = Interface(lambda x: len(x), "text", "label", interpretation="default", analytics_enabled=True)
io = Interface(
lambda x: len(x), "text", "label",
interpretation="default", analytics_enabled=True)
app, _, _ = io.launch(prevent_thread_lock=True)
client = app.test_client()
client = TestClient(app)
io.interpret = mock.MagicMock(return_value=(None, None))
response = client.post('/api/interpret/', json={"data": ["test test"]})
mock_post.assert_any_call(networking.GRADIO_FEATURE_ANALYTICS_URL, data=ANY, timeout=ANY)
response = client.post(
'/api/interpret/', json={"data": ["test test"]})
mock_post.assert_any_call(
utils.analytics_url + 'gradio-feature-analytics/',
data=mock.ANY, timeout=mock.ANY)
self.assertEqual(response.status_code, 200)
io.close()
class TestState(unittest.TestCase):
def test_state_initialization(self):
io = Interface(lambda x: len(x), "text", "label")
app, _, _ = io.launch(prevent_thread_lock=True)
with app.test_request_context():
self.assertIsNone(networking.get_state())
def test_state_value(self):
io = Interface(lambda x: len(x), "text", "label")
app, _, _ = io.launch(prevent_thread_lock=True)
with app.test_request_context():
networking.set_state("test")
client = app.test_client()
client.post('/api/predict/', json={"data": [0]})
self.assertEqual(networking.get_state(), "test")
class TestURLs(unittest.TestCase):
def test_url_ok(self):
@ -214,28 +176,25 @@ class TestURLs(unittest.TestCase):
self.assertTrue(res)
class TestQueuing(unittest.TestCase):
def test_queueing(self):
io = Interface(lambda x: x, "text", "text")
app, _, _ = io.launch(prevent_thread_lock=True)
client = app.test_client()
# mock queue methods and post method
networking.queue.pop = mock.MagicMock(return_value=(None, None, None, 'predict'))
networking.queue.pass_job = mock.MagicMock(return_value=(None, None))
networking.queue.fail_job = mock.MagicMock(return_value=(None, None))
networking.queue.start_job = mock.MagicMock(return_value=None)
requests.post = mock.MagicMock(return_value=mock.MagicMock(status_code=200))
# execute queue action successfully
networking.queue_thread('test_path', test_mode=True)
networking.queue.pass_job.assert_called_once()
# execute queue action unsuccessfully
requests.post = mock.MagicMock(return_value=mock.MagicMock(status_code=500))
networking.queue_thread('test_path', test_mode=True)
networking.queue.fail_job.assert_called_once()
# no more things on the queue so methods shouldn't be called any more times
networking.queue.pop = mock.MagicMock(return_value=None)
networking.queue.pass_job.assert_called_once()
networking.queue.fail_job.assert_called_once()
# class TestQueuing(unittest.TestCase):
# def test_queueing(self):
# # mock queue methods and post method
# networking.queue.pop = mock.MagicMock(return_value=(None, None, None, 'predict'))
# networking.queue.pass_job = mock.MagicMock(return_value=(None, None))
# networking.queue.fail_job = mock.MagicMock(return_value=(None, None))
# networking.queue.start_job = mock.MagicMock(return_value=None)
# requests.post = mock.MagicMock(return_value=mock.MagicMock(status_code=200))
# # execute queue action successfully
# networking.queue_thread('test_path', test_mode=True)
# networking.queue.pass_job.assert_called_once()
# # execute queue action unsuccessfully
# requests.post = mock.MagicMock(return_value=mock.MagicMock(status_code=500))
# networking.queue_thread('test_path', test_mode=True)
# networking.queue.fail_job.assert_called_once()
# # no more things on the queue so methods shouldn't be called any more times
# networking.queue.pop = mock.MagicMock(return_value=None)
# networking.queue.pass_job.assert_called_once()
# networking.queue.fail_job.assert_called_once()
if __name__ == '__main__':

17
test/test_utils.py
View File

@ -1,3 +1,4 @@
import ipaddress
import os
import pkg_resources
import requests
@ -104,5 +105,21 @@ class TestUtils(unittest.TestCase):
readme_to_html("https://github.com/gradio-app/gradio/blob/master/README.md")
class TestIPAddress(unittest.TestCase):
def test_get_ip(self):
ip = get_local_ip_address()
try: # check whether ip is valid
ipaddress.ip_address(ip)
except ValueError:
self.fail("Invalid IP address")
@mock.patch("requests.get")
def test_get_ip_without_internet(self, mock_get):
mock_get.side_effect = requests.ConnectionError()
ip = get_local_ip_address()
self.assertEqual(ip, "No internet connection")
if __name__ == '__main__':
unittest.main()