test: improve pytest fixtures

integration_tests/conftest.py

@@ -1,6 +1,7 @@
 import os
 import pathlib
 import signal
+import socket
 import subprocess
 import sys
 import time
@@ -32,62 +33,82 @@ def kill_process(process: subprocess.Popen) -> None:
         pass
 
 
-@pytest.fixture(scope="session")
-def session():
-    os.environ["ROBYN_URL"] = "127.0.0.1"
+def start_server(domain: str, port: int, is_dev: bool = False) -> subprocess.Popen:
+    """
+    Call this method to wait for the server to start
+    """
+    # Start the server
     current_file_path = pathlib.Path(__file__).parent.resolve()
     base_routes = os.path.join(current_file_path, "./base_routes.py")
     command = ["python3", base_routes]
+    if is_dev:
+        command.append("--dev")
     process = spawn_process(command)
-    time.sleep(1)
+
+    # Wait for the server to be reachable
+    timeout = 5  # The maximum time we will wait for an answer
+    start_time = time.time()
+    while True:
+        current_time = time.time()
+        if current_time - start_time > timeout:
+            # Robyn didn't start correctly before timeout, kill the process and exit with an exception
+            kill_process(process)
+            raise ConnectionError("Could not reach Robyn server")
+        try:
+            sock = socket.create_connection((domain, port), timeout=5)
+            sock.close()
+            break  # We were able to reach the server, exit the loop
+        except Exception:
+            pass
+    return process
+
+
+@pytest.fixture(scope="session")
+def session():
+    domain = "127.0.0.1"
+    port = 8080
+    os.environ["ROBYN_URL"] = domain
+    process = start_server(domain, port)
     yield
     kill_process(process)
 
 
 @pytest.fixture(scope="session")
 def default_session():
-    current_file_path = pathlib.Path(__file__).parent.resolve()
-    base_routes = os.path.join(current_file_path, "./base_routes.py")
-    command = ["python3", base_routes]
-    process = spawn_process(command)
-    time.sleep(1)
+    domain = "127.0.0.1"
+    port = 8080
+    process = start_server(domain, port)
     yield
     kill_process(process)
 
 
 @pytest.fixture(scope="session")
 def global_session():
-    os.environ["ROBYN_URL"] = "0.0.0.0"
-    current_file_path = pathlib.Path(__file__).parent.resolve()
-    base_routes = os.path.join(current_file_path, "./base_routes.py")
-    command = ["python3", base_routes]
-    process = spawn_process(command)
-    time.sleep(1)
+    domain = "0.0.0.0"
+    port = 8080
+    os.environ["ROBYN_URL"] = domain
+    process = start_server(domain, port)
     yield
     kill_process(process)
 
 
 @pytest.fixture(scope="session")
 def dev_session():
-    os.environ["ROBYN_URL"] = "127.0.0.1"
-    os.environ["ROBYN_PORT"] = "8081"
-    current_file_path = pathlib.Path(__file__).parent.resolve()
-    base_routes = os.path.join(current_file_path, "./base_routes.py")
-    command = ["python3", base_routes, "--dev"]
-    process = spawn_process(command)
-    time.sleep(1)
+    domain = "127.0.0.1"
+    port = 8081
+    os.environ["ROBYN_URL"] = domain
+    os.environ["ROBYN_PORT"] = str(port)
+    process = start_server(domain, port)
     yield
     kill_process(process)
 
 
 @pytest.fixture(scope="session")
 def test_session():
-    os.environ["ROBYN_URL"] = "127.0.0.1"
-    os.environ["ROBYN_PORT"] = "8080"
-    current_file_path = pathlib.Path(__file__).parent.resolve()
-    base_routes = os.path.join(current_file_path, "./base_routes.py")
-    command = ["python3", base_routes, "--dev"]
-    process = spawn_process(command)
-    time.sleep(1)
+    domain = "127.0.0.1"
+    port = 8080
+    os.environ["ROBYN_URL"] = domain
+    os.environ["ROBYN_PORT"] = str(port)
+    process = start_server(domain, port, is_dev=True)
     yield
     kill_process(process)