Poll for liveness

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")