Testing

Unit tests that never open a Bloomberg session.

xbbg.testing serves canned tables through xbbg's request middleware, so the code under test calls the real xbbg API and gets back data you control.

What xbbg.testing provides

Name	Signature	Purpose
create_mock_response	(*, service, operation, data, service_xml=None, event_type=None, correlation_ids=None, request_id=None)	Build a canned response for one service and operation
MockResponse	dataclass: service, operation, table, event=None, request_operation=None, metadata={}	The canned response; table is a native xbbg ArrowTable
mock_engine	(responses, *, strict=True)	Context manager that serves the responses to matching requests
create_mock_event	(event_type)	Bloomberg SDK test Event (needs blpapi)
deserialize_service	(service_xml)	Bloomberg SDK service from a schema XML string (needs blpapi)
get_admin_message_definition	(message_name)	Admin message definition such as "SessionStarted" (needs blpapi)
append_message_dict	(event, element_def, payload, properties=None)	Append a message built from a dict to a test event (needs blpapi)

Import the helpers from xbbg.testing. The examples use pytest, but nothing in the module depends on it.

How mock_engine works

mock_engine is a synchronous context manager that registers a request middleware on entry and removes it on exit. For each request that reaches the middleware chain it:

1. Finds the first queued MockResponse whose service equals the request's service and whose operation equals the request's operation (for raw requests, request_operation). Securities, fields, and dates are not compared, so queue responses in the order your code makes the calls.
2. Removes that response from the queue: each response is served once.
3. Returns its table, converted to the request's backend like a live result, and sets context.metadata["mocked"] = True and context.metadata["mock_response"] for other middleware.

If nothing matches, strict=True (the default) raises LookupError("No mock response matched <service>::<operation>"). With strict=False the request continues down the chain to the real engine. Besides MockResponse objects, the queue accepts a native ArrowTable, an ArrowRecordBatch, or a list of row dicts; these match any request, but only while they are first in the queue.

Your first test

import xbbg
from xbbg.testing import create_mock_response, mock_engine


def test_reference_request():
    response = create_mock_response(
        service="//blp/refdata",
        operation="ReferenceDataRequest",
        data={"AAPL US Equity": {"PX_LAST": 101.23, "VOLUME": 1200}},
    )

    with mock_engine([response]):
        table = xbbg.request(
            "//blp/refdata",
            "ReferenceDataRequest",
            securities=["AAPL US Equity"],
            fields=["PX_LAST", "VOLUME"],
            backend="native",
        )

    assert table.to_pylist() == [
        {"ticker": "AAPL US Equity", "field": "PX_LAST", "value": "101.23"},
        {"ticker": "AAPL US Equity", "field": "VOLUME", "value": "1200"},
    ]

xbbg.request() and xbbg.arequest() go straight to the middleware chain, so this test runs fully offline. Pass backend= explicitly in tests: without it a mocked request() returns the configured default backend (a Narwhals DataFrame), while a live request() returns the native Arrow table.

Shaping canned data

data is turned into rows by operation:

operation	data	Rows
ReferenceDataRequest	{ticker: {field: value}}	ticker, field, value
HistoricalDataRequest	{ticker: {date: {field: value}}} or {ticker: [{"date": date, field: value}]}	ticker, date, field, value
Any operation	A list of row dicts	Exactly those rows

In the two mapping forms every value and date is converted to a string, mirroring xbbg's default long output; None stays null. The security column takes the engine's key column name: ticker by default, or security after configure(key_column="security"). Explicit row dicts keep their Python types and column names, and any other operation (BQL, BSRCH, field info, intraday bars and ticks, ...) needs explicit rows, otherwise ValueError is raised. response.metadata["rows"] holds the rows that were built.

import xbbg
from xbbg.testing import create_mock_response, mock_engine


def test_history_as_pandas():
    response = create_mock_response(
        service="//blp/refdata",
        operation="HistoricalDataRequest",
        data={
            "SPX Index": {
                "2024-01-02": {"PX_LAST": 4742.83},
                "2024-01-03": {"PX_LAST": 4704.81},
            },
        },
    )

    with mock_engine([response]):
        frame = xbbg.request(
            "//blp/refdata",
            "HistoricalDataRequest",
            securities=["SPX Index"],
            fields=["PX_LAST"],
            start_date="2024-01-02",
            end_date="2024-01-03",
            backend="pandas",
        )

    assert list(frame.columns) == ["ticker", "date", "field", "value"]
    assert frame["value"].tolist() == ["4742.83", "4704.81"]


def test_bql_with_explicit_rows():
    response = create_mock_response(
        service="//blp/bqlsvc",
        operation="sendQuery",
        data=[{"ID": "AAPL US Equity", "px_last": 101.23}],
    )

    with mock_engine([response]):
        frame = xbbg.bql("get(px_last) for('AAPL US Equity')", backend="pandas")

    assert frame.to_dict("records") == [{"ID": "AAPL US Equity", "px_last": 101.23}]

bql() builds its request without contacting the engine, so it can be mocked offline like request().

Typed helpers such as bdp and bdh

Since xbbg 1.8.0, bdp() and bdh() resolve field types in the request terminal, after the middleware chain, so mock_engine answers them without starting an engine or querying //blp/apiflds:

import xbbg
from xbbg.testing import create_mock_response, mock_engine


def test_bdp_offline():
    response = create_mock_response(
        service="//blp/refdata",
        operation="ReferenceDataRequest",
        data={"AAPL US Equity": {"PX_LAST": 101.23}},
    )

    with mock_engine([response]):
        frame = xbbg.bdp("AAPL US Equity", "PX_LAST", backend="pandas")

    assert frame.to_dict("records") == [
        {"ticker": "AAPL US Equity", "field": "PX_LAST", "value": "101.23"},
    ]

• Keep typed calls free of Bloomberg overrides and extra request-element keywords in offline tests. With them, bdp, bdh, bds, bdib, bdtick, bqr, beqs, blkp, bport, bcurves, and bgovts look up the request schema before the middleware runs, which starts an engine and, with no Bloomberg reachable, waits out its connection attempts.
• Middleware sees only the caller's field_types in context.params.field_types; context.field_type_default names the default type the terminal would apply.
• In xbbg 1.7.0 and earlier, bdp() and bdh() resolved field types before the middleware ran, so a mocked call still started a real engine and failed offline with BlpInternalError. Upgrade, or use request()/arequest() there.

Async tests

mock_engine has no async with form, but the middleware it installs is process-wide, so async requests made inside the with block are intercepted too:

import asyncio

import xbbg
from xbbg.testing import create_mock_response, mock_engine


def test_async_requests():
    responses = [
        create_mock_response(
            service="//blp/refdata",
            operation="ReferenceDataRequest",
            data={"AAPL US Equity": {"PX_LAST": 101.23}},
        ),
        create_mock_response(
            service="//blp/refdata",
            operation="HistoricalDataRequest",
            data={"AAPL US Equity": [{"date": "2024-01-02", "PX_LAST": 185.64}]},
        ),
    ]

    async def run():
        last = await xbbg.arequest(
            "//blp/refdata",
            "ReferenceDataRequest",
            securities=["AAPL US Equity"],
            fields=["PX_LAST"],
            backend="native",
        )
        hist = await xbbg.arequest(
            "//blp/refdata",
            "HistoricalDataRequest",
            securities=["AAPL US Equity"],
            fields=["PX_LAST"],
            start_date="2024-01-02",
            end_date="2024-01-02",
            backend="native",
        )
        return last, hist

    with mock_engine(responses):
        last, hist = asyncio.run(run())

    assert last.to_pylist()[0]["value"] == "101.23"
    assert hist.to_pylist() == [
        {"ticker": "AAPL US Equity", "date": "2024-01-02", "field": "PX_LAST", "value": "185.64"},
    ]

The same works inside async def tests run by an async pytest plugin. Because matching ignores securities, concurrent requests to the same operation (for example through asyncio.gather) take queued responses in the order they reach the middleware.

Strict matching and your own middleware

import pytest

import xbbg
from xbbg.testing import create_mock_response, mock_engine


def test_each_response_is_served_once():
    response = create_mock_response(
        service="//blp/refdata",
        operation="ReferenceDataRequest",
        data={"IBM US Equity": {"PX_LAST": 123.45}},
    )

    with mock_engine([response]):
        xbbg.request("//blp/refdata", "ReferenceDataRequest", securities=["IBM US Equity"], fields=["PX_LAST"])
        with pytest.raises(LookupError, match="No mock response matched"):
            xbbg.request("//blp/refdata", "ReferenceDataRequest", securities=["IBM US Equity"], fields=["PX_LAST"])


def test_audit_middleware_sees_mocked_requests():
    seen = []

    async def audit(context, call_next):
        result = await call_next(context)
        seen.append((context.params.operation, context.metadata.get("mocked")))
        return result

    response = create_mock_response(
        service="//blp/refdata",
        operation="ReferenceDataRequest",
        data={"AAPL US Equity": {"PX_LAST": 101.23}},
    )

    xbbg.add_middleware(audit)
    try:
        with mock_engine([response]):
            xbbg.request("//blp/refdata", "ReferenceDataRequest", securities=["AAPL US Equity"], fields=["PX_LAST"])
    finally:
        xbbg.remove_middleware(audit)

    assert seen == [("ReferenceDataRequest", True)]

mock_engine appends its middleware last, so middleware registered before it wraps the mock and sees mocked requests, as in the second test. See Request middleware.

What is and is not simulated

Simulated	Not simulated
The request path through request()/arequest() and the typed helpers, including parameter validation and your middleware	Bloomberg sessions and authentication; the schema lookup that typed helpers make before middleware when you pass overrides or extra request keywords (see above)
Backend conversion of the canned table (backend=, or the configured default)	Rust-side shaping: format=, typed value columns, pivots, time zones; you get the canned table back as built
request_id, params, and environment on the middleware context	Result metadata: eid_data, security_errors, delayed, bloomberg_request_ids; on_delayed checks and warnings
	Bloomberg errors such as BlpRequestError or BlpTimeoutError
	Subscriptions (subscribe, stream, auction and iNAV streams), bta, delay_status, identity checks, and xbbg.ext native recipes, which call the engine directly instead of going through middleware

To test error handling, raise the exception from a middleware of your own; it runs without mock_engine and without Bloomberg:

import pytest

import xbbg


def test_request_failure_is_reported():
    async def reject(context, call_next):
        raise xbbg.BlpRequestError(
            "Request failed on //blp/refdata::ReferenceDataRequest",
            service=context.params.service,
            operation=context.params.operation,
            request_id=context.request_id,
        )

    xbbg.add_middleware(reject)
    try:
        with pytest.raises(xbbg.BlpRequestError) as excinfo:
            xbbg.request("//blp/refdata", "ReferenceDataRequest", securities=["AAPL US Equity"], fields=["PX_LAST"])
    finally:
        xbbg.remove_middleware(reject)

    assert excinfo.value.operation == "ReferenceDataRequest"
    assert excinfo.value.bloomberg_request_ids == []

Bloomberg SDK test utilities

create_mock_event, deserialize_service, get_admin_message_definition, and append_message_dict are thin wrappers over blpapi.test, the test utilities in Bloomberg's blpapi Python package, and raise ImportError without it. They build real SDK Event objects for code that consumes blpapi directly:

import blpapi

from xbbg.testing import append_message_dict, create_mock_event, get_admin_message_definition


def test_admin_event():
    event = create_mock_event(blpapi.Event.SESSION_STATUS)
    definition = get_admin_message_definition("SessionStarted")
    append_message_dict(event, definition, {})

    assert [str(message.messageType()) for message in event] == ["SessionStarted"]

When create_mock_response receives service_xml (or a blpapi.Service object as service), it also builds an SDK Event from the same data and stores it on MockResponse.event, using event_type (default blpapi.Event.RESPONSE), correlation_ids, and request_id as message properties; data must then also be a valid payload for the operation's response schema. mock_engine serves only MockResponse.table and never reads the event.

Related pages

Request middleware

The chain mock_engine plugs into.

Errors and warnings

Exception classes and attributes to assert on.

Async

Async request APIs and event-loop rules.

API surface

Every public Python name.