backend/tests/test_runtime_thread_id_fallback.py

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