Unit and integration tests for the Kalshi API client library.
# Run all tests
pytest tests/
# Run with verbose output
pytest tests/ -v
# Run specific test file
pytest tests/test_exchange.py
# Run specific test class
pytest tests/test_exchange.py::TestExchangeStatus
# Run with coverage
pytest tests/ --cov=pykalshimock_response: Factory for creating mock HTTP responses with JSON data and status codesclient: Pre-configuredKalshiClientwith mocked auth and HTTP session (no real API calls)
| File | Description |
|---|---|
test_client.py |
Core HTTP client: auth headers, error handling, retries |
test_models.py |
Pydantic model validation for all data types |
test_portfolio.py |
Portfolio operations: positions, fills, orders |
test_workflow.py |
End-to-end workflows: balance, order placement, orderbook |
test_feed.py |
WebSocket feed: subscriptions, handlers, message dispatch |
test_exchange.py |
Exchange status, schedule |
test_api_keys.py |
API key management: list, create, generate, delete, rate limits |
test_series.py |
Series, public trades, batch candlesticks |
test_markets.py |
Markets and events: fetch, list, candlesticks, filters |
test_integration.py |
Live API tests (skipped without credentials) |
- Auth header generation with RSA-PSS signatures
- HTTP response handling (success, errors)
- Custom exceptions:
AuthenticationError,ResourceNotFoundError,InsufficientFundsError - Error message and code extraction
- All Pydantic models validate correctly
- Enum parsing (Action, Side, OrderStatus, MarketStatus)
- Optional fields handled
- Extra fields ignored (forward compatibility)
- Type coercion and validation errors
get_positions()with filters (ticker, event_ticker, count_filter)get_fills()with filters (ticker, order_id, min_ts, max_ts)get_order()by ID- 404 handling for missing orders
portfolio.get_balance()methodportfolio.place_order()with Market objectsmarket.get_orderbook()returns typed response
- URL selection (demo vs production)
- Handler registration (direct and decorator)
- Subscription management
- Message dispatch and parsing for all channel types:
ticker,trade,fillorderbook_snapshot,orderbook_delta
- Error handling (malformed JSON, handler exceptions)
- Auth header generation
client.exchange.get_status()- operational statusclient.exchange.is_trading()- quick boolean checkclient.exchange.get_schedule()- trading hoursclient.exchange.get_user_data_timestamp()- data validation timestamp- Cached property behavior
client.api_keys.list()- list all keysclient.api_keys.create()- create with public keyclient.api_keys.generate()- generate key pairclient.api_keys.delete()- delete keyclient.api_keys.get_limits()- rate limit info- 404 handling for non-existent keys
client.get_series()- fetch single seriesclient.get_all_series()- list with category filterseries.get_markets()- markets in a seriesclient.get_trades()- public trade history with filtersclient.get_candlesticks_batch()- batch candlestick retrieval- Pagination with
fetch_all=True
client.get_market()- fetch single marketclient.get_markets()- list with filters (series, event, status)market.get_candlesticks()- historical OHLC data- Series ticker resolution from event
client.get_event()- fetch single eventclient.get_events()- list with filters- Market equality and hashing
- Attribute delegation to underlying model
def test_example(client, mock_response):
client._session.request.return_value = mock_response({
"data": {"key": "value"}
})
result = client.get("/endpoint")
assert result["data"]["key"] == "value"def test_not_found(client, mock_response):
client._session.request.return_value = mock_response(
{"message": "Not found"}, status_code=404
)
with pytest.raises(ResourceNotFoundError):
client.get("/missing")def test_pagination(client, mock_response):
client._session.request.side_effect = [
mock_response({"items": [1], "cursor": "page2"}),
mock_response({"items": [2], "cursor": ""}),
]
result = client.get_items(fetch_all=True)
assert len(result) == 2
assert client._session.request.call_count == 2def test_filters(client, mock_response):
client._session.request.return_value = mock_response({"items": []})
client.get_items(status="active", limit=50)
call_url = client._session.request.call_args.args[1]
assert "status=active" in call_url
assert "limit=50" in call_urltest_integration.py contains tests that hit the real Kalshi API. These are skipped by default unless valid credentials are configured:
# Set environment variables
export KALSHI_API_KEY_ID="your_key"
export KALSHI_PRIVATE_KEY_PATH="/path/to/key.pem"
# Run integration tests
pytest tests/test_integration.py -vIntegration tests verify:
- Balance retrieval
- Market listing and fetching
- Orderbook data
- Position queries
- Fill history
- Order queries