fix(tests): poll for logs in smoke tests to handle CloudWatch ingestion latency (#774)

Author: sid-rl

tests/smoketests/sdk/test_async_devbox.py

@@ -11,6 +11,7 @@
 from runloop_api_client.sdk import AsyncDevbox, AsyncRunloopSDK
 from tests.smoketests.utils import unique_name
 from runloop_api_client.lib.polling import PollingConfig
+from runloop_api_client.lib.polling_async import async_poll_until
 
 pytestmark = [pytest.mark.smoketest, pytest.mark.asyncio]
 
@@ -1042,18 +1043,20 @@ class TestAsyncDevboxLogs:
 
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     async def test_logs_basic(self, shared_devbox: AsyncDevbox) -> None:
-        """Test retrieving devbox logs returns valid response structure."""
+        """Test retrieving unfiltered devbox logs."""
         test_message = "async basic log test message"
         result = await shared_devbox.cmd.exec(f'echo "{test_message}"')
         assert result.exit_code == 0
 
-        logs = await shared_devbox.logs()
-
-        assert logs is not None
-        assert hasattr(logs, "logs")
-        assert isinstance(logs.logs, list)
-        log_content = " ".join(str(log) for log in logs.logs)
-        assert test_message in log_content
+        # Log ingestion is async — logs may not be queryable immediately after
+        # command execution. Poll every 1s with a 10s timeout (well within the
+        # 30s test timeout) to accommodate variable ingestion latency.
+        logs = await async_poll_until(
+            retriever=lambda: shared_devbox.logs(),
+            is_terminal=lambda l: any(test_message in (log.message or "") for log in l.logs),
+            config=PollingConfig(timeout_seconds=10, interval_seconds=1),
+        )
+        assert any(test_message in (log.message or "") for log in logs.logs)
 
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     async def test_logs_with_execution_filter(self, shared_devbox: AsyncDevbox) -> None:
@@ -1062,13 +1065,12 @@ async def test_logs_with_execution_filter(self, shared_devbox: AsyncDevbox) -> N
         result = await shared_devbox.cmd.exec(f'echo "{test_message}"')
         assert result.exit_code == 0
 
-        logs = await shared_devbox.logs(execution_id=result.execution_id)
-
-        assert logs is not None
-        assert hasattr(logs, "logs")
-        assert isinstance(logs.logs, list)
-        log_content = " ".join(str(log) for log in logs.logs)
-        assert test_message in log_content
+        logs = await async_poll_until(
+            retriever=lambda: shared_devbox.logs(execution_id=result.execution_id),
+            is_terminal=lambda l: any(test_message in (log.message or "") for log in l.logs),
+            config=PollingConfig(timeout_seconds=10, interval_seconds=1),
+        )
+        assert any(test_message in (log.message or "") for log in logs.logs)
 
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     async def test_logs_with_shell_name_filter(self, shared_devbox: AsyncDevbox) -> None:
@@ -1080,10 +1082,9 @@ async def test_logs_with_shell_name_filter(self, shared_devbox: AsyncDevbox) ->
         result = await shell.exec(f'echo "{test_message}"')
         assert result.exit_code == 0
 
-        logs = await shared_devbox.logs(shell_name=shell_name)
-
-        assert logs is not None
-        assert hasattr(logs, "logs")
-        assert isinstance(logs.logs, list)
-        log_content = " ".join(str(log) for log in logs.logs)
-        assert test_message in log_content
+        logs = await async_poll_until(
+            retriever=lambda: shared_devbox.logs(shell_name=shell_name),
+            is_terminal=lambda l: any(test_message in (log.message or "") for log in l.logs),
+            config=PollingConfig(timeout_seconds=10, interval_seconds=1),
+        )
+        assert any(test_message in (log.message or "") for log in logs.logs)

tests/smoketests/sdk/test_devbox.py

@@ -10,7 +10,7 @@
 
 from runloop_api_client.sdk import Devbox, RunloopSDK
 from tests.smoketests.utils import unique_name
-from runloop_api_client.lib.polling import PollingConfig
+from runloop_api_client.lib.polling import PollingConfig, poll_until
 
 pytestmark = [pytest.mark.smoketest]
 
@@ -1028,18 +1028,20 @@ class TestDevboxLogs:
 
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     def test_logs_basic(self, shared_devbox: Devbox) -> None:
-        """Test retrieving devbox logs returns valid response structure."""
+        """Test retrieving unfiltered devbox logs."""
         test_message = "basic log test message"
         result = shared_devbox.cmd.exec(f'echo "{test_message}"')
         assert result.exit_code == 0
 
-        logs = shared_devbox.logs()
-
-        assert logs is not None
-        assert hasattr(logs, "logs")
-        assert isinstance(logs.logs, list)
-        log_content = " ".join(str(log) for log in logs.logs)
-        assert test_message in log_content
+        # Log ingestion is async — logs may not be queryable immediately after
+        # command execution. Poll every 1s with a 10s timeout (well within the
+        # 30s test timeout) to accommodate variable ingestion latency.
+        logs = poll_until(
+            retriever=lambda: shared_devbox.logs(),
+            is_terminal=lambda l: any(test_message in (log.message or "") for log in l.logs),
+            config=PollingConfig(timeout_seconds=10, interval_seconds=1),
+        )
+        assert any(test_message in (log.message or "") for log in logs.logs)
 
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     def test_logs_with_execution_filter(self, shared_devbox: Devbox) -> None:
@@ -1048,13 +1050,12 @@ def test_logs_with_execution_filter(self, shared_devbox: Devbox) -> None:
         result = shared_devbox.cmd.exec(f'echo "{test_message}"')
         assert result.exit_code == 0
 
-        logs = shared_devbox.logs(execution_id=result.execution_id)
-
-        assert logs is not None
-        assert hasattr(logs, "logs")
-        assert isinstance(logs.logs, list)
-        log_content = " ".join(str(log) for log in logs.logs)
-        assert test_message in log_content
+        logs = poll_until(
+            retriever=lambda: shared_devbox.logs(execution_id=result.execution_id),
+            is_terminal=lambda l: any(test_message in (log.message or "") for log in l.logs),
+            config=PollingConfig(timeout_seconds=10, interval_seconds=1),
+        )
+        assert any(test_message in (log.message or "") for log in logs.logs)
 
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     def test_logs_with_shell_name_filter(self, shared_devbox: Devbox) -> None:
@@ -1066,10 +1067,9 @@ def test_logs_with_shell_name_filter(self, shared_devbox: Devbox) -> None:
         result = shell.exec(f'echo "{test_message}"')
         assert result.exit_code == 0
 
-        logs = shared_devbox.logs(shell_name=shell_name)
-
-        assert logs is not None
-        assert hasattr(logs, "logs")
-        assert isinstance(logs.logs, list)
-        log_content = " ".join(str(log) for log in logs.logs)
-        assert test_message in log_content
+        logs = poll_until(
+            retriever=lambda: shared_devbox.logs(shell_name=shell_name),
+            is_terminal=lambda l: any(test_message in (log.message or "") for log in l.logs),
+            config=PollingConfig(timeout_seconds=10, interval_seconds=1),
+        )
+        assert any(test_message in (log.message or "") for log in logs.logs)