From 6b56f5c64c7752f68fe464bf98f477e4edebdddc Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Fri, 19 Jan 2024 19:42:52 -0800 Subject: [PATCH] throw if moto server port if already used https://stackoverflow.com/questions/69955686/why-cant-i-run-the-project-on-port-5000 https://github.com/amundsen-io/amundsen/issues/1813 https://stackoverflow.com/questions/2470971/fast-way-to-test-if-a-port-is-in-use-using-python --- tests/conftest.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 07beba07e0..9c53301776 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -27,6 +27,7 @@ import os import re +import socket import string import uuid from datetime import datetime @@ -1587,7 +1588,7 @@ def fixture_aws_credentials() -> Generator[None, None, None]: os.environ.pop("AWS_DEFAULT_REGION") -MOTO_SERVER = ThreadedMotoServer(ip_address="localhost", port=5000) +MOTO_SERVER = ThreadedMotoServer(ip_address="localhost", port=5001) def pytest_sessionfinish( @@ -1600,10 +1601,17 @@ def pytest_sessionfinish( @pytest.fixture(scope="session") def moto_server() -> ThreadedMotoServer: + # this will throw an exception if the port is already in use + is_port_in_use(MOTO_SERVER._ip_address, MOTO_SERVER._port) MOTO_SERVER.start() return MOTO_SERVER +def is_port_in_use(ip_address: str, port: int) -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind((ip_address, port)) + + @pytest.fixture(scope="session") def moto_endpoint_url(moto_server: ThreadedMotoServer) -> str: _url = f"http://{moto_server._ip_address}:{moto_server._port}"