Function daemon_sock

tests/test_daemon.py:56–102

Start a daemon once per session and return the socket path.

()

@pytest.fixture(scope="session")
def daemon_sock() -> Iterator[str]:
    """Start a daemon once per session and return the socket path."""
    import cocoindex_code.daemon as dm

    # Use a short path to stay within AF_UNIX limit
    user_dir = Path(tempfile.mkdtemp(prefix="ccc_d_"))
    user_dir.mkdir(parents=True, exist_ok=True)

    # Use COCOINDEX_CODE_DIR env var for isolation instead of direct module patching.
    # Direct patching of dm.user_settings_dir leaks across test modules and causes
    # stop_daemon() in other fixtures to read the wrong PID file (pytest's own PID).
    old_env = os.environ.get("COCOINDEX_CODE_DIR")
    os.environ["COCOINDEX_CODE_DIR"] = str(user_dir)

    save_user_settings(make_test_user_settings())

    thread = threading.Thread(target=dm.run_daemon, daemon=True)
    thread.start()

    sock_path = dm.daemon_socket_path()

    deadline = time.monotonic() + 20
    while time.monotonic() < deadline:
        if os.path.exists(sock_path):
            break
        time.sleep(0.1)
    else:
        raise TimeoutError("Daemon did not start")

    yield sock_path

    # Gracefully shut down the daemon thread so named pipes are released on Windows
    try:
        conn = Client(sock_path, family=connection_family())
        conn.send_bytes(encode_request(HandshakeRequest(version=__version__)))
        conn.recv_bytes()
        conn.send_bytes(encode_request(StopRequest()))
        conn.recv_bytes()
        conn.close()
    except Exception:
        pass
    thread.join(timeout=5)

    if old_env is None:
        os.environ.pop("COCOINDEX_CODE_DIR", None)
    else:
        os.environ["COCOINDEX_CODE_DIR"] = old_env


def _recv_index_response(conn: Connection) -> tuple[list[IndexProgressUpdate], IndexResponse]: