"""#5 regression: _session_has_compression_in_flight must offload both blocking sources to thread pool."""
import threading
from unittest.mock import MagicMock

import pytest


def _make_runner(holder_value=None, record_thread=False, thread_sink=None):
    from gateway.run import GatewayRunner
    runner = GatewayRunner.__new__(GatewayRunner)

    store = MagicMock()
    store._lock = threading.Lock()
    store._loaded = True
    store._entries = {"k": MagicMock(session_id="sess-123")}
    store._ensure_loaded_locked = lambda: None
    runner.session_store = store

    raw_db = MagicMock()
    if record_thread and thread_sink is not None:
        def _holder(sid):
            thread_sink["thread"] = threading.get_ident()
            return holder_value
        raw_db.get_compression_lock_holder = _holder
    else:
        raw_db.get_compression_lock_holder = MagicMock(return_value=holder_value)

    session_db = MagicMock()
    session_db._db = raw_db
    runner._session_db = session_db
    return runner


@pytest.mark.asyncio
async def test_returns_false_when_no_session_store():
    from gateway.run import GatewayRunner
    runner = GatewayRunner.__new__(GatewayRunner)
    runner.session_store = None
    runner._session_db = MagicMock()
    assert await runner._session_has_compression_in_flight("k") is False


@pytest.mark.asyncio
async def test_returns_false_when_holder_is_not_a_string():
    """Lock holders are session-id strings. A MagicMock auto-attr must not
    look like an in-flight compression and skip hygiene (#96953)."""
    runner = _make_runner(holder_value=MagicMock())
    assert await runner._session_has_compression_in_flight("k") is False
    runner = _make_runner(holder_value=True)
    assert await runner._session_has_compression_in_flight("k") is False
    runner = _make_runner(holder_value="")
    assert await runner._session_has_compression_in_flight("k") is False
    runner = _make_runner(holder_value="agent-1")
    assert await runner._session_has_compression_in_flight("k") is True


@pytest.mark.asyncio
async def test_db_call_runs_off_event_loop():
    """Regression core: get_compression_lock_holder MUST execute in non-event-loop thread."""
    sink = {}
    runner = _make_runner(holder_value="agent-1", record_thread=True, thread_sink=sink)
    loop_thread = threading.get_ident()
    await runner._session_has_compression_in_flight("k")
    assert "thread" in sink, "underlying db.get_compression_lock_holder was not called"
    assert sink["thread"] != loop_thread, (
        "DB call still on event loop thread — #5 NOT fixed (to_thread not applied)"
    )