mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-04-27 12:18:14 +00:00
112 lines
3.1 KiB
Python
112 lines
3.1 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 deerflow.runtime.user_context import (
|
|
DEFAULT_USER_ID,
|
|
CurrentUser,
|
|
get_current_user,
|
|
get_effective_user_id,
|
|
require_current_user,
|
|
reset_current_user,
|
|
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)
|