test(routers): PR6 T6.7 — POST /api/threads workspace_id integration

`create_thread` in `app/gateway/routers/threads.py` already relies on
the AUTO sentinel default for ``ThreadMetaStore.create``, so once T6.1
landed the workspace_id was implicitly carried from the contextvar.
This commit adds the integration coverage: TestClient + stub auth
middleware + ``MemoryThreadMetaStore`` round-trip showing the persisted
row carries the right ``workspace_id`` under two distinct workspaces.

No production code change — the route wiring was correct, but the
guarantee was previously only proven at the unit level.
This commit is contained in:
1445043649
2026-05-13 17:35:27 +08:00
parent 0456606dc1
commit f6a922921b
@@ -0,0 +1,59 @@
"""PR6 T6.7 — POST /api/threads writes workspace_id from contextvar.
The `routers/threads.py:create_thread` path delegates to
``ThreadMetaStore.create`` *without* an explicit ``workspace_id`` — it
relies on the AUTO sentinel pulling the value from the active workspace
contextvar that AuthMiddleware (or the test stub) sets. This test
covers the integration through the FastAPI TestClient stack: post a
thread under workspace A, then verify the persisted record carries
``workspace_id="ws-alpha"``.
"""
from __future__ import annotations
from collections.abc import Callable
from _router_auth_helpers import make_authed_test_app
from fastapi.testclient import TestClient
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from app.gateway.auth.models import ActiveWorkspace
from app.gateway.routers import threads
from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
def _workspace_factory(wid: str) -> Callable[[], ActiveWorkspace]:
def _factory() -> ActiveWorkspace:
return ActiveWorkspace(id=wid, role="owner")
return _factory
def _build_app(workspace_id: str):
app = make_authed_test_app(workspace_factory=_workspace_factory(workspace_id))
store = InMemoryStore()
checkpointer = InMemorySaver()
app.state.store = store
app.state.checkpointer = checkpointer
app.state.thread_store = MemoryThreadMetaStore(store)
app.include_router(threads.router)
return app, store
def test_post_thread_stamps_workspace_id_from_contextvar():
app, store = _build_app("ws-alpha")
with TestClient(app) as client:
response = client.post("/api/threads", json={"thread_id": "t1", "metadata": {}})
assert response.status_code == 200, response.text
item = store.get(("threads",), "t1")
assert item is not None
assert item.value["workspace_id"] == "ws-alpha"
def test_post_thread_under_different_workspace():
app, store = _build_app("ws-beta")
with TestClient(app) as client:
client.post("/api/threads", json={"thread_id": "t2", "metadata": {}})
assert store.get(("threads",), "t2").value["workspace_id"] == "ws-beta"