75 lines
1.6 KiB
Python
75 lines
1.6 KiB
Python
from multiprocessing import Process
|
|
from time import sleep
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from image_prediction.flask import make_prediction_server, run_prediction_server
|
|
|
|
|
|
@pytest.fixture
|
|
def host():
|
|
return "127.0.0.1"
|
|
|
|
|
|
@pytest.fixture
|
|
def port(host):
|
|
import socket
|
|
sock = socket.socket()
|
|
sock.bind((host, 0))
|
|
return sock.getsockname()[1]
|
|
|
|
|
|
@pytest.fixture
|
|
def url(host, port):
|
|
return f"http://{host}:{port}"
|
|
|
|
|
|
@pytest.fixture
|
|
def predict_fn():
|
|
def predict(_):
|
|
return 42
|
|
|
|
return predict
|
|
|
|
|
|
@pytest.fixture
|
|
def run_server_args(host, port, predict_fn):
|
|
prediction_server = make_prediction_server(predict_fn)
|
|
return {"app": prediction_server, "host": host, "port": port}
|
|
|
|
|
|
@pytest.fixture
|
|
def server(run_server_args):
|
|
|
|
def get_server_process():
|
|
return Process(target=run_prediction_server, kwargs=run_server_args)
|
|
|
|
server = get_server_process()
|
|
server.start()
|
|
yield
|
|
server.terminate()
|
|
|
|
|
|
# def test_run_server():
|
|
# prediction_server = make_prediction_server(predict_fn)
|
|
# return Process(target=run_prediction_server, kwargs={"app": prediction_server, "host": host, "port": port})
|
|
|
|
|
|
def test_server_predict(server, url):
|
|
response = requests.post(url)
|
|
response.raise_for_status()
|
|
assert response.json() == 42
|
|
|
|
|
|
def test_server_health_check(server, url):
|
|
response = requests.get(f"{url}/health")
|
|
response.raise_for_status()
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_server_ready_check(server, url):
|
|
response = requests.get(f"{url}/ready")
|
|
response.raise_for_status()
|
|
assert response.status_code == 200
|