diff --git a/pyproject.toml b/pyproject.toml index 03035383..5a25d73a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -187,7 +187,7 @@ known-local-folder = ["apify_client"] max-branches = 18 [tool.pytest.ini_options] -addopts = "-r a --verbose" +addopts = "-r a --verbose --dist worksteal" asyncio_default_fixture_loop_scope = "function" asyncio_mode = "auto" pythonpath = ["."] diff --git a/tests/integration/test_build.py b/tests/integration/test_build.py index 2f1bd1ce..0da69b36 100644 --- a/tests/integration/test_build.py +++ b/tests/integration/test_build.py @@ -185,17 +185,18 @@ async def test_build_delete_and_abort(client: ApifyClient | ApifyClientAsync) -> actor_client = client.actor(created_actor.id) try: - # Build both versions - we need 2 builds because we can't delete the default build + # Both versions are built because the default build cannot be deleted. `build` returns as soon as the + # build is queued, so starting both before awaiting either lets the platform run them concurrently. first_build = await maybe_await(actor_client.build(version_number='0.1')) assert isinstance(first_build, Build) - first_build_client = client.build(first_build.id) - await maybe_await(first_build_client.wait_for_finish()) - second_build = await maybe_await(actor_client.build(version_number='0.2')) assert isinstance(second_build, Build) + + first_build_client = client.build(first_build.id) second_build_client = client.build(second_build.id) - # Wait for the second build to finish + await maybe_await(first_build_client.wait_for_finish()) + finished_build = await maybe_await(second_build_client.wait_for_finish()) assert isinstance(finished_build, Build) assert finished_build.status in ('SUCCEEDED', 'FAILED') diff --git a/tests/integration/test_request_queue.py b/tests/integration/test_request_queue.py index 33446d7f..692be3c6 100644 --- a/tests/integration/test_request_queue.py +++ b/tests/integration/test_request_queue.py @@ -573,26 +573,23 @@ async def test_request_queue_unlock_requests(client: ApifyClient | ApifyClientAs await ensure_queue_is_populated(rq_client, expected_count=5) - result = await maybe_await(rq_client.list_and_lock_head(limit=3, lock_duration=timedelta(seconds=60))) - assert isinstance(result, LockedRequestQueueHead) - lock_response = result + lock_response = await maybe_await(rq_client.list_and_lock_head(limit=3, lock_duration=timedelta(seconds=60))) + assert isinstance(lock_response, LockedRequestQueueHead) assert len(lock_response.items) == 3 - locked_ids = {item.id for item in lock_response.items} - # Locks are acknowledged before they are visible to subsequent reads, so unlocking immediately can - # see fewer locks than were just acquired. Since locked requests are excluded from the queue head, - # poll `list_head` until the locked IDs disappear from it (best-effort mitigation of the race). - async def all_locks_visible() -> bool: - head = await maybe_await(rq_client.list_head(limit=5)) - assert isinstance(head, RequestQueueHead) - return locked_ids.isdisjoint(item.id for item in head.items) + # Locks are acknowledged before they are all visible to `unlock_requests`, so a single call can report + # fewer locks than were just acquired. Each call unlocks whatever is visible to it, so the counts are + # accumulated until every lock is accounted for. + unlocked_counts: list[int] = [] - await poll_until_condition(all_locks_visible) + async def unlocked_so_far() -> int: + unlock_response = await maybe_await(rq_client.unlock_requests()) + assert isinstance(unlock_response, UnlockRequestsResult) + unlocked_counts.append(unlock_response.unlocked_count) + return sum(unlocked_counts) - # Unlock all requests - unlock_response = await maybe_await(rq_client.unlock_requests()) - assert isinstance(unlock_response, UnlockRequestsResult) - assert unlock_response.unlocked_count == 3 + unlocked_total = await poll_until_condition(unlocked_so_far, lambda total: total == 3) + assert unlocked_total == 3 finally: await maybe_await(rq_client.delete())