[ble][tcat] add support for BLE link disconnection in TCAT simulation (#13240)

This adds a CLI command 'simulation_ble_disconnect' in the TCAT Commissioner, to simulate
a sudden BLE link disconnection while the TLS session is still active. Also no Disconnect
TLV is sent in this case. This feature is useful for simulation testing of TCAT and for
CI testing.
This commit is contained in:
Esko Dijk
2026-07-16 10:39:15 -07:00
committed by Jonathan Hui
parent 95fc632715
commit 48346e3924
4 changed files with 37 additions and 5 deletions
+7 -2
View File
@@ -256,8 +256,13 @@ void platformBleProcess(otInstance *aInstance, const fd_set *aReadFdSet, const f
}
else if (rval == 0)
{
// socket is closed, which should not happen
assert(false);
// A zero-length datagram: simulates Commissioner suddenly disconnected the BLE link.
if (sIsConnected)
{
sIsConnected = false;
otLogDebgPlat("BLE client link disconnected");
otPlatBleGapOnDisconnected(aInstance, 0);
}
}
else if (errno != EINTR && errno != EAGAIN)
{
+7 -2
View File
@@ -60,6 +60,11 @@ class UdpStream:
else:
return b''
async def simulation_ble_disconnect(self):
# Simulate a BLE link break (e.g. peer out of range) by sending a zero-length UDP
# datagram. Unlike `disconnect`, this does not send a Disconnect TLV and does not
# perform a clean TLS shutdown.
self.socket.sendto(b'', self.address)
async def disconnect(self):
if self.socket is not None:
self.socket.close()
self.socket.close()
@@ -244,6 +244,27 @@ class DisconnectCommand(Command):
return CommandResultNone()
class SimulationBleDisconnectCommand(Command):
def get_help_string(self) -> str:
return 'Simulate a BLE link break to a simulated TCAT device (no Disconnect TLV, no TLS shutdown).'
async def execute_default(self, args, context) -> CommandResult:
ble_stream = context['ble_stream']
if not isinstance(ble_stream, UdpStream):
return CommandResultError('only available for a simulation connection (use \'simulation <id>\' first).')
print('Disconnecting simulated BLE link...')
# Signal the abrupt link break to the device, then drop local stream state without a TLS
# shutdown (no close-notify), mirroring a real abrupt disconnect on the client side too.
await ble_stream.simulation_ble_disconnect()
await ble_stream.disconnect()
context['ble_stream'] = None
context['ble_sstream'] = None
print('Done')
return CommandResultNone()
class ExtractDatasetCommand(BleCommand):
def get_log_string(self) -> str:
+2 -1
View File
@@ -38,7 +38,7 @@ from cli.base_commands import (DisconnectCommand, HelpCommand, HelloCommand, Com
GetRandomNumberChallenge, ThreadStateCommand, ScanCommand, PresentHash,
DiagnosticTlvsCommand, GetApplicationLayersCommand, SendVendorData,
SendApplicationData1, SendApplicationData2, SendApplicationData3, SendApplicationData4,
SimulationCommand, connect_helper, disconnect_helper)
SimulationCommand, SimulationBleDisconnectCommand, connect_helper, disconnect_helper)
from .command import CommandResultNone, CommandResult
from .tlv_commands import TlvCommand
from cli.dataset_commands import (DatasetCommand)
@@ -72,6 +72,7 @@ class CLI:
'thread': ThreadStateCommand(),
'scan': ScanCommand(),
'simulation': SimulationCommand(),
'simulation_ble_disconnect': SimulationBleDisconnectCommand(),
'random_challenge': GetRandomNumberChallenge(),
'present_hash': PresentHash(),
'tlv': TlvCommand(),