test: stabilize proxy tests

Author: jpnurmi

tests/assertions.py

@@ -666,13 +666,18 @@ def assert_failed_proxy_auth_request(stdout):
     )
 
 
-def wait_for_file(path, timeout=10.0, poll_interval=0.1):
-    import glob
+def wait_for(condition, timeout=10.0, interval=0.1):
     import time
 
     deadline = time.time() + timeout
     while time.time() < deadline:
-        if glob.glob(str(path)):
+        if condition():
             return True
-        time.sleep(poll_interval)
-    return False
+        time.sleep(interval)
+    return condition()
+
+
+def wait_for_file(path, timeout=10.0, interval=0.1):
+    import glob
+
+    return wait_for(lambda: glob.glob(str(path)), timeout, interval)

tests/proxy.py

@@ -8,7 +8,7 @@
 
 import pytest
 
-from tests.assertions import assert_no_proxy_request
+from tests.assertions import assert_no_proxy_request, wait_for
 
 
 @contextlib.contextmanager
@@ -126,13 +126,14 @@ def proxy_test_finally(
     proxy_process,
     proxy_log_assert=assert_no_proxy_request,
     expected_proxy_logsize=None,
+    timeout=10,
 ):
     if expected_proxy_logsize is None:
         expected_proxy_logsize = expected_httpserver_logsize
 
     if proxy_process:
         # Give mitmdump some time to get a response from the mock server
-        time.sleep(0.5)
+        wait_for(lambda: len(httpserver.log) >= expected_httpserver_logsize, timeout)
         proxy_process.terminate()
         stdout_bytes, _ = proxy_process.communicate()
         stdout = stdout_bytes.decode("utf-8", errors="replace")

tests/test_dotnet_signals.py

@@ -8,6 +8,7 @@
 import pytest
 
 from tests import adb
+from tests.assertions import wait_for
 from tests.conditions import is_android, is_arm32, is_tsan, is_x86, is_asan
 
 project_fixture_path = pathlib.Path("tests/fixtures/dotnet_signal")
@@ -281,15 +282,6 @@ def test_aot_signals_inproc(cmake):
 ANDROID_PACKAGE = "io.sentry.ndk.dotnet.signal.test"
 
 
-def wait_for(condition, timeout=10, interval=0.5):
-    start = time.time()
-    while time.time() - start < timeout:
-        if condition():
-            return True
-        time.sleep(interval)
-    return condition()
-
-
 def run_android(args=None, strategy=None, reinit=False, timeout=30):
     if args is None:
         args = []