File size: 1,756 Bytes
f4111a5 96341d9 f4111a5 96341d9 f4111a5 aec7208 f4111a5 aec7208 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | from types import SimpleNamespace
from src.agents.middlewares.thread_data_middleware import ThreadDataMiddleware
from src.utils.runtime import get_runtime_context, get_runtime_thread_id
class DictLike:
def __init__(self, data):
self._data = data
def get(self, key, default=None):
return self._data.get(key, default)
def items(self):
return self._data.items()
def test_get_runtime_context_returns_empty_dict_for_none_context():
runtime = SimpleNamespace(context=None, config={})
assert get_runtime_context(runtime) == {}
def test_get_runtime_thread_id_falls_back_to_metadata_when_context_missing():
runtime = SimpleNamespace(context=None, config={"metadata": {"thread_id": "thread-meta-1"}})
assert get_runtime_thread_id(runtime) == "thread-meta-1"
def test_get_runtime_thread_id_supports_dict_like_runtime_config():
runtime = SimpleNamespace(
context=DictLike({}),
config=DictLike({"metadata": DictLike({"thread_id": "thread-meta-like"})}),
)
assert get_runtime_thread_id(runtime) == "thread-meta-like"
def test_thread_data_middleware_uses_metadata_thread_id(tmp_path):
middleware = ThreadDataMiddleware(base_dir=str(tmp_path))
runtime = SimpleNamespace(context=None, config={"metadata": {"thread_id": "thread-meta-2"}})
result = middleware.before_agent(state={}, runtime=runtime)
assert result is not None
assert "thread-meta-2" in result["thread_data"]["workspace_path"]
def test_thread_data_middleware_returns_none_without_any_thread_id(tmp_path):
middleware = ThreadDataMiddleware(base_dir=str(tmp_path))
runtime = SimpleNamespace(context=None, config={})
assert middleware.before_agent(state={}, runtime=runtime) is None
|