mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-13 02:14:23 +00:00
Drop live-service e2e tests; defer to a dedicated e2e PR
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.
This commit is contained in:
Alex committed
1 parent
7c91517d62
commit
61a9749c9b
6 files changed
+4
-1312
No files matched your search
@@ -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 <hi>",
|
||||
"blocks": [
|
||||
{"type": "heading", "text": "Overview", "level": 2},
|
||||
{"type": "paragraph", "text": "Body <script>alert(1)</script>."},
|
||||
{"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("<!doctype html>")
|
||||
assert "Findings <hi>" in html_doc
|
||||
assert "<script>alert(1)</script>" in html_doc
|
||||
assert "<script>alert(1)</script>" not in html_doc
|
||||
assert "<li>one</li>" in html_doc and "<td>a</td>" 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 "<h1>Updated</h1>" 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
|
||||
@@ -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
|
||||
@@ -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"]) == {}
|
||||
+4
-11
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
Reference in new issue
Block a user