mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
ea74367502
* fix(runtime): honor LangGraph Server identity for user-scoped data * fix(runtime): scope custom agent SOUL by resolved user
329 lines
9.7 KiB
Python
329 lines
9.7 KiB
Python
"""Tests for runtime.user_context — contextvar three-state semantics.
|
|
|
|
These tests opt out of the autouse contextvar fixture (added in
|
|
commit 6) because they explicitly test the cases where the contextvar
|
|
is set or unset.
|
|
"""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from langchain_core.runnables import RunnableLambda
|
|
from langgraph.runtime import Runtime, ServerInfo
|
|
|
|
from deerflow.config.paths import Paths, make_safe_user_id
|
|
from deerflow.runtime.user_context import (
|
|
DEFAULT_USER_ID,
|
|
CurrentUser,
|
|
get_current_user,
|
|
get_effective_user_id,
|
|
require_current_user,
|
|
reset_current_user,
|
|
resolve_config_user_id,
|
|
resolve_runtime_user_id,
|
|
set_current_user,
|
|
)
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_default_is_none():
|
|
"""Before any set, contextvar returns None."""
|
|
assert get_current_user() is None
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_set_and_reset_roundtrip():
|
|
"""set_current_user returns a token that reset restores."""
|
|
user = SimpleNamespace(id="user-1")
|
|
token = set_current_user(user)
|
|
try:
|
|
assert get_current_user() is user
|
|
finally:
|
|
reset_current_user(token)
|
|
assert get_current_user() is None
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_require_current_user_raises_when_unset():
|
|
"""require_current_user raises RuntimeError if contextvar is unset."""
|
|
assert get_current_user() is None
|
|
with pytest.raises(RuntimeError, match="without user context"):
|
|
require_current_user()
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_require_current_user_returns_user_when_set():
|
|
"""require_current_user returns the user when contextvar is set."""
|
|
user = SimpleNamespace(id="user-2")
|
|
token = set_current_user(user)
|
|
try:
|
|
assert require_current_user() is user
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_protocol_accepts_duck_typed():
|
|
"""CurrentUser is a runtime_checkable Protocol matching any .id-bearing object."""
|
|
user = SimpleNamespace(id="user-3")
|
|
assert isinstance(user, CurrentUser)
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_protocol_rejects_no_id():
|
|
"""Objects without .id do not satisfy CurrentUser Protocol."""
|
|
not_a_user = SimpleNamespace(email="no-id@example.com")
|
|
assert not isinstance(not_a_user, CurrentUser)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_effective_user_id / DEFAULT_USER_ID tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_default_user_id_is_default():
|
|
assert DEFAULT_USER_ID == "default"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_effective_user_id_returns_default_when_no_user():
|
|
"""No user in context -> fallback to DEFAULT_USER_ID."""
|
|
assert get_effective_user_id() == "default"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_effective_user_id_returns_user_id_when_set():
|
|
user = SimpleNamespace(id="u-abc-123")
|
|
token = set_current_user(user)
|
|
try:
|
|
assert get_effective_user_id() == "u-abc-123"
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_effective_user_id_coerces_to_str():
|
|
"""User.id might be a UUID object; must come back as str."""
|
|
import uuid
|
|
|
|
uid = uuid.uuid4()
|
|
|
|
user = SimpleNamespace(id=uid)
|
|
token = set_current_user(user)
|
|
try:
|
|
assert get_effective_user_id() == str(uid)
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_runtime_user_id tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_config_user_id_prefers_server_auth_over_client_user_id():
|
|
config = {
|
|
"configurable": {
|
|
"langgraph_auth_user_id": "authenticated-user",
|
|
"user_id": "spoofed-configurable-user",
|
|
},
|
|
"context": {
|
|
"user_id": "spoofed-context-user",
|
|
},
|
|
}
|
|
|
|
assert resolve_config_user_id(config) == "authenticated-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_config_user_id_normalizes_external_auth_identity_for_storage(tmp_path):
|
|
raw_identity = "alice@example.com"
|
|
|
|
resolved = resolve_config_user_id(
|
|
{
|
|
"configurable": {
|
|
"langgraph_auth_user_id": raw_identity,
|
|
}
|
|
}
|
|
)
|
|
|
|
assert resolved == make_safe_user_id(raw_identity)
|
|
assert Paths(tmp_path).user_dir(resolved).name == resolved
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_config_user_id_keeps_colliding_sanitized_auth_identities_distinct():
|
|
dotted = resolve_config_user_id({"configurable": {"langgraph_auth_user_id": "alice.example"}})
|
|
slashed = resolve_config_user_id({"configurable": {"langgraph_auth_user_id": "alice/example"}})
|
|
|
|
assert dotted == make_safe_user_id("alice.example")
|
|
assert slashed == make_safe_user_id("alice/example")
|
|
assert dotted != slashed
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
@pytest.mark.parametrize("invalid_auth_user_id", [123, object()])
|
|
def test_config_user_id_ignores_non_string_server_auth_id(invalid_auth_user_id):
|
|
config = {
|
|
"configurable": {
|
|
"langgraph_auth_user_id": invalid_auth_user_id,
|
|
},
|
|
"context": {
|
|
"user_id": "gateway-user",
|
|
},
|
|
}
|
|
|
|
assert resolve_config_user_id(config) == "gateway-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_config_user_id_uses_gateway_runtime_context_without_server_auth():
|
|
config = {
|
|
"configurable": {
|
|
"user_id": "legacy-configurable-user",
|
|
},
|
|
"context": {
|
|
"user_id": "gateway-user",
|
|
},
|
|
}
|
|
|
|
assert resolve_config_user_id(config) == "gateway-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_config_user_id_falls_back_to_legacy_configurable_user():
|
|
assert resolve_config_user_id({"configurable": {"user_id": "legacy-user"}}) == "legacy-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_langgraph_auth_takes_precedence_over_context_user_id(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"langgraph.config.get_config",
|
|
lambda: {"configurable": {"langgraph_auth_user_id": "langgraph-user"}},
|
|
)
|
|
|
|
runtime = SimpleNamespace(context={"user_id": "runtime-user"})
|
|
|
|
assert resolve_runtime_user_id(runtime) == "langgraph-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_user_id_uses_gateway_context_without_langgraph_auth(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"langgraph.config.get_config",
|
|
lambda: {"configurable": {}},
|
|
)
|
|
|
|
runtime = SimpleNamespace(context={"user_id": "runtime-user"})
|
|
|
|
assert resolve_runtime_user_id(runtime) == "runtime-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_server_info_identity_takes_precedence_over_runtime_context(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"langgraph.config.get_config",
|
|
lambda: {"configurable": {"langgraph_auth_user_id": "config-auth-user"}},
|
|
)
|
|
runtime = Runtime(
|
|
server_info=ServerInfo(
|
|
assistant_id="assistant-1",
|
|
graph_id="graph-1",
|
|
user=SimpleNamespace(identity="server-info-user"),
|
|
),
|
|
context={"user_id": "runtime-user"},
|
|
)
|
|
|
|
assert resolve_runtime_user_id(runtime) == "server-info-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_server_info_normalizes_external_auth_identity(monkeypatch):
|
|
monkeypatch.setattr("langgraph.config.get_config", lambda: {"configurable": {}})
|
|
raw_identity = "alice@example.com"
|
|
runtime = Runtime(
|
|
server_info=ServerInfo(
|
|
assistant_id="assistant-1",
|
|
graph_id="graph-1",
|
|
user=SimpleNamespace(identity=raw_identity),
|
|
),
|
|
)
|
|
|
|
assert resolve_runtime_user_id(runtime) == make_safe_user_id(raw_identity)
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_server_info_ignores_non_string_identity():
|
|
runtime = Runtime(
|
|
server_info=ServerInfo(
|
|
assistant_id="assistant-1",
|
|
graph_id="graph-1",
|
|
user=SimpleNamespace(identity=object()),
|
|
),
|
|
context={"user_id": "runtime-user"},
|
|
)
|
|
|
|
assert resolve_runtime_user_id(runtime) == "runtime-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_user_id_uses_langgraph_auth_user_id(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"langgraph.config.get_config",
|
|
lambda: {"configurable": {"langgraph_auth_user_id": "langgraph-user"}},
|
|
)
|
|
|
|
assert resolve_runtime_user_id(None) == "langgraph-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_user_id_reads_langgraph_auth_from_real_runnable_config():
|
|
resolver = RunnableLambda(lambda _: resolve_runtime_user_id(None))
|
|
raw_identity = "runnable@example.com"
|
|
|
|
result = resolver.invoke(
|
|
None,
|
|
config={
|
|
"configurable": {
|
|
"langgraph_auth_user_id": raw_identity,
|
|
}
|
|
},
|
|
)
|
|
|
|
assert result == make_safe_user_id(raw_identity)
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_user_id_falls_back_to_langgraph_auth_user_identity(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"langgraph.config.get_config",
|
|
lambda: {
|
|
"configurable": {
|
|
"langgraph_auth_user": SimpleNamespace(identity="identity-user"),
|
|
}
|
|
},
|
|
)
|
|
|
|
assert resolve_runtime_user_id(None) == "identity-user"
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_user_id_falls_back_to_contextvar(monkeypatch):
|
|
monkeypatch.setattr("langgraph.config.get_config", lambda: {"configurable": {}})
|
|
token = set_current_user(SimpleNamespace(id="context-user"))
|
|
try:
|
|
assert resolve_runtime_user_id(None) == "context-user"
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
|
|
@pytest.mark.no_auto_user
|
|
def test_runtime_user_id_falls_back_to_default_outside_runnable_context(monkeypatch):
|
|
def raise_no_config():
|
|
raise RuntimeError("no runnable config")
|
|
|
|
monkeypatch.setattr("langgraph.config.get_config", raise_no_config)
|
|
|
|
assert resolve_runtime_user_id(None) == DEFAULT_USER_ID
|