Artur Mukhamadiev
d77b52ae86
- tests/integration: frontend served by backend, runtime error mapping, backend + real pi image with scripted model/Confluence (shared example, variants, failure paths, cancellation/busy gate, isolation canaries, HTTP download ownership, 16 MiB prompt round trip), real OpenAI-compatible adapter + real image over a scripted transport, real Chrome against the real backend with a scripted runtime peer, backend crash/restart reconciliation, and an opt-in live model check (marker: live). - backend: map runtime terminal codes (model_output_limit, model_context_exceeded, query_timeout, connectivity_failed) to the contract's HTTP statuses; make the artifact 404 body identical for no-session, wrong-session, unknown and expired IDs. - Makefile, scripts/run-backend.sh, deploy/confluence-web.env.example, root README for the integrated application; integration pytest marker.
88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
"""Test-owned launcher: real backend HTTP server with a scripted runtime peer.
|
|
|
|
Used by the browser <-> backend pair check. The container is a scripted
|
|
in-process peer (no Docker), the model is the fake adapter and Confluence is
|
|
the network-free dev substitute. The scenario for the *next* query is read
|
|
from a control file so a browser test can switch behaviors between queries
|
|
without any dev-only endpoint existing in the production application.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import uvicorn
|
|
|
|
from backend.app import create_app
|
|
from backend.artifacts import ArtifactStore
|
|
from backend.containers import FakeContainerManager
|
|
from backend.dev.fake_confluence import create_client
|
|
from backend.dev.fake_peer import ScriptedContainerPeer
|
|
from backend.model import FakeModelAdapter
|
|
from backend.settings import Settings
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
CONTROL = Path(os.environ["CW_PEER_CONTROL_FILE"])
|
|
|
|
MALICIOUS_MARKDOWN = (
|
|
"# Untrusted answer\n\n"
|
|
'<script>window.__xss = "script"</script>\n\n'
|
|
'<img src="https://evil.example.net/pixel.png" onerror="window.__xss=\'img\'">\n\n'
|
|
"\n\n"
|
|
'<iframe src="https://evil.example.net/frame"></iframe>\n\n'
|
|
'<form action="https://evil.example.net/post"><input name="q"></form>\n\n'
|
|
"[js link](javascript:alert(1))\n\n"
|
|
"Legit citation: [Deployment Guide](https://approved.example.com/pages/viewpage.action?pageId=847291)\n"
|
|
)
|
|
|
|
|
|
async def malicious_steps(reader, writer, peer):
|
|
start = await peer._read_line(reader)
|
|
assert start.get("type") == "start"
|
|
await peer._write_msg(writer, {
|
|
"v": 1, "type": "tool_request", "id": "a_1",
|
|
"payload": {"tool": "confluence_search", "parameters": {"query": "<script>alert('q')</script>"}},
|
|
})
|
|
await peer._read_line(reader)
|
|
await peer._write_msg(writer, {
|
|
"v": 1, "type": "collection_start", "id": "a_2",
|
|
"payload": {"markdown": MALICIOUS_MARKDOWN, "warnings": [{"code": "history_truncated", "message": "<b>bold</b> warning"}]},
|
|
})
|
|
ready = await peer._read_line(reader)
|
|
assert ready.get("type") == "collection_ready"
|
|
await peer._write_msg(writer, {"v": 1, "type": "complete", "id": "a_3", "payload": {"accepted_transfer_count": 0}})
|
|
|
|
|
|
def peer_factory():
|
|
scenario = CONTROL.read_text(encoding="utf-8").strip() if CONTROL.exists() else "standard"
|
|
if scenario == "malicious":
|
|
return ScriptedContainerPeer(custom_steps=malicious_steps)
|
|
return ScriptedContainerPeer(scenario=scenario)
|
|
|
|
|
|
def main() -> None:
|
|
port = int(os.environ.get("CW_DEV_PORT", "8765"))
|
|
settings = Settings(
|
|
dev_mode=True,
|
|
frontend_dist_dir=REPO_ROOT / "frontend",
|
|
artifact_storage_dir=Path(os.environ["CW_ARTIFACT_DIR"]),
|
|
bind_host="127.0.0.1",
|
|
bind_port=port,
|
|
query_timeout_seconds=30.0,
|
|
cleanup_timeout_seconds=5.0,
|
|
)
|
|
app = create_app(
|
|
settings=settings,
|
|
container_manager=FakeContainerManager(peer_factory),
|
|
artifact_store=ArtifactStore(settings.artifact_storage_dir),
|
|
model_adapter=FakeModelAdapter(),
|
|
confluence_client_factory=create_client,
|
|
)
|
|
uvicorn.run(app, host="127.0.0.1", port=port, workers=1, log_level="warning")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|