test_a_ble package

Submodules

test_a_ble.ble_manager module

BLE Manager.

Manages BLE device discovery, connection, and communication.

class test_a_ble.ble_manager.BLEManager[source]

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[source]

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()[source]: 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][source]

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][source]

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]][source]: Return information about discovered devices in a structured format.

async read_characteristic(characteristic_uuid: str) → bytearray[source]

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)[source]

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[source]

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[source]

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[source]

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][source]: Retrieve connected peripherals with specified services.

test_a_ble.cli module

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][source]

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][source]

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[source]: Return the global console object for rich output.

test_a_ble.cli.main()[source]: Execute the main function.

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

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)[source]: Run BLE tests based on command line arguments.

test_a_ble.test_context module

Test Context for BLE tests.

Provides environment for test execution.

class test_a_ble.test_context.NotificationResult(*values)[source]

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)[source]

Bases: object

A helper class to manage notification subscriptions and waiters.

clear_waiter()[source]: Clear the waiter for the subscription.

on_notification(data)[source]: Handle a notification.

set_waiter(waiter: NotificationWaiter, process_collected_notifications: bool = True)[source]: 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)[source]

Bases: object

Helper class to wait for notifications.

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

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[source]

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)[source]

Bases: object

Context for test execution.

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

async cleanup_tasks()[source]

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[source]

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[source]: Log a critical message within the current test context.

debug(message: str) → None[source]: Log a debug message within the current test context.

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

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[source]: Log an error message within the current test context.

get_test_summary() → Dict[str, Any][source]

Generate a summary of all test results.

Returns:: Dictionary with test summary statistics

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

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[source]: Log an info message within the current test context.

log(message: str, level: str = 'info') → None[source]

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[source]

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[source]

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[source]

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[source]

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)[source]

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[source]

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][source]

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][source]

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[source]: Log a warning message within the current test context.

exception test_a_ble.test_context.TestException(message='')[source]

Bases: Exception

Base class for test exceptions.

status = 'error'

exception test_a_ble.test_context.TestFailure(message='')[source]

Bases: TestException

Exception raised when a test fails.

status = 'fail'

exception test_a_ble.test_context.TestSkip(message='')[source]

Bases: TestException

Exception raised when a test is skipped.

status = 'skip'

class test_a_ble.test_context.TestStatus(*values)[source]

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)[source]

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)[source]

Decorate a BLE test class.

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

test_a_ble.test_runner module

Test Runner.

Discovers and executes BLE tests

class test_a_ble.test_runner.TestRunner(ble_manager: BLEManager)[source]

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]]]]][source]

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][source]

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][source]

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

Module contents

BLE IoT Device Testing Framework.

A modular, extensible framework for testing Bluetooth Low Energy IoT devices.

test_a_ble.setup_logging(verbose: bool = False, log_file: str | None = None)[source]

Configure logging for the BLE test framework.

Parameters:

verbose – If True, set log level to DEBUG, otherwise WARNING to hide INFO logs during test
log_file – Optional path to a log file to write logs to