wehub-resource-sync
450 lines
16 KiB
Python
450 lines
16 KiB
Python
"""P6 gate tests — server: OpenAI-compatible endpoint, WS session API, REST."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from coworker.providers import (
|
|
AssistantTurn,
|
|
ModelCapabilities,
|
|
ProviderClient,
|
|
ToolCall,
|
|
)
|
|
from coworker.server import SessionManager, create_app
|
|
from coworker.sessions import SessionRecord
|
|
|
|
|
|
class ScriptedProvider(ProviderClient):
|
|
"""A ProviderClient that returns queued AssistantTurns (streams via base default)."""
|
|
|
|
def __init__(self, turns):
|
|
self._turns = list(turns)
|
|
|
|
def complete(self, *, model, messages, tools=None, **settings):
|
|
return self._turns.pop(0)
|
|
|
|
def capabilities(self, model):
|
|
return ModelCapabilities()
|
|
|
|
|
|
def _text(text):
|
|
return AssistantTurn(text=text, finish_reason="stop")
|
|
|
|
|
|
def _tool(name, args, call_id="call_1"):
|
|
return AssistantTurn(tool_calls=[ToolCall(id=call_id, name=name, arguments=args)])
|
|
|
|
|
|
def _client(tmp_path, turns):
|
|
manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider(turns))
|
|
return TestClient(create_app(manager))
|
|
|
|
|
|
# -- REST -----------------------------------------------------------------------
|
|
|
|
|
|
def test_chat_completions_openai_shape(tmp_path):
|
|
client = _client(tmp_path, [_text("hello world")])
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": "gpt-5.5", "messages": [{"role": "user", "content": "hi"}]},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["object"] == "chat.completion"
|
|
assert body["choices"][0]["message"]["content"] == "hello world"
|
|
assert body["choices"][0]["finish_reason"] == "stop"
|
|
|
|
|
|
def test_agents_and_memory_rest(tmp_path):
|
|
client = _client(tmp_path, [])
|
|
agents = client.get("/v1/agents").json()["agents"]
|
|
# The picker lists surfaced personas, default (cowork) first; Chat is hidden by default.
|
|
names = [a["name"] for a in agents]
|
|
assert names and names[0] == "cowork"
|
|
assert {"cowork", "code"} <= set(names)
|
|
assert "skills" in client.get("/v1/skills").json() # catalog (may be empty)
|
|
|
|
added = client.post("/v1/memory", json={"content": "prefer pathlib"}).json()
|
|
assert added["content"] == "prefer pathlib"
|
|
assert any(
|
|
m["content"] == "prefer pathlib"
|
|
for m in client.get("/v1/memory").json()["memory"]
|
|
)
|
|
|
|
|
|
def test_connector_tool_settings_and_audit_rest(tmp_path):
|
|
client = _client(tmp_path, [])
|
|
connectors = {
|
|
c["name"]: c for c in client.get("/v1/connectors").json()["connectors"]
|
|
}
|
|
assert any(t["name"] == "browser_open_url" for t in connectors["browser"]["tools"])
|
|
|
|
res = client.patch(
|
|
"/v1/connectors/browser/tools", json={"enabled": {"browser_open_url": False}}
|
|
).json()
|
|
assert res["ok"] is True
|
|
connectors = {
|
|
c["name"]: c for c in client.get("/v1/connectors").json()["connectors"]
|
|
}
|
|
browser_tools = {t["name"]: t for t in connectors["browser"]["tools"]}
|
|
assert browser_tools["browser_open_url"]["enabled"] is False
|
|
|
|
assert client.get("/v1/audit", params={"session_id": "none"}).json()["events"] == []
|
|
assert client.get("/v1/browser/state").json()["status"] in {
|
|
"closed",
|
|
"open",
|
|
"error",
|
|
}
|
|
|
|
|
|
def test_artifacts_list_and_read_previewable_files(tmp_path):
|
|
(tmp_path / "brief.md").write_text("# Brief\n\nHello", encoding="utf-8")
|
|
(tmp_path / "page.html").write_text("<h1>Preview</h1>", encoding="utf-8")
|
|
(tmp_path / ".secret.md").write_text("hidden", encoding="utf-8")
|
|
(tmp_path / "node_modules").mkdir()
|
|
(tmp_path / "node_modules" / "noise.md").write_text("skip", encoding="utf-8")
|
|
|
|
client = _client(tmp_path, [])
|
|
artifacts = client.get("/v1/sessions/unknown/artifacts").json()["artifacts"]
|
|
by_path = {a["path"]: a for a in artifacts}
|
|
|
|
assert by_path["brief.md"]["kind"] == "markdown"
|
|
assert by_path["page.html"]["kind"] == "html"
|
|
assert ".secret.md" not in by_path
|
|
assert "node_modules/noise.md" not in by_path
|
|
|
|
md = client.get(
|
|
"/v1/sessions/unknown/artifacts/read", params={"path": "brief.md"}
|
|
).json()
|
|
assert md["ok"] is True
|
|
assert md["kind"] == "markdown"
|
|
assert md["content"].startswith("# Brief")
|
|
|
|
html = client.get(
|
|
"/v1/sessions/unknown/artifacts/read", params={"path": "page.html"}
|
|
).json()
|
|
assert html["ok"] is True
|
|
assert html["kind"] == "html"
|
|
assert "<h1>Preview</h1>" in html["content"]
|
|
|
|
|
|
def test_artifact_read_rejects_path_escape(tmp_path):
|
|
client = _client(tmp_path, [])
|
|
escaped = client.get(
|
|
"/v1/sessions/unknown/artifacts/read", params={"path": "../outside.md"}
|
|
).json()
|
|
assert escaped["ok"] is False
|
|
assert "escapes" in escaped["error"]
|
|
|
|
|
|
def test_sessions_hide_scheduled_internal_runs(tmp_path):
|
|
manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider([]))
|
|
manager.session_store.save(
|
|
SessionRecord(
|
|
session_id="normal",
|
|
workspace=str(tmp_path),
|
|
model="gpt-5.5",
|
|
mode="interactive",
|
|
messages=[{"role": "user", "content": "normal task"}],
|
|
title="Normal task",
|
|
agent="cowork",
|
|
)
|
|
)
|
|
manager.session_store.save(
|
|
SessionRecord(
|
|
session_id="__run__daily-news-1",
|
|
workspace=str(tmp_path),
|
|
model="gpt-5.5",
|
|
mode="interactive",
|
|
messages=[{"role": "user", "content": "scheduled run"}],
|
|
title="Daily news briefing",
|
|
agent="cowork",
|
|
)
|
|
)
|
|
manager.session_store.save(
|
|
SessionRecord(
|
|
session_id="__task__daily-news",
|
|
workspace=str(tmp_path),
|
|
model="gpt-5.5",
|
|
mode="interactive",
|
|
messages=[{"role": "user", "content": "scheduled task"}],
|
|
title="Daily news briefing",
|
|
agent="cowork",
|
|
)
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
session_ids = {
|
|
s["session_id"] for s in client.get("/v1/sessions").json()["sessions"]
|
|
}
|
|
assert "normal" in session_ids
|
|
assert "__run__daily-news-1" not in session_ids
|
|
assert "__task__daily-news" not in session_ids
|
|
|
|
|
|
def test_sessions_can_be_renamed_and_deleted(tmp_path):
|
|
manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider([]))
|
|
manager.session_store.save(
|
|
SessionRecord(
|
|
session_id="rename-me",
|
|
workspace=str(tmp_path),
|
|
model="gpt-5.5",
|
|
mode="interactive",
|
|
messages=[{"role": "user", "content": "original"}],
|
|
title="Original title",
|
|
agent="cowork",
|
|
)
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
|
|
renamed = client.patch(
|
|
"/v1/sessions/rename-me", json={"title": " Better title "}
|
|
).json()
|
|
assert renamed["ok"] is True
|
|
sessions = client.get("/v1/sessions").json()["sessions"]
|
|
assert any(
|
|
s["session_id"] == "rename-me" and s["title"] == "Better title"
|
|
for s in sessions
|
|
)
|
|
|
|
deleted = client.delete("/v1/sessions/rename-me").json()
|
|
assert deleted["ok"] is True
|
|
sessions = client.get("/v1/sessions").json()["sessions"]
|
|
assert all(s["session_id"] != "rename-me" for s in sessions)
|
|
assert client.get("/v1/sessions/rename-me/messages").json()["messages"] == []
|
|
|
|
|
|
def test_sessions_can_be_pinned_and_archived(tmp_path):
|
|
manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider([]))
|
|
for sid in ("older", "newer"):
|
|
manager.session_store.save(
|
|
SessionRecord(
|
|
session_id=sid,
|
|
workspace=str(tmp_path),
|
|
model="gpt-5.5",
|
|
mode="interactive",
|
|
messages=[{"role": "user", "content": sid}],
|
|
agent="cowork",
|
|
)
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
|
|
assert (
|
|
client.patch("/v1/sessions/older", json={"pinned": True}).json()["ok"] is True
|
|
)
|
|
sessions = client.get("/v1/sessions").json()["sessions"]
|
|
assert sessions[0]["session_id"] == "older" and sessions[0]["pinned"] is True
|
|
|
|
assert (
|
|
client.patch("/v1/sessions/newer", json={"archived": True}).json()["ok"] is True
|
|
)
|
|
by_id = {s["session_id"]: s for s in client.get("/v1/sessions").json()["sessions"]}
|
|
assert by_id["newer"]["archived"] is True
|
|
|
|
assert (
|
|
client.patch("/v1/sessions/older", json={"pinned": False}).json()["ok"] is True
|
|
)
|
|
assert (
|
|
client.patch("/v1/sessions/newer", json={"archived": False}).json()["ok"]
|
|
is True
|
|
)
|
|
by_id = {s["session_id"]: s for s in client.get("/v1/sessions").json()["sessions"]}
|
|
assert by_id["older"]["pinned"] is False and by_id["newer"]["archived"] is False
|
|
|
|
|
|
# -- WebSocket ------------------------------------------------------------------
|
|
|
|
|
|
def _drain(ws, on_permission=None):
|
|
"""Collect event types until turn_done; optionally answer permission_required."""
|
|
types = []
|
|
while True:
|
|
event = ws.receive_json()
|
|
types.append(event["type"])
|
|
if event["type"] == "permission_required" and on_permission:
|
|
ws.send_json({"type": "approval", "decision": on_permission})
|
|
if event["type"] == "turn_done":
|
|
return types
|
|
|
|
|
|
def test_ws_simple_turn(tmp_path):
|
|
client = _client(tmp_path, [_text("done thinking")])
|
|
with client.websocket_connect("/ws/session/s1") as ws:
|
|
assert ws.receive_json()["type"] == "ready"
|
|
ws.send_json({"type": "user_message", "text": "hello"})
|
|
types = _drain(ws)
|
|
assert "assistant_message" in types
|
|
assert "turn_end" in types
|
|
|
|
|
|
def test_ws_approval_round_trip(tmp_path):
|
|
client = _client(
|
|
tmp_path,
|
|
[
|
|
_tool("write_file", {"path": "made.py", "content": "print(1)\n"}),
|
|
_text("wrote it"),
|
|
],
|
|
)
|
|
with client.websocket_connect("/ws/session/s2") as ws:
|
|
assert ws.receive_json()["type"] == "ready"
|
|
ws.send_json({"type": "user_message", "text": "create made.py"})
|
|
types = _drain(ws, on_permission="once")
|
|
assert "permission_required" in types
|
|
assert "tool_finished" in types
|
|
assert (tmp_path / "made.py").read_text() == "print(1)\n"
|
|
|
|
|
|
def test_ws_session_persisted_while_parked_on_approval(tmp_path):
|
|
"""A crash mid-turn must not eat the conversation: by the time the engine parks on an
|
|
approval, the session (user message + assistant tool call) is already on disk."""
|
|
manager = SessionManager(
|
|
workspace=tmp_path,
|
|
provider=ScriptedProvider(
|
|
[_tool("write_file", {"path": "x.py", "content": "1\n"}), _text("done")]
|
|
),
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
with client.websocket_connect("/ws/session/persist1") as ws:
|
|
assert ws.receive_json()["type"] == "ready"
|
|
ws.send_json({"type": "user_message", "text": "make x.py"})
|
|
while ws.receive_json()["type"] != "permission_required":
|
|
pass
|
|
# Parked on the approval — nothing approved, turn far from done. Already saved?
|
|
rec = manager.session_store.load("persist1")
|
|
assert rec is not None
|
|
roles = [m.get("role") for m in rec.messages]
|
|
assert "user" in roles # turn_start checkpoint
|
|
assert "assistant" in roles # iteration progress checkpoint
|
|
ws.send_json({"type": "approval", "decision": "deny"})
|
|
while ws.receive_json()["type"] != "turn_done":
|
|
pass
|
|
|
|
|
|
def test_ws_browser_tool_audit_round_trip(tmp_path):
|
|
client = _client(tmp_path, [_tool("browser_close", {}), _text("closed")])
|
|
with client.websocket_connect("/ws/session/browser-audit?agent=cowork") as ws:
|
|
assert ws.receive_json()["type"] == "ready"
|
|
ws.send_json({"type": "user_message", "text": "close browser"})
|
|
types = _drain(ws, on_permission="once")
|
|
assert "permission_required" in types
|
|
assert "tool_finished" in types
|
|
|
|
rows = client.get(
|
|
"/v1/audit", params={"session_id": "browser-audit", "connector": "browser"}
|
|
).json()["events"]
|
|
assert any(
|
|
r["tool"] == "browser_close" and r["stage"] == "approval_resolved" for r in rows
|
|
)
|
|
assert any(r["tool"] == "browser_close" and r["stage"] == "finished" for r in rows)
|
|
|
|
|
|
def test_open_and_recent_workspaces(tmp_path):
|
|
proj = tmp_path / "proj"
|
|
proj.mkdir()
|
|
client = _client(tmp_path, [])
|
|
opened = client.post("/v1/workspaces/open", json={"path": str(proj)}).json()
|
|
assert opened["ok"] is True
|
|
recents = client.get("/v1/workspaces/recent").json()["workspaces"]
|
|
assert any(w["path"] == str(proj.resolve()) for w in recents)
|
|
|
|
|
|
def test_open_invalid_workspace(tmp_path):
|
|
client = _client(tmp_path, [])
|
|
bad = client.post(
|
|
"/v1/workspaces/open", json={"path": str(tmp_path / "nope")}
|
|
).json()
|
|
assert bad["ok"] is False
|
|
|
|
|
|
def test_open_workspace_create(tmp_path):
|
|
client = _client(tmp_path, [])
|
|
fresh = tmp_path / "fresh-project"
|
|
assert not fresh.exists()
|
|
res = client.post(
|
|
"/v1/workspaces/open", json={"path": str(fresh), "create": True}
|
|
).json()
|
|
assert res["ok"] is True
|
|
assert fresh.is_dir()
|
|
|
|
|
|
def test_ws_requires_workspace_when_no_default(tmp_path):
|
|
# Manager with no default workspace: a session with no folder is rejected.
|
|
manager = SessionManager(
|
|
workspace=None, data_dir=tmp_path, provider=ScriptedProvider([])
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
with client.websocket_connect("/ws/session/nofolder") as ws:
|
|
first = ws.receive_json()
|
|
assert first["type"] == "error"
|
|
assert "workspace" in first["data"]["error"]
|
|
|
|
|
|
def test_ws_with_workspace_query(tmp_path):
|
|
from urllib.parse import quote
|
|
|
|
proj = tmp_path / "proj"
|
|
proj.mkdir()
|
|
manager = SessionManager(
|
|
workspace=None,
|
|
data_dir=tmp_path,
|
|
provider=ScriptedProvider([_text("hi from proj")]),
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
with client.websocket_connect(f"/ws/session/s?workspace={quote(str(proj))}") as ws:
|
|
ready = ws.receive_json()
|
|
assert ready["type"] == "ready"
|
|
assert ready["data"]["workspace"] == str(proj.resolve())
|
|
ws.send_json({"type": "user_message", "text": "hello"})
|
|
assert "turn_end" in _drain(ws)
|
|
|
|
|
|
def test_ws_chat_agent_needs_no_workspace(tmp_path):
|
|
manager = SessionManager(
|
|
workspace=None,
|
|
data_dir=tmp_path,
|
|
provider=ScriptedProvider([_text("hi from chat")]),
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
with client.websocket_connect("/ws/session/chat1?agent=chat") as ws:
|
|
ready = ws.receive_json()
|
|
assert ready["type"] == "ready"
|
|
assert ready["data"]["agent"] == "chat"
|
|
assert ready["data"]["workspace"] is None
|
|
ws.send_json({"type": "user_message", "text": "hello"})
|
|
assert "turn_end" in _drain(ws)
|
|
|
|
|
|
def test_ws_set_mode_auto_skips_approval(tmp_path):
|
|
from urllib.parse import quote
|
|
|
|
proj = tmp_path / "proj"
|
|
proj.mkdir()
|
|
manager = SessionManager(
|
|
workspace=None,
|
|
data_dir=tmp_path,
|
|
provider=ScriptedProvider(
|
|
[_tool("write_file", {"path": "a.py", "content": "x"}), _text("done")]
|
|
),
|
|
)
|
|
client = TestClient(create_app(manager))
|
|
with client.websocket_connect(f"/ws/session/sm?workspace={quote(str(proj))}") as ws:
|
|
assert ws.receive_json()["type"] == "ready"
|
|
ws.send_json({"type": "set_mode", "mode": "auto"})
|
|
ws.send_json({"type": "user_message", "text": "write a.py"})
|
|
types = _drain(ws) # no approval handler — would hang if it asked
|
|
assert "permission_required" not in types
|
|
assert (proj / "a.py").read_text() == "x"
|
|
|
|
|
|
def test_ws_session_resume_via_store(tmp_path):
|
|
# First connection runs a turn and persists the session.
|
|
client = _client(tmp_path, [_text("first answer")])
|
|
with client.websocket_connect("/ws/session/keep") as ws:
|
|
ws.receive_json()
|
|
ws.send_json({"type": "user_message", "text": "remember this"})
|
|
_drain(ws)
|
|
# The session is now listed via REST.
|
|
sessions = client.get("/v1/sessions").json()["sessions"]
|
|
assert any(s["session_id"] == "keep" and s["messages"] > 0 for s in sessions)
|