TODO: Finalize tests

Author: Pijukatel

src/apify_client/clients/resource_clients/log.py

@@ -390,6 +390,7 @@ 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
 
     def __init__(self, *, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=5)) -> None:
         """Initialize `StatusMessageRedirector`.
@@ -399,9 +400,11 @@ def __init__(self, *, to_logger: logging.Logger, check_period: timedelta = timed
             check_period: The period with which the status message will be polled.
         """
         self._to_logger = to_logger
+        self._to_logger.propagate = self._force_propagate
         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.
 
@@ -420,15 +423,15 @@ 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 run_data.get('isStatusMessageTerminal', True)
+            return not(run_data.get('isStatusMessageTerminal', False))
         return True
 
 
 class StatusMessageRedirectorAsync(StatusMessageRedirector):
     """Async variant of `StatusMessageRedirector` that is logging in task."""
 
     def __init__(
-        self, *, run_client: RunClientAsync, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=5)
+        self, *, run_client: RunClientAsync, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
     ) -> None:
         """Initialize `StatusMessageRedirectorAsync`.
 
@@ -478,7 +481,7 @@ class StatusMessageRedirectorSync(StatusMessageRedirector):
     """Sync variant of `StatusMessageRedirector` that is logging in thread."""
 
     def __init__(
-        self, *, run_client: RunClient, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=5)
+        self, *, run_client: RunClient, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
     ) -> None:
         """Initialize `StatusMessageRedirectorSync`.
 

src/apify_client/clients/resource_clients/run.py

@@ -5,6 +5,7 @@
 import random
 import string
 import time
+from datetime import timedelta
 from typing import TYPE_CHECKING, Any
 
 from apify_shared.utils import filter_out_none_values_recursively, ignore_docs, parse_date_fields
@@ -20,7 +21,7 @@
     StatusMessageRedirectorAsync,
     StatusMessageRedirectorSync,
     StreamedLogAsync,
-    StreamedLogSync,
+    StreamedLogSync, StatusMessageRedirector,
 )
 from apify_client.clients.resource_clients.request_queue import RequestQueueClient, RequestQueueClientAsync
 
@@ -321,14 +322,18 @@ def charge(
             ),
         )
 
-    def get_status_message_logger(self, to_logger: logging.Logger | None = None) -> StatusMessageRedirectorSync:
+
+    def get_status_message_redirector(
+        self, to_logger: logging.Logger | None = None, check_period: timedelta = timedelta(seconds=1)
+    ) -> StatusMessageRedirectorSync:
         """Get `StatusMessageRedirector` instance that can be used to redirect logs.
 
-         `StatusMessageRedirector` can be directly called or used as a context manager.
+        `StatusMessageRedirector` can be directly called or used as a context manager.
 
         Args:
             to_logger: `Logger` used for logging the status and status messages. If not provided, a new logger is
             created.
+            check_period: The period with which the status message will be polled.
 
         Returns:
             `StatusMessageRedirector` instance for redirected logs.
@@ -344,7 +349,8 @@ def get_status_message_logger(self, to_logger: logging.Logger | None = None) ->
             name = '-'.join(part for part in (actor_name, run_id) if part)
             to_logger = create_redirect_logger(f'apify.{name}')
 
-        return StatusMessageRedirectorSync(run_client=self, to_logger=to_logger)
+        return StatusMessageRedirectorSync(run_client=self, to_logger=to_logger, check_period=check_period)
+
 
 
 class RunClientAsync(ActorJobBaseClientAsync):
@@ -642,7 +648,7 @@ async def charge(
         )
 
     async def get_status_message_redirector(
-        self, to_logger: logging.Logger | None = None
+        self, to_logger: logging.Logger | None = None, check_period: timedelta = timedelta(seconds=1),
     ) -> StatusMessageRedirectorAsync:
         """Get `StatusMessageRedirector` instance that can be used to redirect logs.
 
@@ -651,6 +657,7 @@ async def get_status_message_redirector(
         Args:
             to_logger: `Logger` used for logging the status and status messages. If not provided, a new logger is
             created.
+            check_period: The period with which the status message will be polled.
 
         Returns:
             `StatusMessageRedirector` instance for redirected logs.
@@ -666,4 +673,4 @@ async def get_status_message_redirector(
             name = '-'.join(part for part in (actor_name, run_id) if part)
             to_logger = create_redirect_logger(f'apify.{name}')
 
-        return StatusMessageRedirectorAsync(run_client=self, to_logger=to_logger)
+        return StatusMessageRedirectorAsync(run_client=self, to_logger=to_logger, check_period=check_period)

tests/unit/test_logging.py

@@ -3,7 +3,7 @@
 import logging
 import time
 from collections.abc import AsyncIterator, Iterator
-from datetime import datetime
+from datetime import datetime, timedelta
 from unittest.mock import patch
 
 import httpx
@@ -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
+from apify_client.clients.resource_clients.log import StreamedLog, StatusMessageRedirector
 
 _MOCKED_API_URL = 'https://example.com'
 _MOCKED_RUN_ID = 'mocked_run_id'
@@ -57,32 +57,30 @@
 
 @pytest.fixture
 def mock_api() -> None:
-    actor_runs_responses = iter(
-        (
-            httpx.Response(
-                content=json.dumps(
-                    {'data': {'id': _MOCKED_RUN_ID, 'actId': _MOCKED_ACTOR_ID, 'status': ActorJobStatus.RUNNING}}
-                ),
-                status_code=200,
-            ),
-            httpx.Response(
+
+    def create_status_responses_generator():
+        for i in range(10):
+            yield httpx.Response(
                 content=json.dumps(
-                    {'data': {'id': _MOCKED_RUN_ID, 'actId': _MOCKED_ACTOR_ID, 'status': ActorJobStatus.RUNNING}}
+                    {'data': {'id': _MOCKED_RUN_ID, 'actId': _MOCKED_ACTOR_ID, 'status': ActorJobStatus.RUNNING,
+                              'statusMessage': f'Status {i}', 'isStatusMessageTerminal': False}}
                 ),
                 status_code=200,
-            ),
-            httpx.Response(
+            )
+        while True:
+            yield httpx.Response(
                 content=json.dumps(
-                    {'data': {'id': _MOCKED_RUN_ID, 'actId': _MOCKED_ACTOR_ID, 'status': ActorJobStatus.SUCCEEDED}}
+                    {'data': {'id': _MOCKED_RUN_ID, 'actId': _MOCKED_ACTOR_ID, 'status': ActorJobStatus.SUCCEEDED,
+                              'statusMessage': f'Status 101', 'isStatusMessageTerminal': True}}
                 ),
                 status_code=200,
-            ),
-        )
-    )
+            )
+
+    response_generator = create_status_responses_generator()
 
     def actor_runs_side_effect(_: httpx.Request) -> httpx.Response:
         time.sleep(0.1)
-        return next(actor_runs_responses)
+        return next(response_generator)
 
     respx.get(url=f'{_MOCKED_API_URL}/v2/actor-runs/{_MOCKED_RUN_ID}').mock(side_effect=actor_runs_side_effect)
 
@@ -129,7 +127,9 @@ def close(self) -> None:
 
 @pytest.fixture
 def propagate_stream_logs() -> None:
-    StreamedLog._force_propagate = True  # Enable propagation of logs to the caplog fixture
+    # Enable propagation of logs to the caplog fixture
+    StreamedLog._force_propagate = True
+    StatusMessageRedirector._force_propagate = True
     logging.getLogger(f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}').setLevel(logging.DEBUG)
 
 
@@ -332,3 +332,47 @@ def test_actor_call_redirect_logs_to_custom_logger_sync(
     for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, 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
+) -> 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))
+    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)
+
+    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
+) -> 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)
+
+    assert caplog.records[0].message == 'Status: RUNNING, Message: Status 1'
+    assert caplog.records[1].message == 'Status: SUCCEEDED, Message: Status 2'