feat(runtime): add workspace_context module + tests

新建 backend/packages/harness/deerflow/runtime/workspace_context.py，仿 user_context.py 的 API 形态： - CurrentWorkspace Protocol（要求 .id: str + .role: str） - _current_workspace ContextVar + set/reset/get/require - DEFAULT_WORKSPACE_ID = "default" + get_effective_workspace_id（fallback 友好，不抛错；用于文件系统路径） - AUTO 哨兵 + resolve_workspace_id 三态（AUTO/str/None）与 user_context 的区别：CurrentWorkspace 额外要求 .role 字段，让 Protocol 同时约束"workspace 是哪个"和"caller 在该 workspace 内的角色"（Stage 0 只见 'owner'， Stage 2 RBAC 打开 admin/member）。 backend/tests/test_workspace_context.py 16 个 test，覆盖： - 4 个 set/reset/require 行为 - 3 个 Protocol structural check（接受 .id+.role / 拒少 .role / 拒少 .id） - 4 个 get_effective_workspace_id（含 UUID → str 强转） - 5 个 resolve_workspace_id 三态（AUTO/AUTO 无 ctx 抛错/explicit str/explicit None/AUTO 强转 str） Stage 0 PR3 T3.1 + T3.2。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 21:04:49 +08:00
parent cbbb83a706
commit f63089aea8
2 changed files with 353 additions and 0 deletions
@@ -0,0 +1,171 @@
+"""Tests for runtime.workspace_context — workspace contextvar semantics.
+
+Mirrors :mod:`test_user_context` but for the workspace contextvar
+introduced in Stage 0 PR3. No autouse workspace fixture exists yet
+(PR4 will add it together with the AuthMiddleware injection), so these
+tests run against a clean contextvar.
+"""
+
+import uuid
+from types import SimpleNamespace
+
+import pytest
+
+from deerflow.runtime.workspace_context import (
+    AUTO,
+    DEFAULT_WORKSPACE_ID,
+    CurrentWorkspace,
+    get_current_workspace,
+    get_effective_workspace_id,
+    require_current_workspace,
+    reset_current_workspace,
+    resolve_workspace_id,
+    set_current_workspace,
+)
+
+# ---------------------------------------------------------------------------
+# get_current_workspace / require_current_workspace / set+reset round-trip
+# ---------------------------------------------------------------------------
+
+
+def test_default_is_none():
+    """Before any set, contextvar returns None."""
+    assert get_current_workspace() is None
+
+
+def test_set_and_reset_roundtrip():
+    """set_current_workspace returns a token that reset restores."""
+    workspace = SimpleNamespace(id="ws-1", role="owner")
+    token = set_current_workspace(workspace)
+    try:
+        assert get_current_workspace() is workspace
+    finally:
+        reset_current_workspace(token)
+    assert get_current_workspace() is None
+
+
+def test_require_current_workspace_raises_when_unset():
+    """require_current_workspace raises RuntimeError if contextvar is unset."""
+    assert get_current_workspace() is None
+    with pytest.raises(RuntimeError, match="without workspace context"):
+        require_current_workspace()
+
+
+def test_require_current_workspace_returns_workspace_when_set():
+    """require_current_workspace returns the workspace when contextvar is set."""
+    workspace = SimpleNamespace(id="ws-2", role="admin")
+    token = set_current_workspace(workspace)
+    try:
+        assert require_current_workspace() is workspace
+    finally:
+        reset_current_workspace(token)
+
+
+# ---------------------------------------------------------------------------
+# CurrentWorkspace Protocol — must require BOTH .id and .role
+# ---------------------------------------------------------------------------
+
+
+def test_protocol_accepts_id_and_role():
+    """CurrentWorkspace is satisfied by any object with .id and .role."""
+    workspace = SimpleNamespace(id="ws-3", role="member")
+    assert isinstance(workspace, CurrentWorkspace)
+
+
+def test_protocol_rejects_missing_role():
+    """An object with only .id (no .role) is NOT a workspace."""
+    user_shaped = SimpleNamespace(id="ws-4")
+    assert not isinstance(user_shaped, CurrentWorkspace)
+
+
+def test_protocol_rejects_no_id():
+    """An object without .id does not satisfy CurrentWorkspace."""
+    not_a_workspace = SimpleNamespace(role="owner")
+    assert not isinstance(not_a_workspace, CurrentWorkspace)
+
+
+# ---------------------------------------------------------------------------
+# get_effective_workspace_id / DEFAULT_WORKSPACE_ID tests
+# ---------------------------------------------------------------------------
+
+
+def test_default_workspace_id_is_default():
+    assert DEFAULT_WORKSPACE_ID == "default"
+
+
+def test_effective_workspace_id_returns_default_when_no_workspace():
+    """No workspace in context -> fallback to DEFAULT_WORKSPACE_ID."""
+    assert get_effective_workspace_id() == "default"
+
+
+def test_effective_workspace_id_returns_workspace_id_when_set():
+    workspace = SimpleNamespace(id="ws-abc-123", role="owner")
+    token = set_current_workspace(workspace)
+    try:
+        assert get_effective_workspace_id() == "ws-abc-123"
+    finally:
+        reset_current_workspace(token)
+
+
+def test_effective_workspace_id_coerces_to_str():
+    """workspace.id might be a UUID object; must come back as str."""
+    wid = uuid.uuid4()
+    workspace = SimpleNamespace(id=wid, role="owner")
+    token = set_current_workspace(workspace)
+    try:
+        assert get_effective_workspace_id() == str(wid)
+    finally:
+        reset_current_workspace(token)
+
+
+# ---------------------------------------------------------------------------
+# resolve_workspace_id three-state semantics
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_auto_reads_from_contextvar():
+    workspace = SimpleNamespace(id="ws-resolve-1", role="owner")
+    token = set_current_workspace(workspace)
+    try:
+        assert resolve_workspace_id(AUTO) == "ws-resolve-1"
+    finally:
+        reset_current_workspace(token)
+
+
+def test_resolve_auto_raises_when_unset():
+    assert get_current_workspace() is None
+    with pytest.raises(RuntimeError, match="workspace_id=AUTO but no workspace"):
+        resolve_workspace_id(AUTO, method_name="TestRepo.search")
+
+
+def test_resolve_explicit_str_overrides_contextvar():
+    workspace = SimpleNamespace(id="ws-ctx", role="owner")
+    token = set_current_workspace(workspace)
+    try:
+        # Explicit value beats contextvar — admin override / test path.
+        assert resolve_workspace_id("ws-explicit") == "ws-explicit"
+    finally:
+        reset_current_workspace(token)
+
+
+def test_resolve_explicit_none_means_no_filter():
+    workspace = SimpleNamespace(id="ws-ctx-2", role="owner")
+    token = set_current_workspace(workspace)
+    try:
+        # Explicit None opts out of workspace filtering (migration scripts).
+        assert resolve_workspace_id(None) is None
+    finally:
+        reset_current_workspace(token)
+
+
+def test_resolve_auto_coerces_uuid_to_str():
+    """resolve_workspace_id with AUTO returns str even if workspace.id is UUID."""
+    wid = uuid.uuid4()
+    workspace = SimpleNamespace(id=wid, role="owner")
+    token = set_current_workspace(workspace)
+    try:
+        resolved = resolve_workspace_id(AUTO)
+        assert resolved == str(wid)
+        assert isinstance(resolved, str)
+    finally:
+        reset_current_workspace(token)