awesome-templates
deepseek-harness/e2e_test.py
无中文版
deepseek-harness/e2e_test.py
#!/usr/bin/env python3
"""End-to-end test suite for the ``deepseek-harness`` template.
Exercises built-in routes (/health, /commands), the custom ``deepseek``
command, FILE_OPS round-trip and PROCESS shell execution against a
running server.
The ``deepseek`` command calls the DeepSeek API through the ``openai``
client inside the sandbox, which requires ``DEEPSEEK_API_KEY`` in the
sandbox environment. Without the key the command test is marked
``[SKIP]`` instead of failing.
Usage::
python3 e2e_test.py # http://localhost:9000
E2E_BASE_URL=http://host:9000 python3 e2e_test.py
E2E_DEEPSEEK_PROMPT="Say hi" python3 e2e_test.py # custom LLM prompt
"""
from __future__ import annotations
import base64
import json
import os
import sys
import httpx
BASE_URL = os.environ.get("E2E_BASE_URL", "http://localhost:9000").rstrip("/")
WORKSPACE = os.environ.get("E2E_WORKSPACE", "") # empty → auto-detect
HAS_DEEPSEEK_KEY = bool(os.environ.get("DEEPSEEK_API_KEY", "").strip())
RUN_LLM = os.environ.get("E2E_RUN_LLM", "").strip().lower()
REQUIRE = os.environ.get("E2E_REQUIRE_LLM", "").strip().lower() in {"1", "true", "yes"}
MARKER = "ABX_E2E_OK"
LLM_PROMPT = os.environ.get("E2E_DEEPSEEK_PROMPT", f"Reply with exactly: {MARKER}")
MODEL = os.environ.get("E2E_DEEPSEEK_MODEL", "deepseek-coder")
TIMEOUT = httpx.Timeout(connect=10.0, read=150.0, write=30.0, pool=10.0)
RESULTS: list[tuple[str, str, str]] = []
def record(status: str, test: str, detail: str = "") -> None:
RESULTS.append((status, test, detail))
line = f"[{status}] {test}"
if detail:
line += f" — {detail}"
print(line, flush=True)
def _short(value: object, limit: int = 160) -> str:
text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)
return text if len(text) <= limit else text[: limit - 3] + "..."
def detect_workspace(client: httpx.Client) -> str:
"""Find a writable directory accepted by the server's path guard."""
candidates = ([WORKSPACE] if WORKSPACE else []) + ["/home/user", "/workspace"]
for candidate in candidates:
probe = f"{candidate}/.e2e_probe"
try:
resp = client.post("/files/upload-stream", json={
"path": probe,
"content_base64": base64.b64encode(b"probe").decode(),
})
if resp.status_code == 200:
client.delete("/files", params={"path": probe})
return candidate
except Exception: # noqa: BLE001
continue
return ""
def preflight(client: httpx.Client) -> bool:
try:
resp = client.get("/health")
ok = resp.status_code == 200 and resp.json().get("status") == "ok"
print(f"[{'PASS' if ok else 'FAIL'}] preflight GET /health — HTTP {resp.status_code}",
flush=True)
return ok
except Exception as exc: # noqa: BLE001
print(f"[FAIL] preflight GET /health — {exc}", flush=True)
return False
def group_builtin(client: httpx.Client) -> None:
try:
resp = client.get("/commands")
body = resp.json()
entries = body.get("commands", body if isinstance(body, list) else [])
names = {c.get("name") for c in entries}
ok = resp.status_code == 200 and "deepseek" in names
record("PASS" if ok else "FAIL", "GET /commands",
f"HTTP {resp.status_code}, listed={sorted(n for n in names if n)}")
except Exception as exc: # noqa: BLE001
record("FAIL", "GET /commands", f"exception: {exc}")
def group_commands(client: httpx.Client) -> None:
# Missing required argument must surface a clean 400.
try:
resp = client.post("/commands/deepseek", json={})
ok = resp.status_code == 400 and "error" in resp.json()
record("PASS" if ok else "FAIL", "POST /commands/deepseek (missing prompt)",
f"HTTP {resp.status_code} (expected 400)")
except Exception as exc: # noqa: BLE001
record("FAIL", "POST /commands/deepseek (missing prompt)", f"exception: {exc}")
# Real LLM call requires DEEPSEEK_API_KEY in the sandbox environment.
ep = "POST /commands/deepseek (LLM)"
forced_off = RUN_LLM in {"0", "false", "no"}
if REQUIRE and forced_off:
record("FAIL", ep, "E2E_REQUIRE_LLM=1 conflicts with E2E_RUN_LLM=0")
elif forced_off or not HAS_DEEPSEEK_KEY:
status = "FAIL" if REQUIRE or RUN_LLM in {"1", "true", "yes"} else "SKIP"
if forced_off and not REQUIRE:
status = "SKIP"
reason = "E2E_RUN_LLM=0" if forced_off else "DEEPSEEK_API_KEY not set"
record(status, ep, reason)
else:
try:
resp = client.post("/commands/deepseek",
json={"prompt": LLM_PROMPT, "model": MODEL})
body = resp.json()
stdout = str(body.get("result", ""))
found = MARKER in stdout
ok = resp.status_code == 200 and found
record("PASS" if ok else "FAIL", ep,
f"HTTP {resp.status_code}, marker={found}, output={_short(stdout, 120)}")
except Exception as exc: # noqa: BLE001
record("FAIL", ep, f"exception: {exc}")
def group_files(client: httpx.Client) -> None:
if not WORKSPACE:
record("SKIP", "files upload/download round-trip",
"no writable workspace detected — set E2E_WORKSPACE")
return
path = f"{WORKSPACE}/e2e_deepseek_harness.txt"
encoded = base64.b64encode(b"deepseek-harness e2e file round-trip\n").decode()
try:
up = client.post("/files/upload-stream",
json={"path": path, "content_base64": encoded})
down = client.get("/files/download-stream", params={"path": path})
ok = up.status_code == 200 and down.status_code == 200 \
and down.json().get("content_base64") == encoded
record("PASS" if ok else "FAIL", "files upload/download round-trip",
f"upload HTTP {up.status_code}, download HTTP {down.status_code}")
except Exception as exc: # noqa: BLE001
record("FAIL", "files upload/download round-trip", f"exception: {exc}")
try:
resp = client.get("/files/list", params={"path": WORKSPACE})
ok = resp.status_code == 200
record("PASS" if ok else "FAIL", "GET /files/list", f"HTTP {resp.status_code}")
except Exception as exc: # noqa: BLE001
record("FAIL", "GET /files/list", f"exception: {exc}")
def group_shell(client: httpx.Client) -> None:
try:
resp = client.post("/shell", json={"command": "python3 --version"})
body = resp.json()
stdout = str(body.get("stdout", "")).strip()
ok = resp.status_code == 200 and stdout.startswith("Python")
record("PASS" if ok else "FAIL", "POST /shell python3 --version",
f"HTTP {resp.status_code}, {stdout!r}")
except Exception as exc: # noqa: BLE001
record("FAIL", "POST /shell python3 --version", f"exception: {exc}")
try:
resp = client.post("/shell", json={"command": "echo hello"})
ok = resp.status_code == 200 and "hello" in str(resp.json().get("stdout", ""))
record("PASS" if ok else "FAIL", "POST /shell echo", f"HTTP {resp.status_code}")
except Exception as exc: # noqa: BLE001
record("FAIL", "POST /shell echo", f"exception: {exc}")
def summary() -> int:
passed = sum(1 for s, _, _ in RESULTS if s == "PASS")
failed = sum(1 for s, _, _ in RESULTS if s == "FAIL")
skipped = sum(1 for s, _, _ in RESULTS if s == "SKIP")
print("\n" + "=" * 74, flush=True)
print("SUMMARY — deepseek-harness", flush=True)
print("=" * 74, flush=True)
print(f"{passed} passed, {failed} failed, {skipped} skipped "
f"(total {len(RESULTS)})", flush=True)
if failed:
print("\nFailed tests:", flush=True)
for s, t, d in RESULTS:
if s == "FAIL":
print(f" - {t}: {d}", flush=True)
print("=" * 74, flush=True)
return 1 if failed else 0
def main() -> int:
print("deepseek-harness end-to-end suite", flush=True)
print(f" base_url : {BASE_URL}", flush=True)
print(f" deepseek key : {'set' if HAS_DEEPSEEK_KEY else 'not set'}", flush=True)
print(f" model / prompt : {MODEL} / {LLM_PROMPT!r}", flush=True)
print("-" * 74, flush=True)
with httpx.Client(base_url=BASE_URL, timeout=TIMEOUT) as client:
if not preflight(client):
print("server unreachable — aborting", flush=True)
return 2
print("\n-- Built-in routes --", flush=True)
group_builtin(client)
print("\n-- Custom commands (deepseek) --", flush=True)
group_commands(client)
print("\n-- File ops --", flush=True)
global WORKSPACE
WORKSPACE = detect_workspace(client)
group_files(client)
print("\n-- Shell --", flush=True)
group_shell(client)
return summary()
if __name__ == "__main__":
sys.exit(main())