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