From a189299beeb8918c4b606e24526007e1eb1a3786 Mon Sep 17 00:00:00 2001 From: Tyler Romero Date: Mon, 7 Oct 2024 16:37:44 -0700 Subject: [PATCH] Poll for liveness --- test/api/test_image_queries_live.py | 40 +++++++++++++++++------------ 1 file changed, 24 insertions(+), 16 deletions(-) diff --git a/test/api/test_image_queries_live.py b/test/api/test_image_queries_live.py index d02b15af..fe2e8242 100644 --- a/test/api/test_image_queries_live.py +++ b/test/api/test_image_queries_live.py @@ -1,4 +1,7 @@ +import time + import pytest +import requests from fastapi import status from groundlight import ApiException, Groundlight from model import Detector @@ -9,6 +12,7 @@ # Tests in this file require a live edge-endpoint server and GL Api token in order to run. # Not ideal for unit-testing. TEST_ENDPOINT = "http://localhost:6717" +MAX_WAIT_TIME_S = 10 # Detector ID associated with the detector with parameters # - name="edge_testing_det", @@ -20,27 +24,31 @@ @pytest.fixture(scope="module", autouse=True) def ensure_edge_endpoint_is_live(): """Ensure that the edge-endpoint server is live before running tests.""" - import requests - - try: - response = requests.get(TEST_ENDPOINT + "/health/live") - response.raise_for_status() - assert response.json().get("status") == "alive", "Edge endpoint is not live." - except requests.RequestException as e: - pytest.fail(f"Edge endpoint is not live: {e}") + start_time = time.time() + while time.time() - start_time < MAX_WAIT_TIME_S: + try: + response = requests.get(TEST_ENDPOINT + "/health/live") + response.raise_for_status() + if response.json().get("status") == "alive": + return + except requests.RequestException: + time.sleep(1) # wait for 1 second before retrying + pytest.fail("Edge endpoint is not live after polling for 10 seconds.") @pytest.fixture(scope="module", autouse=True) def ensure_edge_endpoint_is_ready(): """Ensure that the edge-endpoint server is ready before running tests.""" - import requests - - try: - response = requests.get(TEST_ENDPOINT + "/health/ready") - response.raise_for_status() - assert response.json().get("status") == "ready", "Edge endpoint is not ready." - except requests.RequestException as e: - pytest.fail(f"Edge endpoint is not live: {e}") + start_time = time.time() + while time.time() - start_time < MAX_WAIT_TIME_S: + try: + response = requests.get(TEST_ENDPOINT + "/health/ready") + response.raise_for_status() + if response.json().get("status") == "ready": + return + except requests.RequestException: + time.sleep(1) # wait for 1 second before retrying + pytest.fail("Edge endpoint is not ready after polling for 10 seconds.") @pytest.fixture(name="gl")