From e8becd2bfcf38c5f8cb3f0251873ea3e830097f7 Mon Sep 17 00:00:00 2001 From: Harkamal Jot Singh Kumar Date: Tue, 11 Feb 2025 11:49:56 +0000 Subject: [PATCH 1/2] fix flaky tests --- tests/unit/test_flow.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_flow.py b/tests/unit/test_flow.py index a3314d1..4b3a4e3 100644 --- a/tests/unit/test_flow.py +++ b/tests/unit/test_flow.py @@ -251,14 +251,26 @@ def instance(self): CLIENT_SECRETS_INFO, scopes=self.SCOPES ) + + def is_port_in_use(self, port, host='localhost'): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + return s.connect_ex((host, port)) == 0 + @pytest.fixture def port(self): # Creating a new server at the same port will result in # a 'Address already in use' error for a brief # period of time after the socket has been closed. - # Work around this in the tests by choosing a random port. + # Work around this in the tests by choosing a different port each time. # https://stackoverflow.com/questions/6380057/python-binding-socket-address-already-in-use - yield random.randrange(60400, 60900) + random_port = -1 + for _ in range(10): + random_port = random.randrange(60400, 60900) + if not self.is_port_in_use(random_port): + break + else: + raise OSError("Could not find a free port") + yield random_port @pytest.fixture def socket(self, port): From 416315ad6cc4ec0e9700dc1fd28fdda3961e3bf3 Mon Sep 17 00:00:00 2001 From: Harkamal Jot Singh Kumar Date: Tue, 11 Feb 2025 11:50:29 +0000 Subject: [PATCH 2/2] fix formatting changes --- tests/unit/test_flow.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/unit/test_flow.py b/tests/unit/test_flow.py index 4b3a4e3..5941bb9 100644 --- a/tests/unit/test_flow.py +++ b/tests/unit/test_flow.py @@ -251,8 +251,7 @@ def instance(self): CLIENT_SECRETS_INFO, scopes=self.SCOPES ) - - def is_port_in_use(self, port, host='localhost'): + def is_port_in_use(self, port, host="localhost"): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: return s.connect_ex((host, port)) == 0