Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions src/apify_client/_resource_clients/actor.py
Original file line number Diff line number Diff line change
Expand Up @@ -386,7 +386,7 @@ def call(
default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
will redirect logs to the provided logger. The logger is also used to capture status and status message
of the other Actor run.
timeout: Timeout for the API HTTP request.
timeout: Timeout for each API HTTP request, both the start and every poll.

Returns:
The run object.
Expand Down Expand Up @@ -414,7 +414,7 @@ def call(
)

if not logger:
return run_client.wait_for_finish(wait_duration=wait_duration)
return run_client.wait_for_finish(wait_duration=wait_duration, timeout=timeout)

if logger == 'default':
logger = None
Expand All @@ -426,7 +426,7 @@ def call(
streamed_log = run_client.get_streamed_log(to_logger=logger)

with status_redirector, streamed_log:
return run_client.wait_for_finish(wait_duration=wait_duration)
return run_client.wait_for_finish(wait_duration=wait_duration, timeout=timeout)

def build(
self,
Expand Down Expand Up @@ -946,7 +946,7 @@ async def call(
default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
will redirect logs to the provided logger. The logger is also used to capture status and status message
of the other Actor run.
timeout: Timeout for the API HTTP request.
timeout: Timeout for each API HTTP request, both the start and every poll.

Returns:
The run object.
Expand Down Expand Up @@ -975,7 +975,7 @@ async def call(
)

if not logger:
return await run_client.wait_for_finish(wait_duration=wait_duration)
return await run_client.wait_for_finish(wait_duration=wait_duration, timeout=timeout)

if logger == 'default':
logger = None
Expand All @@ -987,7 +987,7 @@ async def call(
streamed_log = await run_client.get_streamed_log(to_logger=logger)

async with status_redirector, streamed_log:
return await run_client.wait_for_finish(wait_duration=wait_duration)
return await run_client.wait_for_finish(wait_duration=wait_duration, timeout=timeout)

async def build(
self,
Expand Down
8 changes: 4 additions & 4 deletions src/apify_client/_resource_clients/task.py
Original file line number Diff line number Diff line change
Expand Up @@ -347,7 +347,7 @@ def call(
than the whole memory limit of the account is rejected with `actor-memory-limit-exceeded` as well and
never starts, so `True` retries it forever. The time spent retrying doesn't count toward
`wait_duration`.
timeout: Timeout for the API HTTP request.
timeout: Timeout for each API HTTP request, both the start and every poll.

Returns:
The run object.
Expand All @@ -372,7 +372,7 @@ def call(
http_client=self._http_client,
client_registry=self._client_registry,
)
return run_client.wait_for_finish(wait_duration=wait_duration)
return run_client.wait_for_finish(wait_duration=wait_duration, timeout=timeout)

def get_input(self, *, timeout: Timeout = 'short') -> dict:
"""Retrieve the default input for this task.
Expand Down Expand Up @@ -761,7 +761,7 @@ async def call(
than the whole memory limit of the account is rejected with `actor-memory-limit-exceeded` as well and
never starts, so `True` retries it forever. The time spent retrying doesn't count toward
`wait_duration`.
timeout: Timeout for the API HTTP request.
timeout: Timeout for each API HTTP request, both the start and every poll.

Returns:
The run object.
Expand All @@ -785,7 +785,7 @@ async def call(
http_client=self._http_client,
client_registry=self._client_registry,
)
return await run_client.wait_for_finish(wait_duration=wait_duration)
return await run_client.wait_for_finish(wait_duration=wait_duration, timeout=timeout)

async def get_input(self, *, timeout: Timeout = 'short') -> dict:
"""Retrieve the default input for this task.
Expand Down
76 changes: 76 additions & 0 deletions tests/unit/test_call_poll_timeout.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
from __future__ import annotations

from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock, Mock

import pytest

from apify_client import ApifyClient, ApifyClientAsync


@pytest.mark.parametrize('resource', [pytest.param('actor', id='actor'), pytest.param('task', id='task')])
def test_call_passes_timeout_to_poll_sync(monkeypatch: pytest.MonkeyPatch, resource: str) -> None:
"""Apply the per-request timeout to both the start and poll requests."""
client = ApifyClient(token='test_token')
sub_client = getattr(client, resource)('resource-id')
start = Mock(return_value=Mock(id='run-id'))
poll = Mock(return_value=Mock(id='run-id'))
monkeypatch.setattr(sub_client, 'start', start)
monkeypatch.setattr(sub_client._client_registry, 'run_client', Mock(return_value=Mock(wait_for_finish=poll)))

timeout = timedelta(seconds=90)
kwargs = {'logger': None} if resource == 'actor' else {}
sub_client.call(timeout=timeout, wait_duration=timedelta(seconds=120), **kwargs)

assert start.call_args.kwargs['timeout'] == timeout
poll.assert_called_once_with(wait_duration=timedelta(seconds=120), timeout=timeout)


@pytest.mark.parametrize('resource', [pytest.param('actor', id='actor'), pytest.param('task', id='task')])
async def test_call_passes_timeout_to_poll_async(monkeypatch: pytest.MonkeyPatch, resource: str) -> None:
"""Apply the per-request timeout to both async start and poll requests."""
client = ApifyClientAsync(token='test_token')
sub_client = getattr(client, resource)('resource-id')
start = AsyncMock(return_value=Mock(id='run-id'))
poll = AsyncMock(return_value=Mock(id='run-id'))
monkeypatch.setattr(sub_client, 'start', start)
monkeypatch.setattr(sub_client._client_registry, 'run_client', Mock(return_value=Mock(wait_for_finish=poll)))

timeout = timedelta(seconds=90)
kwargs = {'logger': None} if resource == 'actor' else {}
await sub_client.call(timeout=timeout, wait_duration=timedelta(seconds=120), **kwargs)

assert start.call_args.kwargs['timeout'] == timeout
poll.assert_awaited_once_with(wait_duration=timedelta(seconds=120), timeout=timeout)


def test_actor_call_passes_timeout_to_poll_with_log_redirect(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep the poll timeout when the default log redirect is active."""
actor = ApifyClient(token='test_token').actor('actor-id')
monkeypatch.setattr(actor, 'start', Mock(return_value=Mock(id='run-id')))
poll = Mock()
run_client = Mock(wait_for_finish=poll)
run_client.get_status_message_watcher.return_value = MagicMock()
run_client.get_streamed_log.return_value = MagicMock()
monkeypatch.setattr(actor._client_registry, 'run_client', Mock(return_value=run_client))

timeout = timedelta(seconds=90)
actor.call(timeout=timeout)

poll.assert_called_once_with(wait_duration=None, timeout=timeout)


async def test_actor_call_passes_timeout_to_poll_with_log_redirect_async(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep the async poll timeout when the default log redirect is active."""
actor = ApifyClientAsync(token='test_token').actor('actor-id')
monkeypatch.setattr(actor, 'start', AsyncMock(return_value=Mock(id='run-id')))
poll = AsyncMock()
run_client = Mock(wait_for_finish=poll)
run_client.get_status_message_watcher = AsyncMock(return_value=MagicMock())
run_client.get_streamed_log = AsyncMock(return_value=MagicMock())
monkeypatch.setattr(actor._client_registry, 'run_client', Mock(return_value=run_client))

timeout = timedelta(seconds=90)
await actor.call(timeout=timeout)

poll.assert_awaited_once_with(wait_duration=None, timeout=timeout)
Loading