feat testsuite: more robust waitings

commit_hash:c1934115b6550dfbe0f5dc06b70fd79504294087

Author: segoon

core/functional_tests/basic_chaos/tests-restart/test_restart.py

@@ -1,11 +1,11 @@
-import asyncio
-import datetime
+import pytest_userver.utils.sync as sync
 
 
 async def wait_for_daemon_stop(_global_daemon_store):
-    deadline = datetime.datetime.now() + datetime.timedelta(seconds=10)
-    while datetime.datetime.now() < deadline and _global_daemon_store.has_running_daemons():
-        await asyncio.sleep(0.05)
+    def is_ready():
+        return not _global_daemon_store.has_running_daemons()
+
+    await sync.wait(is_ready)
 
     assert not _global_daemon_store.has_running_daemons(), 'Daemon has not stopped'
     await _global_daemon_store.aclose()

core/functional_tests/early_monitor_port_open/tests/test_boot.py

@@ -2,6 +2,7 @@
 
 import aiohttp
 import pytest
+import pytest_userver.utils.sync as sync
 
 
 @pytest.mark.skip(reason='the feature is not yet enabled')
@@ -15,16 +16,12 @@ async def test_monitor_port_is_open_before_all_components_are_ready(
 
     # call /monitor
     async with aiohttp.ClientSession() as session:
-        for i in range(120):
-            try:
-                response = await session.get(monitor_baseurl + 'monitor')
-                if response.status == 200:
-                    break
-                await asyncio.sleep(0.5)
-            except aiohttp.ClientConnectorError:
-                await asyncio.sleep(0.5)
-        else:
-            assert False, 'Service has not started'
+
+        async def is_ready():
+            response = await session.get(monitor_baseurl + 'monitor')
+            return response.status == 200
+
+        await sync.wait(is_ready, catch=aiohttp.ClientConnectorError)
 
     # make sure /ping is now ready
     await daemon_task

core/functional_tests/graceful_shutdown/tests/test_graceful_shutdown.py

@@ -1,11 +1,13 @@
 import asyncio
-import datetime
+
+import pytest_userver.utils.sync as sync
 
 
 async def wait_for_daemon_stop(_global_daemon_store):
-    deadline = datetime.datetime.now() + datetime.timedelta(seconds=10)
-    while datetime.datetime.now() < deadline and _global_daemon_store.has_running_daemons():
-        await asyncio.sleep(0.05)
+    def is_ready():
+        return not _global_daemon_store.has_running_daemons()
+
+    await sync.wait(is_ready)
 
     assert not _global_daemon_store.has_running_daemons(), 'Daemon has not stopped'
     await _global_daemon_store.aclose()

core/functional_tests/http2server/tests/test_http2server.py

@@ -9,6 +9,7 @@
 import h2.events
 import h2.settings
 import pytest
+import pytest_userver.utils.sync as sync
 
 DEFAULT_PATH = '/http2server'
 DEFAULT_DATA = {'hello': 'world'}
@@ -134,19 +135,26 @@ async def test_concurrent_requests(
 
     await service_client.update_server_state()
 
-    await asyncio.sleep(1.0)
+    def expect_eq(a, b):
+        if a == b:
+            return
+        raise sync.NotReady
+
+    async def is_ready():
+        metrics = await monitor_client.metrics(prefix='server.requests.http2')
+        expect_eq(len(metrics), 5)
+        total_requests = clients_count * req_per_client + current_streams
+        expect_eq(total_requests, await _get_metric(monitor_client, 'streams-count'))
+        expect_eq(total_requests, await _get_metric(monitor_client, 'streams-close'))
+        expect_eq(0, await _get_metric(monitor_client, 'reset-streams'))
+        expect_eq(0, await _get_metric(monitor_client, 'goaway'))
+        expect_eq(
+            streams_parse_error,
+            await _get_metric(monitor_client, 'streams-parse-error'),
+        )
+        return True
 
-    metrics = await monitor_client.metrics(prefix='server.requests.http2')
-    assert len(metrics) == 5
-    total_requests = clients_count * req_per_client + current_streams
-    assert total_requests == await _get_metric(monitor_client, 'streams-count')
-    assert total_requests == await _get_metric(monitor_client, 'streams-close')
-    assert 0 == await _get_metric(monitor_client, 'reset-streams')
-    assert 0 == await _get_metric(monitor_client, 'goaway')
-    assert streams_parse_error == await _get_metric(
-        monitor_client,
-        'streams-parse-error',
-    )
+    await sync.wait(is_ready)
 
 
 async def test_concurrent_requests_with_big_body(

core/functional_tests/signal_during_boot/tests/test_rotate_logs.py

@@ -1,24 +1,20 @@
-import asyncio
 import os
 import signal
 
+import pytest_userver.utils.sync as sync
+
 TIMEOUT = 60
 
 
 async def wait_for_logfile(fname: str):
-    for i in range(TIMEOUT):
+    async def check_ready():
         with open(fname) as ifile:
             for line in ifile.readlines():
                 if 'Starting to catch signals' in line:
-                    # Yes, sleeping for sync. Yes, that's bad.
-                    # No, it is not possible to wait for a specific event.
-                    await asyncio.sleep(1)
                     return
+        raise sync.NotReady()
 
-        # not yet started, waiting
-        await asyncio.sleep(1)
-
-    assert False, f'log file {fname} does not contain start log entry, has the service started?'
+    await sync.wait_until(check_ready)
 
 
 async def test_boot_logs(
@@ -50,9 +46,8 @@ async def _checker(*, session, process) -> bool:
 
         daemon.process.send_signal(signal.SIGUSR1)
 
-        for i in range(TIMEOUT):
-            if os.path.exists(_service_logfile_path):
-                break
-            await asyncio.sleep(1)
-        else:
-            assert False, f'log file {_service_logfile_path} is not reopened'
+        async def check_ready():
+            if not os.path.exists(_service_logfile_path):
+                raise sync.NotReady()
+
+        await sync.wait_until(check_ready)

core/functional_tests/websocket/tests/test_websocket.py

@@ -2,6 +2,7 @@
 import logging
 
 import pytest
+import pytest_userver.utils.sync as sync
 import websockets
 
 # Disabling redundant logs from third party library
@@ -146,15 +147,14 @@ async def test_ping_pong_close(websocket_client):
         websocket_client.ping_interval = None
         websocket_client.ping_timeout = None
 
-        for _ in range(20):
-            try:
-                await chat.recv()
-                await asyncio.sleep(1)
-            except websockets.exceptions.ConnectionClosed:
-                connection_closed_by_ping = True
-                break
+        async def check_ready():
+            await chat.recv()
+            raise sync.NotReady()
 
-        assert connection_closed_by_ping
+        try:
+            await sync.wait_until(check_ready)
+        except websockets.exceptions.ConnectionClosed:
+            pass
 
 
 async def test_upgrade_header_with_tab_then_reconnect(service_port):

redis/functional_tests/basic_chaos/tests/test_redis.py

@@ -1,6 +1,7 @@
 import asyncio
 
 import pytest
+import pytest_userver.utils.sync as sync
 
 
 async def _check_that_restores(client, gate):
@@ -11,13 +12,11 @@ async def _check_that_restores(client, gate):
 
     assert gate.connections_count() >= 1
 
-    for _ in range(10):
+    async def is_ready():
         res = await client.delete('/chaos?key=foo')
-        if res.status == 200:
-            return
-        await asyncio.sleep(1)
+        return res.status == 200
 
-    assert False, 'Bad results after connection restore'
+    await sync.wait(is_ready)
 
 
 async def _check_crud(client):

redis/functional_tests/cluster_auto_topology_pubsub/tests/test_redis_pubsub.py

@@ -1,6 +1,7 @@
 import asyncio
 
 import pytest
+import pytest_userver.utils.sync as sync
 import redis
 
 # Some messages may be lost (it's a Redis limitation). *_failover tests require more than 100 retries on slow CI
@@ -38,7 +39,8 @@ async def _validate_pubsub(
     for i in range(INPUT_CHANNELS_COUNT):
         channel_name = channel_prefix + str(i)
         message = msg + str(i)
-        for _ in range(REQUESTS_RETRIES):
+
+        async def check_ready():
             publish_method(redis_db, channel_name, message)
 
             response = await service_client.get(
@@ -52,11 +54,11 @@ async def _validate_pubsub(
             if data:
                 assert message in data
                 await service_client.delete(url)
-                break
+                return
 
-            await asyncio.sleep(REQUESTS_RELAX_TIME)
-        else:
-            assert False, f'Retries exceeded trying to read from {channel_name}'
+            raise sync.NotReady()
+
+        await sync.wait_until(check_ready)
 
 
 async def _test_service_subscription(service_client, node, prefix):
@@ -93,13 +95,13 @@ async def _validate_service_publish(service_client, nodes, shards_count=0):
     redis_clients = [(node, node.get_client()) for node in nodes]
 
     async def _get_message(pubsub, retries=5, delay=0.5):
-        ret = None
-        for _ in range(retries):
+        async def check_ready():
             ret = pubsub.get_message()
             if ret is not None:
                 return ret
-            await asyncio.sleep(delay)
-        return ret
+            raise sync.NotReady()
+
+        return await sync.wait_until(check_ready)
 
     async def _ensure_published(
         pubsub,
@@ -120,8 +122,13 @@ async def _ensure_published(
         assert False, 'Retries exceeded'
 
     async def _validate(service_client, pubsub, prefix):
-        for index in range(REQUESTS_RETRIES):
+        index = 1
+
+        async def check_ready():
+            nonlocal index
+            index = index + 1
             msg = prefix + str(index)
+
             try:
                 response = await service_client.get(url, params={'publish': msg})
                 assert response.status == 200
@@ -130,8 +137,9 @@ async def _validate(service_client, pubsub, prefix):
                 return
             except Exception as exc:  # pylint: disable=broad-except
                 print(f'Pubsub validation failed for shard zero: {exc}')
-            await asyncio.sleep(REQUESTS_RELAX_TIME)
-        assert False, 'Retries exceeded for shard zero'
+                raise sync.NotReady()
+
+        await sync.wait_until(check_ready)
 
     async def _validate_round_robin(service_client, pubsub, prefix):
         successes = list()
@@ -180,13 +188,13 @@ async def _validate_service_spublish(service_client, nodes):
     """
 
     async def _get_message(pubsub, retries=5, delay=0.5):
-        ret = None
-        for _ in range(retries):
+        async def check_ready():
             ret = pubsub.get_sharded_message()
             if ret is not None:
                 return ret
-            await asyncio.sleep(delay)
-        return ret
+            raise sync.NotReady()
+
+        return await sync.wait_until(check_ready)
 
     async def _ensure_published(
         pubsub,

redis/functional_tests/integration_tests/tests/test_redis_cluster.py

@@ -1,9 +1,9 @@
 import asyncio
 
+import pytest_userver.utils.sync as sync
 import redis
 
 KEYS_SEQ_LEN = 10  # enough sequential keys to test all shards
-FAILOVER_DEADLINE_SEC = 30  # maximum time allowed to finish failover
 
 
 async def test_happy_path(service_client):
@@ -32,12 +32,11 @@ async def _check_write_all_shards(service_client, key_prefix, value):
 
 
 async def _wait_for_replicas_and_masters_negotiation(service_client, key, value):
-    for _ in range(FAILOVER_DEADLINE_SEC):
-        write_ok = await _check_write_all_shards(service_client, key, value)
-        if write_ok:
-            break
-        await asyncio.sleep(1)
-    assert write_ok
+    async def check_ready():
+        if not await _check_write_all_shards(service_client, key, value):
+            raise sync.NotReady()
+
+    await sync.wait_until(check_ready)
 
 
 async def _check_read_all_shards(service_client, key_prefix, value):
@@ -75,17 +74,17 @@ async def test_failover(service_client, redis_cluster_store):
     await _assert_read_all_shards(service_client, 'hf_key1', 'abc')
 
     # Replica may be syncing, use some retries
-    for _ in range(FAILOVER_DEADLINE_SEC):
-        read_ok = await _check_read_all_shards(service_client, 'hf_key2', 'cde')
-        if read_ok:
-            break
-        await asyncio.sleep(1)
-    assert read_ok
+    async def check_ready():
+        if not await _check_read_all_shards(service_client, 'hf_key2', 'cde'):
+            raise sync.NotReady()
+
+    await sync.wait_until(check_ready)
 
     # Failover master back where it was and make sure it gets there
     assert redis_cluster_store.cluster_failover(target_node=primary)
     await _wait_for_replicas_and_masters_negotiation(service_client, 'hf_key3', 'xyz')
-    for _ in range(FAILOVER_DEADLINE_SEC):
+
+    async def check_ready():
         try:
             redis_cluster_store.flushall(target_nodes=redis.cluster.RedisCluster.PRIMARIES)
             redis_cluster_store.wait(1, 10, target_nodes=redis.cluster.RedisCluster.PRIMARIES)
@@ -94,4 +93,6 @@ async def test_failover(service_client, redis_cluster_store):
             redis_cluster_store.close()
             redis_cluster_store.nodes_manager.reset()
             redis_cluster_store.nodes_manager.initialize()
-            await asyncio.sleep(1)
+            raise sync.NotReady()
+
+    await sync.wait_until(check_ready)

redis/functional_tests/integration_tests/tests/test_redis_sentinel.py

@@ -1,7 +1,8 @@
 import asyncio
 
+import pytest_userver.utils.sync as sync
+
 KEYS_SEQ_LEN = 10  # enough sequential keys to test all shards
-FAILOVER_DEADLINE_SEC = 30  # maximum time allowed to finish failover
 
 
 async def test_happy_path(service_client):
@@ -47,12 +48,10 @@ async def test_failover(service_client, redis_sentinel):
     redis_sentinel.sentinel_failover('test_master1')
 
     # Wait for failover to happen
-    for i in range(FAILOVER_DEADLINE_SEC):
-        write_ok = await _check_write_all_shards(service_client, 'failover_{i}_', 'xyz')
-        if write_ok:
-            break
-        await asyncio.sleep(1)
-    assert write_ok
+    async def is_ready():
+        return await _check_write_all_shards(service_client, 'failover_{i}_', 'xyz')
+
+    await sync.wait(is_ready)
 
     # Now that one of the replicas has become the master,
     # check reading from the remaining replica