API Reference

Test Context

Test Context for BLE tests.

Provides environment for test execution.

class test_a_ble.test_context.NotificationResult(*values)

Bases: Enum

Enum for notification evaluation results.

FAIL = 'fail'

IGNORE = 'ignore'

MATCH = 'match'

class test_a_ble.test_context.NotificationSubscription(characteristic_uuid: str, initial_waiter: NotificationWaiter = None)

Bases: object

A helper class to manage notification subscriptions and waiters.

clear_waiter()

Clear the waiter for the subscription.

on_notification(data)

Handle a notification.

set_waiter(waiter: NotificationWaiter, process_collected_notifications: bool = True)

Set the waiter for the subscription.

class test_a_ble.test_context.NotificationWaiter(characteristic_uuid: str, expected_value: bytes | Callable[[bytes], bool | NotificationResult | Tuple[NotificationResult, str]] | None = None)

Bases: object

Helper class to wait for notifications.

check_notification(data: bytes) → Tuple[bool, str | None]

Check if a notification matches our criteria.

Parameters:

data – The notification data to check

Returns:

Tuple of (is_match, failure_reason) - is_match: True if the notification matches expected criteria - failure_reason: If the notification indicates a failure condition, the reason

on_notification(data) → bool

Handle a notification.

Parameters:

data – The notification data

Returns:

True if the notification matches the expected value, False otherwise

class test_a_ble.test_context.TestContext(ble_manager: BLEManager)

Bases: object

Context for test execution.

Provides access to the BLE device and helper methods for test operations.

async cleanup_tasks()

Clean up any remaining async tasks created during testing.

This should be called before program exit.

async create_notification_waiter(characteristic_uuid: str, expected_value: bytes | Callable[[bytes], bool | NotificationResult | Tuple[NotificationResult, str]] | None = None, process_collected_notifications: bool = True) → NotificationWaiter

Create a notification waiter for a characteristic.

Parameters:

• characteristic_uuid – UUID of characteristic to wait for notification

• expected_value – If provided, validates the notification value. Can be: - bytes: exact value to match - callable: function that takes the notification data and returns a NotificationResult

Returns:

NotificationWaiter instance

critical(message: str) → None

Log a critical message within the current test context.

debug(message: str) → None

Log a debug message within the current test context.

end_test(status: TestStatus | str, message: str = '') → Dict[str, Any]

End the current test and record results.

Parameters:

• status – Test status (TestStatus enum or string value)

• message – Optional message about test result

Returns:

Test result details

error(message: str) → None

Log an error message within the current test context.

get_test_summary() → Dict[str, Any]

Generate a summary of all test results.

Returns:

Dictionary with test summary statistics

handle_notification_waiter_result(waiter: NotificationWaiter, timeout: float) → Dict[str, Any]

Handle the result of a notification waiter.

Parameters:

• waiter – The notification waiter to check

• timeout – The timeout value that was used (for error messages)

Returns:

‘value’: The notification value that matched the expected value (bytes) ‘success’: True if notification received and matched expected ‘received_notifications’: List of all notifications received (list of bytes)

Return type:

Dictionary with notification details if successful

Raises:

• TestFailure – If a notification indicates a test failure

• TimeoutError – If no notification was received within the timeout period

info(message: str) → None

Log an info message within the current test context.

log(message: str, level: str = 'info') → None

Log a message within the current test context.

Parameters:

• message – Message to log

• level – Log level (debug, info, warning, error, critical)

print(message: str) → None

Print a message directly to the console for user-facing output.

Use this for information that should always be visible to the user, regardless of log level settings.

Parameters:

message – The message to display to the user

print_formatted_box(title: str, messages: List[str]) → None

Print a formatted box with consistent alignment.

Parameters:

• title – The title to display at the top of the box

• messages – List of message lines to display in the box

prompt_user(message: str) → str

Display a prompt to the user and wait for input.

Parameters:

message – The message to display to the user

Returns:

User’s input response

start_test(test_name: str) → None

Start a new test and record the start time.

Parameters:

test_name – Name of the test being started

async subscribe_to_characteristic(characteristic_uuid: str, waiter: NotificationWaiter | None = None, process_collected_notifications: bool = True)

Subscribe to a characteristic and create a waiter if provided.

Parameters:

• characteristic_uuid – UUID of characteristic to subscribe to

• waiter – Optional NotificationWaiter instance to use

• process_collected_notifications – If True, process collected notifications

async unsubscribe_all() → None

Unsubscribe from all active notification subscriptions.

Call this at the end of a test to clean up resources.

async wait_for_notification(characteristic_uuid: str, timeout: float = 10.0, expected_value: bytes | Callable[[bytes], bool | NotificationResult | Tuple[NotificationResult, str]] | None = None, process_collected_notifications: bool = True) → Dict[str, Any]

Wait for a notification from a characteristic without user interaction.

Parameters:

• characteristic_uuid – UUID of characteristic to wait for notification

• timeout – Maximum time to wait in seconds

• expected_value – If provided, validates the notification value. Can be: - bytes: exact value to match - callable: function that takes the notification data and returns a NotificationResult

Returns:

‘value’: The notification value ‘success’: True if notification received and matched expected ‘received_notifications’: List of all notifications received

Return type:

Dictionary with notification details

Raises:

• TimeoutError – If no notification is received within the timeout

• TestFailure – If a notification is received but doesn’t match expected criteria

async wait_for_notification_interactive(characteristic_uuid: str, timeout: float = 10.0, expected_value: bytes | Callable[[bytes], bool | NotificationResult | Tuple[NotificationResult, str]] | None = None) → Dict[str, Any]

Wait for a notification from a characteristic with user interaction support.

This method will display a prompt to the user and wait for a notification. The user can type ‘s’ or ‘skip’ to skip the test, or ‘f’ or ‘fail’ to fail it. If the user chooses to skip or fail, the appropriate TestSkip or TestFailure exception will be raised automatically.

Parameters:

• characteristic_uuid – UUID of characteristic to wait for notification

• timeout – Maximum time to wait in seconds

• expected_value – If provided, validates the notification value. Can be: - bytes: exact value to match - callable: function that takes the notification data and returns a NotificationResult

Returns:

‘value’: The notification value ‘success’: True if notification received and matched expected ‘received_notifications’: List of all notifications received

Return type:

Dictionary with notification details

Raises:

• TestSkip – If the user chooses to skip the test

• TestFailure – If the user chooses to fail the test

• TimeoutError – If no notification is received within the timeout

warning(message: str) → None

Log a warning message within the current test context.

exception test_a_ble.test_context.TestException(message='')

Bases: Exception

Base class for test exceptions.

status = 'error'

exception test_a_ble.test_context.TestFailure(message='')

Bases: TestException

Exception raised when a test fails.

status = 'fail'

exception test_a_ble.test_context.TestSkip(message='')

Bases: TestException

Exception raised when a test is skipped.

status = 'skip'

class test_a_ble.test_context.TestStatus(*values)

Bases: Enum

Enum for test execution status.

ERROR = 'error'

FAIL = 'fail'

PASS = 'pass'

RUNNING = 'running'

SKIP = 'skip'

test_a_ble.test_context.ble_test(description=None)

Decorate a BLE test function.

Parameters:

description – Description of the test (optional, will use function name if not provided)

Returns:

Decorated function

test_a_ble.test_context.ble_test_class(description=None)

Decorate a BLE test class.

Parameters:

description – Description of the test class (optional, will use class name if not provided)

Returns:

Decorated class

BLE Manager

BLE Manager.

Manages BLE device discovery, connection, and communication.

class test_a_ble.ble_manager.BLEManager

Bases: object

Manages BLE device discovery, connection, and communication.

async connect_to_device(device_or_address: BLEDevice | str, retry_count: int = 3, retry_delay: float = 1.0) → bool

Connect to a BLE device.

Parameters:

• device_or_address – BLEDevice or device address to connect to

• retry_count – Number of connection attempts before failing

• retry_delay – Delay between retries in seconds

Returns:

True if connection successful, False otherwise

async disconnect()

Disconnect from the connected device and clean up resources.

async discover_devices(timeout: float = 5.0, name_filter: str | None = None, address_filter: str | None = None) → List[BLEDevice]

Scan for BLE devices and return filtered results.

Parameters:

• timeout – Scan duration in seconds

• name_filter – Optional filter for device name (substring match)

• address_filter – Optional filter for device address

Returns:

List of discovered BLE devices matching filters

async discover_services(cache: bool = True) → Dict[str, Any]

Discover services and characteristics of the connected device.

Parameters:

cache – Whether to cache results for future use

Returns:

Dictionary of services and their characteristics

get_discovered_device_info() → List[Dict[str, Any]]

Return information about discovered devices in a structured format.

async read_characteristic(characteristic_uuid: str) → bytearray

Read value from a characteristic.

Parameters:

characteristic_uuid – UUID of the characteristic to read

Returns:

Bytes read from the characteristic

classmethod register_expected_services(service_uuids)

Register service UUIDs that should be used when looking for connected devices.

Parameters:

service_uuids – List or set of service UUID strings in standard format

async subscribe_to_characteristic(characteristic_uuid: str, callback: Callable[[bytearray], None]) → None

Subscribe to notifications from a characteristic.

Parameters:

• characteristic_uuid – UUID of the characteristic to subscribe to

• callback – Function to call when notification is received

async unsubscribe_from_characteristic(characteristic_uuid: str) → None

Unsubscribe from notifications from a characteristic.

Parameters:

characteristic_uuid – UUID of the characteristic to unsubscribe from

async write_characteristic(characteristic_uuid: str, data: bytes | bytearray | memoryview, response: bool = True) → None

Write value to a characteristic.

Parameters:

• characteristic_uuid – UUID of the characteristic to write to

• data – Data to write

• response – Whether to wait for response

test_a_ble.ble_manager.retrieveConnectedPeripheralsWithServices(scanner: BleakScanner, services: List[str] | List[UUID]) → list[BLEDevice]

Retrieve connected peripherals with specified services.

Test Runner

Test Runner.

Discovers and executes BLE tests

class test_a_ble.test_runner.TestRunner(ble_manager: BLEManager)

Bases: object

Discovers and runs tests against BLE devices.

discover_tests(test_specifiers: List[str]) → List[Tuple[str, List[Tuple[str, Callable | Tuple[str, Any, Callable]]]]]

Discover test modules with the given specifiers.

Parameters:

test_specifiers – List of test specifiers

Returns:

Dictionary mapping test names to test functions or (class, method) tuples

async run_test(test_name: str, test_item: Callable | Tuple[str, Any, Callable]) → Dict[str, Any]

Run a single test by name.

Parameters:

test_name – Name of the test to run

Returns:

Test result dictionary

async run_tests(tests: List[Tuple[str, Callable | Tuple[str, Any, Callable]]]) → Dict[str, Any]

Run multiple tests in the order they were defined in the source code.

Parameters:

tests – List of tests to run

Returns:

Summary of test results

CLI

Command Line Interface for BLE Testing Framework.

async test_a_ble.cli.connect_to_device(ble_manager: BLEManager, address: str | None = None, name: str | None = None, interactive: bool = False, scan_timeout: float = 10.0) → Tuple[bool, bool]

Connect to a BLE device by address, name, or interactively.

Parameters:

• ble_manager – BLE Manager instance

• address – Optional device address to connect to

• name – Optional device name to connect to

• interactive – Whether to use interactive mode for device selection

• scan_timeout – Scan timeout in seconds

Returns:

Tuple of (connected successfully, user quit)

async test_a_ble.cli.dynamic_device_selection(ble_manager: BLEManager, timeout: float = 10.0) → Tuple[bool, bool]

Interactive device discovery with real-time updates and concurrent user input.

Parameters:

• ble_manager – BLE Manager instance

• timeout – Maximum scan duration in seconds

Returns:

Tuple of (connected successfully, user quit)

test_a_ble.cli.get_console() → Console

Return the global console object for rich output.

test_a_ble.cli.main()

Execute the main function.

test_a_ble.cli.print_test_results(results: Dict[str, Any], verbose=False)

Print formatted test results.

Parameters:

• results – Test results dictionary

• verbose – If True, show logs for all tests (not just failed ones)

async test_a_ble.cli.run_ble_tests(args)

Run BLE tests based on command line arguments.