Finalize tests.

Add final timeout.

Author: Pijukatel

src/apify_client/clients/resource_clients/actor.py

@@ -317,7 +317,8 @@ def call(
                 waits indefinitely.
             logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                 default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
-                will redirect logs to the provided logger.
+                will redirect logs to the provided logger. The logger is also used to capture status and status message
+                of the other Actor run.
 
         Returns:
             The run object.
@@ -336,12 +337,11 @@ def call(
             return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
         run_client = self.root_client.run(run_id=started_run['id'])
+
         if logger == 'default':
-            log_context = run_client.get_streamed_log()
-        else:
-            log_context = run_client.get_streamed_log(to_logger=logger)
+            logger = None
 
-        with log_context:
+        with run_client.get_status_message_redirector(to_logger=logger), run_client.get_streamed_log(to_logger=logger):
             return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     def build(
@@ -722,7 +722,8 @@ async def call(
                 waits indefinitely.
             logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                 default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
-                will redirect logs to the provided logger.
+                will redirect logs to the provided logger. The logger is also used to capture status and status message
+                of the other Actor run.
 
         Returns:
             The run object.

src/apify_client/clients/resource_clients/log.py

@@ -390,7 +390,12 @@ class StatusMessageRedirector:
     Status message is logged at fixed time intervals, and there is no guarantee that all messages will be logged,
     especially in cases of frequent status message changes.
     """
+
     _force_propagate = False
+    # This is final sleep time to try to get the last status and status message of finished Actor run.
+    # The status and status message can get set on the Actor run with a delay. Sleep time does not guarantee that the
+    # final message will be captured, but increases the chances of that.
+    _final_sleep_time_s = 6
 
     def __init__(self, *, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=5)) -> None:
         """Initialize `StatusMessageRedirector`.
@@ -404,7 +409,6 @@ def __init__(self, *, to_logger: logging.Logger, check_period: timedelta = timed
         self._check_period = check_period.total_seconds()
         self._last_status_message = ''
 
-
     def _log_run_data(self, run_data: dict[str, Any] | None) -> bool:
         """Get relevant run data, log them if changed and return `True` if more data is expected.
 
@@ -423,7 +427,7 @@ def _log_run_data(self, run_data: dict[str, Any] | None) -> bool:
                 self._last_status_message = new_status_message
                 self._to_logger.info(new_status_message)
 
-            return not(run_data.get('isStatusMessageTerminal', False))
+            return not (run_data.get('isStatusMessageTerminal', False))
         return True
 
 
@@ -468,6 +472,7 @@ async def __aexit__(
         self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
     ) -> None:
         """Cancel the logging task."""
+        await asyncio.sleep(self._final_sleep_time_s)
         self.stop()
 
     async def _log_changed_status_message(self) -> None:
@@ -508,6 +513,7 @@ def stop(self) -> None:
         """Signal the _logging_thread thread to stop logging and wait for it to finish."""
         if not self._logging_thread:
             raise RuntimeError('Logging thread is not active')
+        time.sleep(self._final_sleep_time_s)
         self._stop_logging = True
         self._logging_thread.join()
         self._logging_thread = None

src/apify_client/clients/resource_clients/run.py

@@ -21,7 +21,7 @@
     StatusMessageRedirectorAsync,
     StatusMessageRedirectorSync,
     StreamedLogAsync,
-    StreamedLogSync, StatusMessageRedirector,
+    StreamedLogSync,
 )
 from apify_client.clients.resource_clients.request_queue import RequestQueueClient, RequestQueueClientAsync
 
@@ -322,7 +322,6 @@ def charge(
             ),
         )
 
-
     def get_status_message_redirector(
         self, to_logger: logging.Logger | None = None, check_period: timedelta = timedelta(seconds=1)
     ) -> StatusMessageRedirectorSync:
@@ -352,7 +351,6 @@ def get_status_message_redirector(
         return StatusMessageRedirectorSync(run_client=self, to_logger=to_logger, check_period=check_period)
 
 
-
 class RunClientAsync(ActorJobBaseClientAsync):
     """Async sub-client for manipulating a single Actor run."""
 
@@ -648,7 +646,9 @@ async def charge(
         )
 
     async def get_status_message_redirector(
-        self, to_logger: logging.Logger | None = None, check_period: timedelta = timedelta(seconds=1),
+        self,
+        to_logger: logging.Logger | None = None,
+        check_period: timedelta = timedelta(seconds=1),
     ) -> StatusMessageRedirectorAsync:
         """Get `StatusMessageRedirector` instance that can be used to redirect logs.
 

tests/unit/test_logging.py

@@ -14,7 +14,7 @@
 
 from apify_client import ApifyClient, ApifyClientAsync
 from apify_client._logging import RedirectLogFormatter
-from apify_client.clients.resource_clients.log import StreamedLog, StatusMessageRedirector
+from apify_client.clients.resource_clients.log import StatusMessageRedirector, StreamedLog
 
 _MOCKED_API_URL = 'https://example.com'
 _MOCKED_RUN_ID = 'mocked_run_id'
@@ -54,24 +54,60 @@
     ),
 )
 
+_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES = (
+    ('Status: RUNNING, Message: Initial message', logging.INFO),
+    *_EXPECTED_MESSAGES_AND_LEVELS,
+    ('Status: RUNNING, Message: Another message', logging.INFO),
+    ('Status: SUCCEEDED, Message: Final message', logging.INFO),
+)
+
 
 @pytest.fixture
 def mock_api() -> None:
-
-    def create_status_responses_generator():
-        for i in range(10):
+    def create_status_responses_generator() -> Iterator[httpx.Response]:
+        """Simulate actor run that changes status 3 times."""
+        for _ in range(5):
             yield httpx.Response(
                 content=json.dumps(
-                    {'data': {'id': _MOCKED_RUN_ID, 'actId': _MOCKED_ACTOR_ID, 'status': ActorJobStatus.RUNNING,
-                              'statusMessage': f'Status {i}', 'isStatusMessageTerminal': False}}
+                    {
+                        'data': {
+                            'id': _MOCKED_RUN_ID,
+                            'actId': _MOCKED_ACTOR_ID,
+                            'status': ActorJobStatus.RUNNING,
+                            'statusMessage': 'Initial message',
+                            'isStatusMessageTerminal': False,
+                        }
+                    }
+                ),
+                status_code=200,
+            )
+        for _ in range(5):
+            yield httpx.Response(
+                content=json.dumps(
+                    {
+                        'data': {
+                            'id': _MOCKED_RUN_ID,
+                            'actId': _MOCKED_ACTOR_ID,
+                            'status': ActorJobStatus.RUNNING,
+                            'statusMessage': 'Another message',
+                            'isStatusMessageTerminal': False,
+                        }
+                    }
                 ),
                 status_code=200,
             )
         while True:
             yield httpx.Response(
                 content=json.dumps(
-                    {'data': {'id': _MOCKED_RUN_ID, 'actId': _MOCKED_ACTOR_ID, 'status': ActorJobStatus.SUCCEEDED,
-                              'statusMessage': f'Status 101', 'isStatusMessageTerminal': True}}
+                    {
+                        'data': {
+                            'id': _MOCKED_RUN_ID,
+                            'actId': _MOCKED_ACTOR_ID,
+                            'status': ActorJobStatus.SUCCEEDED,
+                            'statusMessage': 'Final message',
+                            'isStatusMessageTerminal': True,
+                        }
+                    }
                 ),
                 status_code=200,
             )
@@ -133,6 +169,15 @@ def propagate_stream_logs() -> None:
     logging.getLogger(f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}').setLevel(logging.DEBUG)
 
 
+@pytest.fixture
+def reduce_final_timeout_for_status_message_redirector() -> None:
+    """Reduce timeout used by the `StatusMessageRedirector`
+
+    This timeout makes sense on the platform, but in tests it is better to reduce it to speed up the tests.
+    """
+    StatusMessageRedirector._final_sleep_time_s = 2
+
+
 @pytest.mark.parametrize(
     ('log_from_start', 'expected_log_count'),
     [
@@ -215,6 +260,7 @@ async def test_actor_call_redirect_logs_to_default_logger_async(
     caplog: LogCaptureFixture,
     mock_api_async: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    reduce_final_timeout_for_status_message_redirector: None,  # noqa: ARG001, fixture
 ) -> None:
     """Test that logs are redirected correctly to the default logger.
 
@@ -231,8 +277,8 @@ async def test_actor_call_redirect_logs_to_default_logger_async(
     assert isinstance(logger.handlers[0], logging.StreamHandler)
 
     # Ensure logs are propagated
-    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
-    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES)
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
 
@@ -242,6 +288,7 @@ def test_actor_call_redirect_logs_to_default_logger_sync(
     caplog: LogCaptureFixture,
     mock_api_sync: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    reduce_final_timeout_for_status_message_redirector: None,  # noqa: ARG001, fixture
 ) -> None:
     """Test that logs are redirected correctly to the default logger.
 
@@ -258,8 +305,8 @@ def test_actor_call_redirect_logs_to_default_logger_sync(
     assert isinstance(logger.handlers[0], logging.StreamHandler)
 
     # Ensure logs are propagated
-    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
-    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES)
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
 
@@ -299,6 +346,7 @@ async def test_actor_call_redirect_logs_to_custom_logger_async(
     caplog: LogCaptureFixture,
     mock_api_async: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    reduce_final_timeout_for_status_message_redirector: None,  # noqa: ARG001, fixture
 ) -> None:
     """Test that logs are redirected correctly to the custom logger."""
     logger_name = 'custom_logger'
@@ -308,8 +356,8 @@ async def test_actor_call_redirect_logs_to_custom_logger_async(
     with caplog.at_level(logging.DEBUG, logger=logger_name):
         await run_client.call(logger=logger)
 
-    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
-    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES)
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
 
@@ -319,6 +367,7 @@ def test_actor_call_redirect_logs_to_custom_logger_sync(
     caplog: LogCaptureFixture,
     mock_api_sync: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    reduce_final_timeout_for_status_message_redirector: None,  # noqa: ARG001, fixture
 ) -> None:
     """Test that logs are redirected correctly to the custom logger."""
     logger_name = 'custom_logger'
@@ -328,51 +377,56 @@ def test_actor_call_redirect_logs_to_custom_logger_sync(
     with caplog.at_level(logging.DEBUG, logger=logger_name):
         run_client.call(logger=logger)
 
-    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
-    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES)
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS_WITH_STATUS_MESSAGES, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
 
+
 @respx.mock
 async def test_redirect_status_message_async(
     *,
     caplog: LogCaptureFixture,
     mock_api: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    reduce_final_timeout_for_status_message_redirector: None,  # noqa: ARG001, fixture
 ) -> None:
     """Test redirected status and status messages."""
 
     run_client = ApifyClientAsync(token='mocked_token', api_url=_MOCKED_API_URL).run(run_id=_MOCKED_RUN_ID)
 
     logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
 
-    status_message_redirector = await run_client.get_status_message_redirector(check_period=timedelta(seconds =0.1))
+    status_message_redirector = await run_client.get_status_message_redirector(check_period=timedelta(seconds=0))
     with caplog.at_level(logging.DEBUG, logger=logger_name):
         async with status_message_redirector:
             # Do stuff while the status from the other actor is being redirected to the logs.
-            await asyncio.sleep(2)
+            await asyncio.sleep(3)
+
+    assert caplog.records[0].message == 'Status: RUNNING, Message: Initial message'
+    assert caplog.records[1].message == 'Status: RUNNING, Message: Another message'
+    assert caplog.records[2].message == 'Status: SUCCEEDED, Message: Final message'
 
-    assert caplog.records[0].message == 'Status: RUNNING, Message: Status 1'
-    assert caplog.records[1].message == 'Status: SUCCEEDED, Message: Status 2'
 
 @respx.mock
 def test_redirect_status_message_sync(
     *,
     caplog: LogCaptureFixture,
     mock_api: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    reduce_final_timeout_for_status_message_redirector: None,  # noqa: ARG001, fixture
 ) -> None:
     """Test redirected status and status messages."""
 
     run_client = ApifyClient(token='mocked_token', api_url=_MOCKED_API_URL).run(run_id=_MOCKED_RUN_ID)
 
     logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
 
-    status_message_redirector = run_client.get_status_message_redirector(check_period=timedelta(seconds =0.1))
-    with caplog.at_level(logging.DEBUG, logger=logger_name):
-        with status_message_redirector:
-            # Do stuff while the status from the other actor is being redirected to the logs.
-            time.sleep(2)
+    status_message_redirector = run_client.get_status_message_redirector(check_period=timedelta(seconds=0))
+    with caplog.at_level(logging.DEBUG, logger=logger_name), status_message_redirector:
+        # Do stuff while the status from the other actor is being redirected to the logs.
+        time.sleep(3)
 
-    assert caplog.records[0].message == 'Status: RUNNING, Message: Status 1'
-    assert caplog.records[1].message == 'Status: SUCCEEDED, Message: Status 2'
+    assert caplog.records[0].message == 'Status: RUNNING, Message: Initial message'
+    assert caplog.records[1].message == 'Status: RUNNING, Message: Another message'
+    assert caplog.records[2].message == 'Status: SUCCEEDED, Message: Final message'