add delay measurement

Author: shroominic

tests/test_v02.py

@@ -1,4 +1,5 @@
 import os
+import time
 
 import pytest
 from codeboxapi import CodeBox, ExecChunk, ExecResult, RemoteFile
@@ -157,65 +158,131 @@ async def test_async_list_operations(codebox: CodeBox):
 
 
 def test_sync_stream_exec(codebox: CodeBox):
-    chunks = list(
-        codebox.stream_exec(
-            "import time;\nfor i in range(3): time.sleep(0.01); print(i)"
-        )
-    )
+    chunks: list[tuple[ExecChunk, float]] = []
+    t0 = time.perf_counter()
+    for chunk in codebox.stream_exec(
+        "import time;\nfor i in range(3): time.sleep(0.01); print(i)"
+    ):
+        chunks.append((chunk, time.perf_counter() - t0))
+
     assert (
         len(chunks) == 3
     ), "iterating over stream_exec should produce 3 chunks (ipython)"
     assert all(
-        isinstance(chunk, ExecChunk) for chunk in chunks
+        isinstance(chunk[0], ExecChunk) for chunk in chunks
     ), "All items should be ExecChunk instances (ipython)"
     assert all(
-        chunk.type == "txt" for chunk in chunks
+        chunk[0].type == "txt" for chunk in chunks
     ), "All chunks should be of type 'txt' (ipython)"
-    assert [chunk.content.strip() for chunk in chunks] == [
+    assert [chunk[0].content.strip() for chunk in chunks] == [
         "0",
         "1",
         "2",
     ], "Chunks should contain correct content (ipython)"
-    chunks = list(
-        codebox.stream_exec(
-            "python -c 'import time\nfor i in range(3): time.sleep(0.01); print(i)'",
-            kernel="bash",
-        )
-    )
+    # Verify chunks arrive with delay
+    assert all(
+        chunks[i][1] < chunks[i + 1][1] for i in range(len(chunks) - 1)
+    ), "Chunks should arrive with delay (ipython)"
+    # Verify delay is approximately 0.01s
+    assert all(
+        abs(chunks[i + 1][1] - chunks[i][1] - 0.01) < 0.005
+        for i in range(len(chunks) - 1)
+    ), "Delay between chunks should be approximately 0.01s (ipython)"
+
+    chunks = []
+    t0 = time.perf_counter()
+    for chunk in codebox.stream_exec(
+        "python -u -c 'import time\nfor i in range(3): time.sleep(0.01); print(i)'",
+        kernel="bash",
+    ):
+        chunks.append((chunk, time.perf_counter() - t0))
+
     assert len(chunks) == 3, "iterating over stream_exec should produce 3 chunks (bash)"
     assert all(
-        isinstance(chunk, ExecChunk) for chunk in chunks
+        isinstance(chunk[0], ExecChunk) for chunk in chunks
     ), "All items should be ExecChunk instances (bash)"
     assert all(
-        chunk.type == "txt" for chunk in chunks
+        chunk[0].type == "txt" for chunk in chunks
     ), "All chunks should be of type 'txt' (bash)"
-    assert [chunk.content.strip() for chunk in chunks] == [
+    assert [chunk[0].content.strip() for chunk in chunks] == [
         "0",
         "1",
         "2",
     ], "Chunks should contain correct content (bash)"
+    # Verify chunks arrive with delay
+    assert all(
+        chunks[i][1] < chunks[i + 1][1] for i in range(len(chunks) - 1)
+    ), "Chunks should arrive with delay (bash)"
+    # Verify delay is approximately 0.01s
+    assert all(
+        abs(chunks[i + 1][1] - chunks[i][1] - 0.01) < 0.005
+        for i in range(len(chunks) - 1)
+    ), "Delay between chunks should be approximately 0.01s (bash)"
 
 
 @pytest.mark.asyncio
 async def test_async_stream_exec(codebox: CodeBox):
-    chunks = [
-        chunk
-        async for chunk in codebox.astream_exec(
-            "import time;\nfor i in range(3): time.sleep(0.01); print(i)"
-        )
-    ]
-    assert len(chunks) == 3, "Stream should produce 3 chunks"
+    chunks: list[tuple[ExecChunk, float]] = []
+    t0 = time.perf_counter()
+    async for chunk in codebox.astream_exec(
+        "import time;\nfor i in range(3): time.sleep(0.01); print(i)"
+    ):
+        chunks.append((chunk, time.perf_counter() - t0))
+
+    assert (
+        len(chunks) == 3
+    ), "iterating over stream_exec should produce 3 chunks (ipython)"
+    assert all(
+        isinstance(chunk[0], ExecChunk) for chunk in chunks
+    ), "All items should be ExecChunk instances (ipython)"
+    assert all(
+        chunk[0].type == "txt" for chunk in chunks
+    ), "All chunks should be of type 'txt' (ipython)"
+    assert [chunk[0].content.strip() for chunk in chunks] == [
+        "0",
+        "1",
+        "2",
+    ], "Chunks should contain correct content (ipython)"
+    # Verify chunks arrive with delay
     assert all(
-        isinstance(chunk, ExecChunk) for chunk in chunks
-    ), "All items should be ExecChunk instances"
+        chunks[i][1] < chunks[i + 1][1] for i in range(len(chunks) - 1)
+    ), "Chunks should arrive with delay (ipython)"
+    # Verify delay is approximately 0.03s
+
     assert all(
-        chunk.type == "txt" for chunk in chunks
-    ), "All chunks should be of type 'txt'"
-    assert [chunk.content.strip() for chunk in chunks] == [
+        abs(chunks[i + 1][1] - chunks[i][1] - 0.01) < 0.005
+        for i in range(len(chunks) - 1)
+    ), "Delay between chunks should be approximately 0.01s (ipython)"
+
+    chunks = []
+    t0 = time.perf_counter()
+    async for chunk in codebox.astream_exec(
+        "python -u -c 'import time\nfor i in range(3): time.sleep(0.01); print(i)'",
+        kernel="bash",
+    ):
+        chunks.append((chunk, time.perf_counter() - t0))
+
+    assert len(chunks) == 3, "iterating over stream_exec should produce 3 chunks (bash)"
+    assert all(
+        isinstance(chunk[0], ExecChunk) for chunk in chunks
+    ), "All items should be ExecChunk instances (bash)"
+    assert all(
+        chunk[0].type == "txt" for chunk in chunks
+    ), "All chunks should be of type 'txt' (bash)"
+    assert [chunk[0].content.strip() for chunk in chunks] == [
         "0",
         "1",
         "2",
-    ], "Chunks should contain correct content"
+    ], "Chunks should contain correct content (bash)"
+    # Verify chunks arrive with delay
+    assert all(
+        chunks[i][1] < chunks[i + 1][1] for i in range(len(chunks) - 1)
+    ), "Chunks should arrive with delay (bash)"
+    # Verify delay is approximately 0.01s
+    assert all(
+        abs(chunks[i + 1][1] - chunks[i][1] - 0.01) < 0.005
+        for i in range(len(chunks) - 1)
+    ), "Delay between chunks should be approximately 0.01s (bash)"
 
 
 def test_sync_error_handling(codebox: CodeBox):