From 61a9749c9be2f336384127a22c53a5bbf5808910 Mon Sep 17 00:00:00 2001 From: Alex Date: Thu, 25 Jun 2026 15:09:11 +0100 Subject: [PATCH] Drop live-service e2e tests; defer to a dedicated e2e PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the integration tests that launch a real Jupyter Kernel Gateway or hit Daytona (artifact_generator / code_executor / workflow-code-node e2e, the gateway integration test, and the Daytona live test) — these belong in a separate e2e PR with proper harnessing. The ephemeral-Postgres integration tests (artifacts repository, run-scoped + input-document authz) stay: they run in CI without any external service and hold the cross-tenant coverage. Trim the now-unused test-only deps (jupyter-kernel-gateway, ipykernel, websocket-client); keep jupyter-client for the retained kernelspec-resolution test. --- .../test_artifact_generator_e2e.py | 398 ------------------ tests/integration/test_code_executor_e2e.py | 279 ------------ .../test_workflow_code_node_e2e.py | 280 ------------ tests/requirements.txt | 15 +- tests/sandbox/test_daytona_live.py | 50 --- .../test_jupyter_gateway_integration.py | 294 ------------- 6 files changed, 4 insertions(+), 1312 deletions(-) delete mode 100644 tests/integration/test_artifact_generator_e2e.py delete mode 100644 tests/integration/test_code_executor_e2e.py delete mode 100644 tests/integration/test_workflow_code_node_e2e.py delete mode 100644 tests/sandbox/test_daytona_live.py delete mode 100644 tests/sandbox/test_jupyter_gateway_integration.py diff --git a/tests/integration/test_artifact_generator_e2e.py b/tests/integration/test_artifact_generator_e2e.py deleted file mode 100644 index 25987d8c..00000000 --- a/tests/integration/test_artifact_generator_e2e.py +++ /dev/null @@ -1,398 +0,0 @@ -"""End-to-end ArtifactGeneratorTool: live Jupyter gateway + ephemeral Postgres + local storage. - -Launches a real ``jupyter kernelgateway`` (no Docker — the credential helper -hangs on dev machines), wires the tool to the ephemeral pytest-postgresql DB and -a temp-dir ``LocalStorage``, and drives create/edit/rewrite through the full -path: a spec is rendered in the kernel by the FIXED renderer, the produced file -is persisted as an artifact version (spec = source of truth), and the rendered -bytes are re-opened with the real library to assert structure. - -Skips gracefully when the gateway binary, websocket-client, or a renderer -library is unavailable. -""" - -from __future__ import annotations - -import io -import shutil -import socket -import subprocess -import time -import uuid - -import pytest - -requests = pytest.importorskip("requests") -pytest.importorskip("websocket") # websocket-client -pytest.importorskip("pptx") -pytest.importorskip("docx") -pytest.importorskip("openpyxl") - -from application.agents.tools.artifact_generator import ArtifactGeneratorTool # noqa: E402 -from application.sandbox.jupyter_gateway import JupyterKernelGatewaySandbox # noqa: E402 -from application.sandbox.manager import SandboxManager # noqa: E402 -from application.sandbox.sandbox_creator import SandboxCreator # noqa: E402 -from application.storage.db.repositories.artifacts import ArtifactsRepository # noqa: E402 -from application.storage.local import LocalStorage # noqa: E402 -from application.storage.storage_creator import StorageCreator # noqa: E402 - -_GATEWAY_BIN = shutil.which("jupyter-kernelgateway") or shutil.which("jupyter") - -pytestmark = [ - pytest.mark.integration, - pytest.mark.skipif( - _GATEWAY_BIN is None, - reason="jupyter kernel gateway not installed (pip install jupyter-kernel-gateway)", - ), -] - - -def _free_port() -> int: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.bind(("127.0.0.1", 0)) - return sock.getsockname()[1] - - -def _gateway_cmd(port: int) -> list: - if _GATEWAY_BIN.endswith("jupyter-kernelgateway"): - base = [_GATEWAY_BIN] - else: - base = [_GATEWAY_BIN, "kernelgateway"] - return base + [ - "--KernelGatewayApp.ip=127.0.0.1", - f"--KernelGatewayApp.port={port}", - "--ZMQChannelsWebsocketConnection.limit_rate=False", - ] - - -@pytest.fixture(scope="module") -def gateway_url(): - port = _free_port() - proc = subprocess.Popen(_gateway_cmd(port), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - url = f"http://127.0.0.1:{port}" - deadline = time.time() + 30 - ready = False - try: - while time.time() < deadline: - if proc.poll() is not None: - pytest.skip("jupyter kernelgateway process exited during startup") - try: - if requests.get(f"{url}/api", timeout=1).status_code == 200: - ready = True - break - except requests.RequestException: - time.sleep(0.3) - if not ready: - pytest.skip("jupyter kernelgateway did not become ready in time") - yield url - finally: - proc.terminate() - try: - proc.wait(timeout=10) - except subprocess.TimeoutExpired: - proc.kill() - - -@pytest.fixture() -def wired_tool(gateway_url, pg_engine, tmp_path, monkeypatch): - """An ArtifactGeneratorTool wired to the live gateway, ephemeral PG, and a temp local store.""" - backend = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=60.0) - SandboxCreator._instance = SandboxManager(backend=backend, max_ttl=1200.0) - - storage = LocalStorage(base_dir=str(tmp_path)) - monkeypatch.setattr(StorageCreator, "_instance", storage, raising=False) - - monkeypatch.setattr("application.storage.db.session.get_engine", lambda: pg_engine) - - conversation_id = str(uuid.uuid4()) - tool = ArtifactGeneratorTool( - tool_config={"conversation_id": conversation_id, "tool_id": str(uuid.uuid4())}, - user_id="user-e2e", - ) - try: - yield tool, conversation_id, pg_engine, storage - finally: - SandboxCreator.reset() - - -def test_create_edit_rewrite_presentation_versions(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - from pptx import Presentation - - # create -> v1 - spec = {"title": "Q3", "slides": [{"title": "Intro", "bullets": ["one", "two"]}]} - created = tool.execute_action("create_artifact", kind="presentation", title="Deck", spec=spec) - assert created["status"] == "ok", created - assert created["version"] == 1 - assert created["mime_type"].endswith("presentationml.presentation") - assert created["filename"] == "Deck.pptx" - artifact_id = created["artifact_id"] - assert tool.get_artifact_id("create_artifact") == artifact_id - - # v1 spec is the source of truth; the rendered .pptx re-opens with one slide. - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(artifact_id, conversation_id=conversation_id) - v1 = repo.get_version(artifact_id, 1) - assert artifact["current_version"] == 1 - assert v1["spec"] == spec - assert v1["produced_by"]["spec_kind"] == "presentation" - prs = Presentation(storage.get_file(v1["storage_path"])) - assert len(prs.slides) == 1 - assert prs.slides[0].shapes.title.text == "Intro" - - # edit (merge-patch) -> v2: add a second slide, keep title. - patch = {"slides": [{"title": "Intro", "bullets": ["one", "two"]}, {"title": "Outro"}]} - edited = tool.execute_action("edit_artifact", id=artifact_id, spec_patch=patch) - assert edited["status"] == "ok", edited - assert edited["version"] == 2 - assert edited["artifact_id"] == artifact_id - - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(artifact_id, conversation_id=conversation_id) - v1_again = repo.get_version(artifact_id, 1) - v2 = repo.get_version(artifact_id, 2) - assert artifact["current_version"] == 2 - # Append-only: v1 unchanged and still present. - assert v1_again["spec"] == spec - assert v2["spec"]["title"] == "Q3" - assert len(v2["spec"]["slides"]) == 2 - prs2 = Presentation(storage.get_file(v2["storage_path"])) - assert len(prs2.slides) == 2 - - # rewrite -> v3: wholesale new spec. - new_spec = {"title": "Fresh", "slides": [{"title": "A"}, {"title": "B"}, {"title": "C"}]} - rewritten = tool.execute_action("rewrite_artifact", id=artifact_id, spec=new_spec) - assert rewritten["status"] == "ok", rewritten - assert rewritten["version"] == 3 - - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(artifact_id, conversation_id=conversation_id) - versions = repo.list_versions(artifact_id) - v3 = repo.get_version(artifact_id, 3) - assert artifact["current_version"] == 3 - assert [v["version"] for v in versions] == [1, 2, 3] - assert v3["spec"] == new_spec - assert len(Presentation(storage.get_file(v3["storage_path"])).slides) == 3 - - -def test_create_document_docx(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - from docx import Document - - spec = {"title": "Report", "sections": [{"heading": "Intro", "paragraphs": ["Hello.", "World."]}]} - created = tool.execute_action("create_artifact", kind="document", spec=spec) - assert created["status"] == "ok", created - assert created["mime_type"].endswith("wordprocessingml.document") - - with pg_engine.connect() as conn: - version = ArtifactsRepository(conn).get_version(created["artifact_id"], 1) - doc = Document(storage.get_file(version["storage_path"])) - texts = [p.text for p in doc.paragraphs] - assert "Hello." in texts and "World." in texts - - -def test_create_spreadsheet_xlsx(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - from openpyxl import load_workbook - - spec = {"sheets": [{"name": "Data", "rows": [["a", "b"], [1, 2]]}]} - created = tool.execute_action("create_artifact", kind="spreadsheet", spec=spec) - assert created["status"] == "ok", created - assert created["mime_type"].endswith("spreadsheetml.sheet") - - with pg_engine.connect() as conn: - version = ArtifactsRepository(conn).get_version(created["artifact_id"], 1) - wb = load_workbook(storage.get_file(version["storage_path"])) - ws = wb["Data"] - assert ws.cell(row=1, column=1).value == "a" - assert ws.cell(row=2, column=2).value == 2 - - -def test_create_edit_html_versions(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - # create -> v1: an html report rendered to report.html (kind html, text/html). - spec = { - "title": "Findings ", - "blocks": [ - {"type": "heading", "text": "Overview", "level": 2}, - {"type": "paragraph", "text": "Body ."}, - {"type": "list", "ordered": False, "items": ["one", "two"]}, - {"type": "table", "headers": ["k", "v"], "rows": [["a", "1"]]}, - ], - } - created = tool.execute_action("create_artifact", kind="html", title="Report", spec=spec) - assert created["status"] == "ok", created - assert created["version"] == 1 - assert created["mime_type"] == "text/html" - assert created["filename"] == "report.html" - assert created["ref"] == "A1" - artifact_id = created["artifact_id"] - - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(artifact_id, conversation_id=conversation_id) - v1 = repo.get_version(artifact_id, 1) - assert artifact["kind"] == "html" - assert artifact["current_version"] == 1 - assert v1["spec"] == spec - assert v1["produced_by"]["spec_kind"] == "html" - - # The stored bytes are a valid HTML doc with escaped title/blocks (no live tag). - html_doc = storage.get_file(v1["storage_path"]).read().decode("utf-8") - assert html_doc.startswith("") - assert "Findings <hi>" in html_doc - assert "<script>alert(1)</script>" in html_doc - assert "" not in html_doc - assert "
  • one
    • " in html_doc and "a" in html_doc - - # edit (merge-patch) -> v2: replace the title, keep v1 intact. - edited = tool.execute_action("edit_artifact", id="A1", spec_patch={"title": "Updated"}) - assert edited["status"] == "ok", edited - assert edited["version"] == 2 - assert edited["artifact_id"] == artifact_id - assert edited["ref"] == "A1" - - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(artifact_id, conversation_id=conversation_id) - v1_again = repo.get_version(artifact_id, 1) - v2 = repo.get_version(artifact_id, 2) - assert artifact["current_version"] == 2 - # Append-only: v1 unchanged; v2 carries the merged spec re-rendered. - assert v1_again["spec"] == spec - assert v2["spec"]["title"] == "Updated" - assert len(v2["spec"]["blocks"]) == 4 - v2_doc = storage.get_file(v2["storage_path"]).read().decode("utf-8") - assert "

    Updated

    " in v2_doc - - -def test_invalid_spec_creates_no_artifact(wired_tool): - tool, conversation_id, pg_engine, _storage = wired_tool - - out = tool.execute_action("create_artifact", kind="presentation", spec={"title": "no slides"}) - assert out["status"] == "error" - assert "invalid presentation spec" in out["error"] - - with pg_engine.connect() as conn: - rows = ArtifactsRepository(conn).list_artifacts(conversation_id=conversation_id) - assert rows == [] - - -def test_edit_cross_tenant_denied(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - # An artifact owned by a DIFFERENT conversation must not be editable here. - other_conversation = str(uuid.uuid4()) - foreign_id = _seed_presentation(pg_engine, storage, other_conversation) - - out = tool.execute_action("edit_artifact", id=foreign_id, spec_patch={"title": "x"}) - assert out["status"] == "error" - assert "not found in this conversation/run" in out["error"] - - # The foreign artifact stayed at v1 (no version appended). - with pg_engine.connect() as conn: - artifact = ArtifactsRepository(conn).get_artifact(foreign_id) - assert artifact["current_version"] == 1 - - -def test_short_ref_numbering_and_edit_by_ref(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - from pptx import Presentation - - # Successive creates in one conversation get A1, A2 (1-based, by created order). - first = tool.execute_action( - "create_artifact", kind="presentation", title="One", spec={"slides": [{"title": "a"}]} - ) - assert first["status"] == "ok", first - assert first["ref"] == "A1" - second = tool.execute_action( - "create_artifact", kind="document", spec={"sections": [{"heading": "h", "paragraphs": ["p"]}]} - ) - assert second["status"] == "ok", second - assert second["ref"] == "A2" - - # edit_artifact(id="A1") resolves the ref -> appends v2 to the FIRST artifact only. - edited = tool.execute_action("edit_artifact", id="A1", spec_patch={"slides": [{"title": "a"}, {"title": "b"}]}) - assert edited["status"] == "ok", edited - assert edited["version"] == 2 - assert edited["artifact_id"] == first["artifact_id"] - # The append keeps the artifact's position, so its ref is stable. - assert edited["ref"] == "A1" - - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(first["artifact_id"], conversation_id=conversation_id) - v1 = repo.get_version(first["artifact_id"], 1) - v2 = repo.get_version(first["artifact_id"], 2) - assert artifact["current_version"] == 2 - # Append-only: v1 preserved with one slide, v2 has two. - assert len(v1["spec"]["slides"]) == 1 - assert len(v2["spec"]["slides"]) == 2 - assert len(Presentation(storage.get_file(v2["storage_path"])).slides) == 2 - - # The uuid path still works alongside refs. - by_uuid = tool.execute_action("edit_artifact", id=first["artifact_id"], spec_patch={"title": "X"}) - assert by_uuid["status"] == "ok" - assert by_uuid["version"] == 3 - - -def test_short_ref_is_conversation_scoped(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - # Seed an artifact in a DIFFERENT conversation; its A1 must not resolve here. - other_conversation = str(uuid.uuid4()) - _seed_presentation(pg_engine, storage, other_conversation) - - # This conversation has no artifacts yet, so A1 resolves to nothing (no cross-parent leak). - out = tool.execute_action("edit_artifact", id="A1", spec_patch={"title": "x"}) - assert out["status"] == "error" - assert "not found in this conversation/run" in out["error"] - - -def test_short_ref_out_of_range_is_clean_error(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - created = tool.execute_action("create_artifact", kind="presentation", spec={"slides": [{"title": "a"}]}) - assert created["ref"] == "A1" - # A2 has no artifact behind it -> clean not-found, no crash. - out = tool.execute_action("edit_artifact", id="A2", spec_patch={"title": "x"}) - assert out["status"] == "error" - assert "not found in this conversation/run" in out["error"] - - -def _seed_presentation(pg_engine, storage, conversation_id) -> str: - """Create a presentation artifact (row + version + bytes) in a given conversation; return its id.""" - spec = {"slides": [{"title": "seed"}]} - data = b"%PDF-stub-bytes" # content is irrelevant to the authz check - with pg_engine.begin() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.create_artifact( - "user-e2e", - "presentation", - conversation_id=conversation_id, - title="seed", - mime_type="application/vnd.openxmlformats-officedocument.presentationml.presentation", - filename="seed.pptx", - storage_path=None, - size=len(data), - sha256="0" * 64, - spec=spec, - produced_by={"tool": "artifact_generator", "action": "create_artifact", "spec_kind": "presentation"}, - ) - artifact_id = str(artifact["id"]) - from sqlalchemy import text - - storage_path = f"inputs/user-e2e/artifacts/{artifact_id}/v1/seed.pptx" - storage.save_file(io.BytesIO(data), storage_path) - conn.execute( - text( - "UPDATE artifact_versions SET storage_path = :p " - "WHERE artifact_id = CAST(:aid AS uuid) AND version = 1" - ), - {"p": storage_path, "aid": artifact_id}, - ) - return artifact_id diff --git a/tests/integration/test_code_executor_e2e.py b/tests/integration/test_code_executor_e2e.py deleted file mode 100644 index 293b6d85..00000000 --- a/tests/integration/test_code_executor_e2e.py +++ /dev/null @@ -1,279 +0,0 @@ -"""End-to-end CodeExecutorTool: live Jupyter gateway + ephemeral Postgres + local storage. - -Launches a real ``jupyter kernelgateway`` (no Docker — the credential helper -hangs on dev machines), wires the tool to the ephemeral pytest-postgresql DB -and a temp-dir ``LocalStorage``, and drives ``run_code`` through the full path: -code writes a file -> artifact row + bytes persisted with server-side size/ -sha256/mime -> compact payload + ``get_artifact_id``. Also covers an input -artifact round-trip and the timeout error path (no hang). - -Skips gracefully when the gateway binary or websocket-client is unavailable. -""" - -from __future__ import annotations - -import hashlib -import io -import shutil -import socket -import subprocess -import time -import uuid - -import pytest -from sqlalchemy import text - -requests = pytest.importorskip("requests") -pytest.importorskip("websocket") # websocket-client - -from application.agents.tools.code_executor import CodeExecutorTool # noqa: E402 -from application.sandbox.jupyter_gateway import JupyterKernelGatewaySandbox # noqa: E402 -from application.sandbox.manager import SandboxManager # noqa: E402 -from application.sandbox.sandbox_creator import SandboxCreator # noqa: E402 -from application.storage.db.repositories.artifacts import ArtifactsRepository # noqa: E402 -from application.storage.local import LocalStorage # noqa: E402 -from application.storage.storage_creator import StorageCreator # noqa: E402 - -_GATEWAY_BIN = shutil.which("jupyter-kernelgateway") or shutil.which("jupyter") - -pytestmark = [ - pytest.mark.integration, - pytest.mark.skipif( - _GATEWAY_BIN is None, - reason="jupyter kernel gateway not installed (pip install jupyter-kernel-gateway)", - ), -] - - -def _free_port() -> int: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.bind(("127.0.0.1", 0)) - return sock.getsockname()[1] - - -def _gateway_cmd(port: int) -> list: - if _GATEWAY_BIN.endswith("jupyter-kernelgateway"): - base = [_GATEWAY_BIN] - else: - base = [_GATEWAY_BIN, "kernelgateway"] - return base + [ - "--KernelGatewayApp.ip=127.0.0.1", - f"--KernelGatewayApp.port={port}", - "--ZMQChannelsWebsocketConnection.limit_rate=False", - ] - - -@pytest.fixture(scope="module") -def gateway_url(): - port = _free_port() - proc = subprocess.Popen(_gateway_cmd(port), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - url = f"http://127.0.0.1:{port}" - deadline = time.time() + 30 - ready = False - try: - while time.time() < deadline: - if proc.poll() is not None: - pytest.skip("jupyter kernelgateway process exited during startup") - try: - if requests.get(f"{url}/api", timeout=1).status_code == 200: - ready = True - break - except requests.RequestException: - time.sleep(0.3) - if not ready: - pytest.skip("jupyter kernelgateway did not become ready in time") - yield url - finally: - proc.terminate() - try: - proc.wait(timeout=10) - except subprocess.TimeoutExpired: - proc.kill() - - -@pytest.fixture() -def wired_tool(gateway_url, pg_engine, tmp_path, monkeypatch): - """A CodeExecutorTool wired to the live gateway, ephemeral PG, and a temp local store.""" - # Sandbox: a fresh manager over the live gateway, installed as the singleton. - backend = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=30.0) - SandboxCreator._instance = SandboxManager(backend=backend, max_ttl=1200.0) - - # Storage: temp-dir local storage as the singleton. - storage = LocalStorage(base_dir=str(tmp_path)) - monkeypatch.setattr(StorageCreator, "_instance", storage, raising=False) - - # DB: route db_session()/db_readonly() at the ephemeral PG engine. - monkeypatch.setattr("application.storage.db.session.get_engine", lambda: pg_engine) - - conversation_id = str(uuid.uuid4()) - tool = CodeExecutorTool( - tool_config={"conversation_id": conversation_id, "tool_id": str(uuid.uuid4())}, - user_id="user-e2e", - ) - try: - yield tool, conversation_id, pg_engine, storage - finally: - SandboxCreator.reset() - - -def test_run_code_persists_produced_artifact(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - code = ( - "with open('report.txt', 'w') as f:\n" - " f.write('hello artifact')\n" - "print('wrote report')\n" - ) - payload = tool.execute_action("run_code", code=code, ttl=120) - - assert payload["status"] == "ok", payload - assert "wrote report" in payload["stdout_tail"] - assert len(payload["artifacts"]) == 1 - art = payload["artifacts"][0] - assert art["filename"] == "report.txt" - assert art["mime_type"] == "text/plain" - assert art["size"] == len(b"hello artifact") - assert art["version"] == 1 - - # get_artifact_id points at the produced artifact (UI rail). - assert tool.get_artifact_id("run_code") == art["artifact_id"] - - # DB row exists, parent-scoped, with server-computed size + sha256. - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(art["artifact_id"], conversation_id=conversation_id) - assert artifact is not None - version = repo.get_version(art["artifact_id"], 1) - assert version["size"] == len(b"hello artifact") - assert version["sha256"] == hashlib.sha256(b"hello artifact").hexdigest() - assert version["mime_type"] == "text/plain" - produced = version["produced_by"] - assert produced["tool"] == "code_executor" and produced["action"] == "run_code" - - # Bytes are actually in storage at the server-derived key, and match. - storage_path = version["storage_path"] - assert storage_path.startswith("inputs/user-e2e/artifacts/") - assert art["artifact_id"] in storage_path - assert storage.get_file(storage_path).read() == b"hello artifact" - - -def test_run_code_input_artifact_roundtrip(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - # Seed an input artifact (row + bytes) scoped to this conversation. - seed_bytes = b"seed-value-123" - artifact_id = _seed_artifact(pg_engine, storage, conversation_id, "seed.txt", seed_bytes) - - code = ( - "data = open('inputs/seed.txt', 'rb').read()\n" - "open('echo.txt', 'wb').write(data + b'-processed')\n" - "print('read', len(data), 'bytes')\n" - ) - payload = tool.execute_action("run_code", code=code, inputs=[artifact_id], ttl=120) - - assert payload["status"] == "ok", payload - assert "read 14 bytes" in payload["stdout_tail"] - assert payload["inputs_loaded"] == ["inputs/seed.txt"] - assert len(payload["artifacts"]) == 1 - out = payload["artifacts"][0] - assert out["filename"] == "echo.txt" - assert out["size"] == len(seed_bytes + b"-processed") - - with pg_engine.connect() as conn: - version = ArtifactsRepository(conn).get_version(out["artifact_id"], 1) - assert storage.get_file(version["storage_path"]).read() == seed_bytes + b"-processed" - - -def test_run_code_input_artifact_cross_tenant_blocked(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - # An artifact that belongs to a DIFFERENT conversation must not be reachable. - other_conversation = str(uuid.uuid4()) - foreign_id = _seed_artifact(pg_engine, storage, other_conversation, "secret.txt", b"top-secret") - - payload = tool.execute_action( - "run_code", code="open('x.txt','w').write('x')", inputs=[foreign_id], ttl=60 - ) - assert payload["status"] == "error" - assert "not found in this conversation/run" in payload["error"] - - -def test_run_code_captures_overwritten_file(wired_tool): - tool, conversation_id, pg_engine, storage = wired_tool - - first = tool.execute_action( - "run_code", code="open('out.txt','w').write('first-content')", ttl=120 - ) - assert first["status"] == "ok", first - assert len(first["artifacts"]) == 1 - first_id = first["artifacts"][0]["artifact_id"] - - # The same persisted session overwrites out.txt with new content; the diff - # is content-aware, so the new bytes must be captured as a fresh artifact - # rather than dropped as an "already-seen path". - second = tool.execute_action( - "run_code", code="open('out.txt','w').write('second-content-longer')", ttl=120 - ) - assert second["status"] == "ok", second - assert len(second["artifacts"]) == 1 - second_art = second["artifacts"][0] - assert second_art["artifact_id"] != first_id - assert second_art["filename"] == "out.txt" - assert second_art["size"] == len(b"second-content-longer") - - with pg_engine.connect() as conn: - version = ArtifactsRepository(conn).get_version(second_art["artifact_id"], 1) - assert storage.get_file(version["storage_path"]).read() == b"second-content-longer" - - -def test_run_code_skips_unchanged_file_on_rerun(wired_tool): - tool, _conversation_id, _pg_engine, _storage = wired_tool - - first = tool.execute_action( - "run_code", code="open('keep.txt','w').write('same')", ttl=120 - ) - assert len(first["artifacts"]) == 1 - - # A re-run that leaves keep.txt untouched must not re-persist it. - second = tool.execute_action("run_code", code="print('noop')", ttl=120) - assert second["status"] == "ok", second - assert second["artifacts"] == [] - - -def test_run_code_timeout_returns_clean_error(wired_tool): - tool, _conversation_id, _pg_engine, _storage = wired_tool - - payload = tool.execute_action("run_code", code="import time; time.sleep(10)", timeout=1) - assert payload["status"] == "error" - # A clean structured error, not a hang: the sandbox interrupted the kernel. - assert "TimeoutError" in payload["error"] - assert payload["artifacts"] == [] - - -def _seed_artifact(pg_engine, storage, conversation_id, filename, data) -> str: - """Create an artifact row + version + stored bytes, returning its id.""" - sha256 = hashlib.sha256(data).hexdigest() - with pg_engine.begin() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.create_artifact( - "user-e2e", - "file", - conversation_id=conversation_id, - title=filename, - mime_type="text/plain", - filename=filename, - storage_path=None, - size=len(data), - sha256=sha256, - ) - artifact_id = str(artifact["id"]) - storage_path = f"inputs/user-e2e/artifacts/{artifact_id}/v1/{filename}" - storage.save_file(io.BytesIO(data), storage_path) - conn.execute( - text( - "UPDATE artifact_versions SET storage_path = :p " - "WHERE artifact_id = CAST(:aid AS uuid) AND version = 1" - ), - {"p": storage_path, "aid": artifact_id}, - ) - return artifact_id diff --git a/tests/integration/test_workflow_code_node_e2e.py b/tests/integration/test_workflow_code_node_e2e.py deleted file mode 100644 index 888c8124..00000000 --- a/tests/integration/test_workflow_code_node_e2e.py +++ /dev/null @@ -1,280 +0,0 @@ -"""End-to-end workflow ``code`` node: live Jupyter gateway + live Postgres + local storage. - -Launches a real ``jupyter kernelgateway``, wires the sandbox manager + a temp-dir -``LocalStorage`` + the integration Postgres engine, and drives a ``WorkflowEngine`` -through a ``code`` node: the node writes a file -> an artifact REFERENCE lands in -``state`` (artifact row persisted, run-scoped; the ref is ``{artifact_id,...}`` with -NO bytes); the ref survives the ``workflow_runs`` state-snapshot serialization; a -downstream node re-reads it by reference; the ``artifacts.*`` namespace and CEL both -resolve its metadata. - -Skips gracefully when the gateway binary / websocket-client / POSTGRES_URI is absent. -""" - -from __future__ import annotations - -import shutil -import socket -import subprocess -import time -import uuid -from types import SimpleNamespace - -import pytest - -requests = pytest.importorskip("requests") -pytest.importorskip("websocket") # websocket-client - -from application.agents.workflow_agent import WorkflowAgent # noqa: E402 -from application.agents.workflows.cel_evaluator import evaluate_cel # noqa: E402 -from application.agents.workflows.schemas import ( # noqa: E402 - NodeType, - Workflow, - WorkflowGraph, - WorkflowNode, -) -from application.agents.workflows.workflow_engine import WorkflowEngine # noqa: E402 -from application.sandbox.jupyter_gateway import JupyterKernelGatewaySandbox # noqa: E402 -from application.sandbox.manager import SandboxManager # noqa: E402 -from application.sandbox.sandbox_creator import SandboxCreator # noqa: E402 -from application.storage.db.repositories.artifacts import ArtifactsRepository # noqa: E402 -from application.storage.local import LocalStorage # noqa: E402 -from application.storage.storage_creator import StorageCreator # noqa: E402 -from application.templates.namespaces import NamespaceManager # noqa: E402 - -_GATEWAY_BIN = shutil.which("jupyter-kernelgateway") or shutil.which("jupyter") - -pytestmark = [ - pytest.mark.integration, - pytest.mark.skipif( - _GATEWAY_BIN is None, - reason="jupyter kernel gateway not installed (pip install jupyter-kernel-gateway)", - ), -] - - -def _free_port() -> int: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.bind(("127.0.0.1", 0)) - return sock.getsockname()[1] - - -def _gateway_cmd(port: int) -> list: - if _GATEWAY_BIN.endswith("jupyter-kernelgateway"): - base = [_GATEWAY_BIN] - else: - base = [_GATEWAY_BIN, "kernelgateway"] - return base + [ - "--KernelGatewayApp.ip=127.0.0.1", - f"--KernelGatewayApp.port={port}", - "--ZMQChannelsWebsocketConnection.limit_rate=False", - ] - - -@pytest.fixture(scope="module") -def gateway_url(): - port = _free_port() - proc = subprocess.Popen(_gateway_cmd(port), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - url = f"http://127.0.0.1:{port}" - deadline = time.time() + 30 - ready = False - try: - while time.time() < deadline: - if proc.poll() is not None: - pytest.skip("jupyter kernelgateway process exited during startup") - try: - if requests.get(f"{url}/api", timeout=1).status_code == 200: - ready = True - break - except requests.RequestException: - time.sleep(0.3) - if not ready: - pytest.skip("jupyter kernelgateway did not become ready in time") - yield url - finally: - proc.terminate() - try: - proc.wait(timeout=10) - except subprocess.TimeoutExpired: - proc.kill() - - -@pytest.fixture() -def wired(gateway_url, pg_engine, tmp_path, monkeypatch): - """Sandbox manager + temp local storage + integration PG, with a fresh run id.""" - backend = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=30.0) - SandboxCreator._instance = SandboxManager(backend=backend, max_ttl=1200.0) - - storage = LocalStorage(base_dir=str(tmp_path)) - monkeypatch.setattr(StorageCreator, "_instance", storage, raising=False) - monkeypatch.setattr("application.storage.db.session.get_engine", lambda: pg_engine) - - run_id = str(uuid.uuid4()) - try: - yield run_id, pg_engine, storage - finally: - SandboxCreator.reset() - - -def _engine(run_id: str) -> WorkflowEngine: - graph = WorkflowGraph(workflow=Workflow(name="Code Node E2E"), nodes=[], edges=[]) - agent = SimpleNamespace( - endpoint="stream", - llm_name="openai", - model_id="gpt-4o-mini", - api_key="k", - chat_history=[], - user="user-wf-e2e", - decoded_token={"sub": "user-wf-e2e"}, - request_id="req-1", - retrieved_docs=None, - ) - return WorkflowEngine(graph, agent, workflow_run_id=run_id) - - -def _code_node(node_id, code, **config): - cfg = {"code": code} - cfg.update(config) - return WorkflowNode( - id=node_id, workflow_id="wf-e2e", type=NodeType.CODE, title="Code", - position={"x": 0, "y": 0}, config=cfg, - ) - - -def test_code_node_persists_artifact_reference_in_state(wired): - run_id, pg_engine, storage = wired - engine = _engine(run_id) - - node = _code_node( - "code_1", - "with open('report.txt', 'w') as f:\n f.write('compliance ok')\n", - output_variable="report", - ) - list(engine._execute_code_node(node)) - - ref = engine.state["report"] - # The state holds an artifact REFERENCE (JSON primitives), never bytes. The - # short handle ``ref`` (A1) lets a later node address the artifact by position. - assert set(ref) == {"artifact_id", "version", "filename", "mime_type", "size", "ref"} - assert ref["filename"] == "report.txt" - assert ref["mime_type"] == "text/plain" - assert ref["size"] == len(b"compliance ok") - assert ref["ref"] == "A1" - assert all(not isinstance(v, (bytes, bytearray)) for v in ref.values()) - assert engine.state["node_code_1_output"] == ref - - # The artifact row is persisted, parent-scoped to THIS run (not cross-tenant). - with pg_engine.connect() as conn: - repo = ArtifactsRepository(conn) - artifact = repo.get_artifact_in_parent(ref["artifact_id"], workflow_run_id=run_id) - assert artifact is not None - assert repo.get_artifact_in_parent(ref["artifact_id"], workflow_run_id=str(uuid.uuid4())) is None - version = repo.get_version(ref["artifact_id"], 1) - assert version["size"] == len(b"compliance ok") - assert version["produced_by"]["node_id"] == "code_1" - # The bytes live in storage, never in state. - assert storage.get_file(version["storage_path"]).read() == b"compliance ok" - - -def test_reference_survives_serialization_and_cel_branches(wired): - run_id, _pg_engine, _storage = wired - engine = _engine(run_id) - - node = _code_node( - "code_meta", - "open('out.json', 'w').write('{\"k\": 1}')\n", - output_variable="report", - ) - list(engine._execute_code_node(node)) - ref = engine.state["report"] - - # Survives the workflow_runs state-snapshot serialization unchanged. - agent = WorkflowAgent.__new__(WorkflowAgent) - serialized = agent._serialize_state(engine.state) - assert serialized["report"] == ref - assert isinstance(serialized["report"], dict) - - # CEL branches on the reference's metadata (nested-dict indexing). - assert evaluate_cel("report.size > 0", engine.state) is True - assert evaluate_cel('report.mime_type == "application/json"', engine.state) is True - assert evaluate_cel('report.mime_type == "application/pdf"', engine.state) is False - - -def test_condition_node_branches_on_artifact_metadata(wired): - run_id, _pg_engine, _storage = wired - engine = _engine(run_id) - - list(engine._execute_code_node( - _code_node("code_c", "open('r.json','w').write('{}')", output_variable="report") - )) - - cond = WorkflowNode( - id="cond_1", workflow_id="wf-e2e", type=NodeType.CONDITION, title="Branch", - position={"x": 0, "y": 0}, - config={ - "mode": "simple", - "cases": [ - {"name": "has json", "expression": 'report.mime_type == "application/json"', - "sourceHandle": "json_case"}, - ], - }, - ) - list(engine._execute_condition_node(cond)) - assert engine._condition_result == "json_case" - - -def test_downstream_node_reads_reference_via_inputs(wired): - run_id, _pg_engine, _storage = wired - engine = _engine(run_id) - - list(engine._execute_code_node( - _code_node("producer", "open('seed.txt','w').write('seed-bytes')", output_variable="seed") - )) - - # A downstream code node references the upstream ref by its state-var name; - # the engine re-fetches the bytes by id and stages them into the workspace. - consumer = _code_node( - "consumer", - "data = open('inputs/seed.txt', 'rb').read()\n" - "open('combined.txt', 'wb').write(data + b'-processed')\n", - inputs=["seed"], - output_variable="combined", - ) - list(engine._execute_code_node(consumer)) - - out = engine.state["combined"] - assert out["filename"] == "combined.txt" - assert out["size"] == len(b"seed-bytes-processed") - assert out["artifact_id"] != engine.state["seed"]["artifact_id"] - - -def test_artifacts_namespace_resolves_reference(wired): - run_id, _pg_engine, _storage = wired - engine = _engine(run_id) - - list(engine._execute_code_node( - _code_node("ns_node", "open('deck.txt','w').write('hello')", output_variable="report") - )) - ref = engine.state["report"] - - # The artifacts.* namespace (shared with the prompt renderer) exposes the - # reference's metadata by output-variable name, and artifact(id) resolves - # parent-scoped metadata — never bytes. - ctx = NamespaceManager().build_context( - artifacts_data={"report": ref}, - artifact_parent={"workflow_run_id": run_id}, - ) - assert ctx["artifacts"]["report"]["id"] == ref["artifact_id"] - assert ctx["artifacts"]["report"]["mime_type"] == "text/plain" - assert ctx["artifacts"]["report"]["filename"] == "deck.txt" - - looked_up = ctx["artifacts"]["artifact"](ref["artifact_id"]) - assert looked_up["id"] == ref["artifact_id"] - assert looked_up["filename"] == "deck.txt" - assert all(not isinstance(v, (bytes, bytearray)) for v in looked_up.values()) - - # A foreign run id never resolves this artifact (no cross-tenant leak). - foreign = NamespaceManager().build_context( - artifact_parent={"workflow_run_id": str(uuid.uuid4())}, - ) - assert foreign["artifacts"]["artifact"](ref["artifact_id"]) == {} diff --git a/tests/requirements.txt b/tests/requirements.txt index 04d6d036..65610d93 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -5,16 +5,9 @@ coverage>=7.4.0 pytest-postgresql>=6.0.0 psycopg[binary]>=3.1 cryptography>=46.0.0 -# Runner engine for the sandbox integration test: tests launch a local -# `jupyter kernelgateway` process (no Docker). Not an app runtime dependency. -# ipykernel provides the python3 kernel the gateway launches per session. -jupyter-kernel-gateway>=2.5.0 -ipykernel>=6.0.0 -websocket-client>=1.6.0 -# Document renderers the artifact_generator runs inside the sandbox kernel. -# In production these are pre-baked into the sandbox image; locally the kernel -# uses the app .venv, so the integration test needs them installed. python-pptx -# and openpyxl already ship in application/requirements.txt; only the rest are -# test-only here (kept out of application/requirements.txt — sandbox-image libs). +# jupyter_client (KernelSpecManager) for the sandbox kernelspec-resolution test. +jupyter-client>=8.0 +# Document renderers used by the artifact_generator unit-test fixtures +# (python-pptx and openpyxl already ship in application/requirements.txt). python-docx>=1.1 reportlab>=4.0,<5 diff --git a/tests/sandbox/test_daytona_live.py b/tests/sandbox/test_daytona_live.py deleted file mode 100644 index 473ef63f..00000000 --- a/tests/sandbox/test_daytona_live.py +++ /dev/null @@ -1,50 +0,0 @@ -"""Optional live smoke test against real Daytona Cloud; skipped unless explicitly enabled. - -Run with both flags set so CI/normal runs never burn cloud quota:: - - RUN_DAYTONA_LIVE=1 DAYTONA_API_KEY=dtn_... .venv/bin/python -m pytest \ - tests/sandbox/test_daytona_live.py -q --no-cov -""" - -import os -import uuid - -import pytest - -pytest.importorskip("daytona") - -_ENABLED = os.environ.get("RUN_DAYTONA_LIVE") == "1" and bool(os.environ.get("DAYTONA_API_KEY")) - -pytestmark = pytest.mark.skipif( - not _ENABLED, - reason="set RUN_DAYTONA_LIVE=1 and DAYTONA_API_KEY to run the live Daytona smoke test", -) - - -@pytest.fixture() -def live_sandbox(): - from application.sandbox.daytona import DaytonaSandbox - - s = DaytonaSandbox( - api_key=os.environ["DAYTONA_API_KEY"], - api_url=os.environ.get("DAYTONA_API_URL") or None, - target=os.environ.get("DAYTONA_TARGET") or None, - ) - session_id = "smoke" + uuid.uuid4().hex[:8] - s.open(session_id) - try: - yield s, session_id - finally: - s.close(session_id) # always delete the cloud sandbox so we never leak cost - - -def test_live_exec_and_file_roundtrip(live_sandbox): - s, session_id = live_sandbox - - res = s.exec(session_id, "print(6 * 7)") - assert res.ok, res.error_value - assert "42" in res.stdout - - s.put_file(session_id, "hello.txt", b"daytona-smoke") - assert s.get_file(session_id, "hello.txt") == b"daytona-smoke" - assert "hello.txt" in s.list_files(session_id) diff --git a/tests/sandbox/test_jupyter_gateway_integration.py b/tests/sandbox/test_jupyter_gateway_integration.py deleted file mode 100644 index 393fc678..00000000 --- a/tests/sandbox/test_jupyter_gateway_integration.py +++ /dev/null @@ -1,294 +0,0 @@ -"""End-to-end test against a locally launched Jupyter Kernel Gateway subprocess. - -Launches a real `jupyter kernelgateway` on an ephemeral port (no Docker — the -Docker credential helper hangs on this machine), points the backend at it, and -exercises the full session lifecycle. Skips gracefully if the gateway binary or -the websocket-client library is unavailable. -""" - -import shutil -import socket -import subprocess -import time - -import pytest - -requests = pytest.importorskip("requests") -pytest.importorskip("websocket") # websocket-client - -from application.sandbox.jupyter_gateway import JupyterKernelGatewaySandbox # noqa: E402 - -_GATEWAY_BIN = shutil.which("jupyter-kernelgateway") or shutil.which("jupyter") - -pytestmark = pytest.mark.skipif( - _GATEWAY_BIN is None, - reason="jupyter kernel gateway not installed (pip install jupyter-kernel-gateway)", -) - - -def _free_port() -> int: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.bind(("127.0.0.1", 0)) - return sock.getsockname()[1] - - -def _gateway_cmd(port: int) -> list: - if _GATEWAY_BIN.endswith("jupyter-kernelgateway"): - base = [_GATEWAY_BIN] - else: - base = [_GATEWAY_BIN, "kernelgateway"] - return base + [ - "--KernelGatewayApp.ip=127.0.0.1", - f"--KernelGatewayApp.port={port}", - # Raise the iopub data-rate limit so large get_file payloads aren't truncated. - "--ZMQChannelsWebsocketConnection.limit_rate=False", - ] - - -@pytest.fixture(scope="module") -def gateway_url(): - port = _free_port() - proc = subprocess.Popen( - _gateway_cmd(port), - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - url = f"http://127.0.0.1:{port}" - deadline = time.time() + 30 - ready = False - try: - while time.time() < deadline: - if proc.poll() is not None: - pytest.skip("jupyter kernelgateway process exited during startup") - try: - resp = requests.get(f"{url}/api", timeout=1) - if resp.status_code == 200: - ready = True - break - except requests.RequestException: - time.sleep(0.3) - if not ready: - pytest.skip("jupyter kernelgateway did not become ready in time") - yield url - finally: - proc.terminate() - try: - proc.wait(timeout=10) - except subprocess.TimeoutExpired: - proc.kill() - - -@pytest.fixture() -def sandbox(gateway_url): - sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=30.0) - yield sb - - -def test_open_and_close_lifecycle(sandbox): - session = "conv-lifecycle" - kernel_id = sandbox.open(session) - assert kernel_id - resp = requests.get(f"{sandbox._base_url}/api/kernels/{kernel_id}", timeout=5) - assert resp.status_code == 200 - sandbox.close(session) - resp = requests.get(f"{sandbox._base_url}/api/kernels/{kernel_id}", timeout=5) - assert resp.status_code == 404 # kernel torn down - - -def test_stateful_exec_carries_variables(sandbox): - session = "conv-state" - sandbox.open(session) - try: - first = sandbox.exec(session, "x = 1") - assert first.ok, first.error_value - second = sandbox.exec(session, "print(x + 41)") - assert second.ok, second.error_value - assert second.stdout.strip() == "42" - assert second.execution_count is not None - finally: - sandbox.close(session) - - -def test_stdout_capture(sandbox): - session = "conv-stdout" - sandbox.open(session) - try: - res = sandbox.exec(session, "print('hello sandbox')") - assert res.ok - assert "hello sandbox" in res.stdout - finally: - sandbox.close(session) - - -def test_error_capture_sets_failure_status(sandbox): - session = "conv-error" - sandbox.open(session) - try: - res = sandbox.exec(session, "raise ValueError('boom')") - assert not res.ok - assert res.status == "error" - assert res.error_name == "ValueError" - assert "boom" in (res.error_value or "") - assert res.traceback - finally: - sandbox.close(session) - - -def test_put_and_get_file_roundtrip_exact_bytes(sandbox): - session = "conv-files" - sandbox.open(session) - try: - payload = bytes(range(256)) * 4 # 1 KiB of every byte value - sandbox.put_file(session, "data/blob.bin", payload) - assert "data/blob.bin" in sandbox.list_files(session) - fetched = sandbox.get_file(session, "data/blob.bin") - assert fetched == payload - finally: - sandbox.close(session) - - -def test_file_written_by_code_is_readable(sandbox): - session = "conv-codefile" - sandbox.open(session) - try: - res = sandbox.exec(session, "open('out.txt', 'w').write('generated')") - assert res.ok, res.error_value - assert sandbox.get_file(session, "out.txt") == b"generated" - finally: - sandbox.close(session) - - -def test_attach_reuses_running_kernel(sandbox): - session = "conv-attach" - first_id = sandbox.open(session) - try: - sandbox.exec(session, "y = 7") - second_id = sandbox.attach(session) - assert second_id == first_id - res = sandbox.exec(session, "print(y)") - assert res.stdout.strip() == "7" - finally: - sandbox.close(session) - - -def test_exec_timeout_reports_error(sandbox): - session = "conv-timeout" - sandbox.open(session) - try: - res = sandbox.exec(session, "import time; time.sleep(5)", timeout=1.0) - assert not res.ok - assert res.error_name == "TimeoutError" - finally: - sandbox.close(session) - - -def test_kernel_killed_mid_exec_returns_error(sandbox): - session = "conv-killed" - kernel_id = sandbox.open(session) - try: - import threading - - def _kill(): - time.sleep(0.5) - requests.delete(f"{sandbox._base_url}/api/kernels/{kernel_id}", timeout=5) - - killer = threading.Thread(target=_kill) - killer.start() - res = sandbox.exec(session, "import time\nfor _ in range(50): time.sleep(0.2)", timeout=20.0) - killer.join() - assert not res.ok # error result, no raise / no hang - assert res.exit_code == -1 - finally: - sandbox.close(session) - - -def test_runaway_loop_times_out_and_session_stays_reusable(sandbox): - session = "conv-runaway" - sandbox.open(session) - try: - start = time.monotonic() - res = sandbox.exec(session, "while True:\n print('x')", timeout=2.0) - elapsed = time.monotonic() - start - assert not res.ok - assert res.error_name == "TimeoutError" - assert elapsed < 2.0 + 8.0 # returns within timeout + interrupt slack - # Interrupt should have freed the kernel: a fresh exec on the same session works. - again = sandbox.exec(session, "print(2 + 2)", timeout=10.0) - assert again.ok, again.error_value - assert again.stdout.strip() == "4" - finally: - sandbox.close(session) - - -def test_output_cap_truncates_runaway_output(gateway_url): - sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=60.0, max_output_bytes=50_000) - session = "conv-bigout" - sb.open(session) - try: - # Emit ~200 KiB (4x the 50 KB cap) as bounded, fast work under a generous timeout, so the - # output cap -- not the wall-clock deadline -- deterministically trips even under load. - res = sb.exec(session, "for _ in range(200): print('A' * 1000)", timeout=60.0) - assert "truncated" in res.stderr - assert len(res.stdout) <= 60_000 # capped well under the produced ~200 KiB - again = sb.exec(session, "print('alive')", timeout=10.0) - assert again.ok and "alive" in again.stdout - finally: - sb.close(session) - - -def test_put_file_rejects_path_traversal(sandbox): - session = "conv-trav-put" - sandbox.open(session) - try: - with pytest.raises(IOError): - sandbox.put_file(session, "../escape.txt", b"nope") - with pytest.raises(IOError): - sandbox.put_file(session, "/etc/escape.txt", b"nope") - # The escape target must not exist (no file written outside the workspace). - probe = sandbox.exec(session, "import os; print(os.path.exists('/etc/escape.txt'))") - assert probe.stdout.strip() == "False" - finally: - sandbox.close(session) - - -def test_get_file_rejects_path_traversal(sandbox): - session = "conv-trav-get" - sandbox.open(session) - try: - with pytest.raises(IOError): - sandbox.get_file(session, "../../etc/passwd") - with pytest.raises(IOError): - sandbox.get_file(session, "/etc/passwd") - finally: - sandbox.close(session) - - -def test_get_file_rejects_oversized_file(gateway_url): - sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=30.0, max_file_bytes=1024) - session = "conv-toobig" - sb.open(session) - try: - sb.put_file(session, "big.bin", b"x" * 4096) # 4 KiB > 1 KiB cap - with pytest.raises(IOError): - sb.get_file(session, "big.bin") - finally: - sb.close(session) - - -def test_get_file_integrity_roundtrip_large(gateway_url): - sb = JupyterKernelGatewaySandbox(gateway_url=gateway_url, default_timeout=60.0) - session = "conv-large" - sb.open(session) - try: - payload = bytes(range(256)) * 8192 # 2 MiB, exercises the iopub path + integrity check - sb.put_file(session, "big.bin", payload) - fetched = sb.get_file(session, "big.bin") - assert fetched == payload - finally: - sb.close(session) - - -def test_open_rejects_illegal_session_id(sandbox): - for bad in ["../evil", "a/b", "a b", "a;b", ""]: - with pytest.raises(ValueError): - sandbox.open(bad)