revert changes

Author: Andrei Neagu

services/dynamic-sidecar/tests/unit/test_modules_outputs_event_filter.py

@@ -1,9 +1,9 @@
 # pylint:disable=redefined-outer-name
 # pylint:disable=unused-argument
 
-from collections.abc import AsyncIterator
+import asyncio
 from pathlib import Path
-from typing import Any, Final
+from typing import AsyncIterator, Iterator
 from unittest.mock import AsyncMock
 
 import pytest
@@ -24,12 +24,12 @@
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_fixed
 
-_TENACITY_RETRY_PARAMS: Final[dict[str, Any]] = {
-    "reraise": True,
-    "retry": retry_if_exception_type(AssertionError),
-    "stop": stop_after_delay(10),
-    "wait": wait_fixed(0.01),
-}
+_TENACITY_RETRY_PARAMS = dict(
+    reraise=True,
+    retry=retry_if_exception_type(AssertionError),
+    stop=stop_after_delay(10),
+    wait=wait_fixed(0.01),
+)
 
 # FIXTURES
 
@@ -75,11 +75,11 @@ async def outputs_manager(
 @pytest.fixture
 def mocked_port_key_content_changed(
     mocker: MockerFixture, outputs_manager: OutputsManager
-) -> AsyncMock:
+) -> Iterator[AsyncMock]:
     async def _mock_upload_outputs(*args, **kwargs) -> None:
         pass
 
-    return mocker.patch.object(
+    yield mocker.patch.object(
         outputs_manager, "port_key_content_changed", side_effect=_mock_upload_outputs
     )
 
@@ -101,8 +101,8 @@ def get_wait_interval(self, dir_size: NonNegativeInt) -> NonNegativeFloat:
 
 
 @pytest.fixture
-def mock_get_directory_total_size(mocker: MockerFixture) -> AsyncMock:
-    return mocker.patch(
+def mock_get_directory_total_size(mocker: MockerFixture) -> Iterator[AsyncMock]:
+    yield mocker.patch(
         "simcore_service_dynamic_sidecar.modules.outputs._event_filter.get_directory_total_size",
         return_value=1,
     )
@@ -120,6 +120,17 @@ async def event_filter(
     await event_filter.shutdown()
 
 
+# UTILS
+
+
+async def _wait_for_event_to_trigger(event_filter: EventFilter) -> None:
+    await asyncio.sleep(event_filter.delay_policy.get_min_interval() * 5)
+
+
+async def _wait_for_event_to_trigger_big_directory(event_filter: EventFilter) -> None:
+    await asyncio.sleep(event_filter.delay_policy.get_wait_interval(1) * 2)
+
+
 # TESTS
 
 
@@ -130,16 +141,13 @@ async def test_event_triggers_once(
 ):
     # event triggers once
     await event_filter.enqueue(port_key_1)
-
-    async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
-        with attempt:
-            assert mocked_port_key_content_changed.call_count == 1
+    await _wait_for_event_to_trigger(event_filter)
+    assert mocked_port_key_content_changed.call_count == 1
 
     # event triggers a second time
     await event_filter.enqueue(port_key_1)
-    async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
-        with attempt:
-            assert mocked_port_key_content_changed.call_count == 2
+    await _wait_for_event_to_trigger(event_filter)
+    assert mocked_port_key_content_changed.call_count == 2
 
 
 async def test_trigger_once_after_event_chain(
@@ -149,9 +157,8 @@ async def test_trigger_once_after_event_chain(
 ):
     for _ in range(100):
         await event_filter.enqueue(port_key_1)
-    async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
-        with attempt:
-            assert mocked_port_key_content_changed.call_count == 1
+    await _wait_for_event_to_trigger(event_filter)
+    assert mocked_port_key_content_changed.call_count == 1
 
 
 async def test_always_trigger_after_delay(
@@ -163,9 +170,8 @@ async def test_always_trigger_after_delay(
     # event trigger after correct interval delay correctly
     for expected_call_count in range(1, 10):
         await event_filter.enqueue(port_key_1)
-        async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
-            with attempt:
-                assert mocked_port_key_content_changed.call_count == expected_call_count
+        await _wait_for_event_to_trigger_big_directory(event_filter)
+        assert mocked_port_key_content_changed.call_count == expected_call_count
 
 
 async def test_minimum_amount_of_get_directory_total_size_calls(
@@ -177,12 +183,14 @@ async def test_minimum_amount_of_get_directory_total_size_calls(
     await event_filter.enqueue(port_key_1)
     # wait a bit for the vent to be picked up
     # by the workers and processed
+    await _wait_for_event_to_trigger(event_filter)
     async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
         with attempt:
             assert mock_get_directory_total_size.call_count == 1
             assert mocked_port_key_content_changed.call_count == 0
 
     # event finished processing and was dispatched
+    await _wait_for_event_to_trigger_big_directory(event_filter)
     async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
         with attempt:
             assert mock_get_directory_total_size.call_count == 2
@@ -198,27 +206,24 @@ async def test_minimum_amount_of_get_directory_total_size_calls_with_continuous_
     await event_filter.enqueue(port_key_1)
     # wait a bit for the vent to be picked up
     # by the workers and processed
-    async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
-        with attempt:
-            assert mock_get_directory_total_size.call_count == 1
-            assert mocked_port_key_content_changed.call_count == 0
+    await _wait_for_event_to_trigger(event_filter)
+    assert mock_get_directory_total_size.call_count == 1
+    assert mocked_port_key_content_changed.call_count == 0
 
     # while changes keep piling up, keep extending the duration
     # no event will trigger
     # size of directory will not be computed
     VERY_LONG_EVENT_CHAIN = 1000
     for _ in range(VERY_LONG_EVENT_CHAIN):
         await event_filter.enqueue(port_key_1)
-        async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
-            with attempt:
-                assert mock_get_directory_total_size.call_count == 1
-                assert mocked_port_key_content_changed.call_count == 0
+        await _wait_for_event_to_trigger(event_filter)
+        assert mock_get_directory_total_size.call_count == 1
+        assert mocked_port_key_content_changed.call_count == 0
 
     # event finished processing and was dispatched
-    async for attempt in AsyncRetrying(**_TENACITY_RETRY_PARAMS):
-        with attempt:
-            assert mock_get_directory_total_size.call_count == 2
-            assert mocked_port_key_content_changed.call_count == 1
+    await _wait_for_event_to_trigger_big_directory(event_filter)
+    assert mock_get_directory_total_size.call_count == 2
+    assert mocked_port_key_content_changed.call_count == 1
 
 
 def test_default_delay_policy():