[tcat] implementation of tcat disconnect command (#10859)

Implementation of tcat disconnect command in python client.

src/core/radio/ble_secure.cpp

@@ -444,6 +444,7 @@ void BleSecure::HandleTlsReceive(uint8_t *aBuf, uint16_t aLength)
 
                 if (error == kErrorAbort)
                 {
+                    LogInfo("Disconnecting TCAT client.");
                     // kErrorAbort indicates that a Disconnect command TLV has been received.
                     Disconnect();
                     // BleSecure is not stopped here, it must remain active in advertising state and

tools/tcat_ble_client/bbtc.py

@@ -97,7 +97,7 @@ async def main():
     ds = ThreadDataset()
     cli = CLI(ds, args, ble_sstream)
     loop = asyncio.get_running_loop()
-    print('Enter \'help\' to see available commands' ' or \'exit\' to exit the application.')
+    print('Enter \'help\' to see available commands or \'exit\' to exit the application.')
     while True:
         user_input = await loop.run_in_executor(None, lambda: input('> '))
         if user_input.lower() == 'exit':

tools/tcat_ble_client/cli/base_commands.py

@@ -73,6 +73,9 @@ class BleCommand(Command):
         pass
 
     async def execute_default(self, args, context):
+        if 'ble_sstream' not in context or context['ble_sstream'] is None:
+            print("TCAT Device not connected.")
+            return CommandResultNone()
         bless: BleStreamSecure = context['ble_sstream']
 
         print(self.get_log_string())
@@ -85,6 +88,7 @@ class BleCommand(Command):
             return CommandResultTLV(tlv_response)
         except DataNotPrepared as err:
             print('Command failed', err)
+        return CommandResultNone()
 
 
 class HelloCommand(BleCommand):
@@ -298,7 +302,7 @@ class PingCommand(Command):
         response = await bless.send_with_resp(data)
         elapsed_time = 1e3 * (time() - elapsed_time)
         if not response:
-            return
+            return CommandResultNone()
 
         tlv_response = TLV.from_bytes(response)
         if tlv_response.value != to_send:
@@ -352,7 +356,8 @@ class ScanCommand(Command):
         return 'Perform scan for TCAT devices.'
 
     async def execute_default(self, args, context):
-        if not (context['ble_sstream'] is None):
+        if 'ble_sstream' in context and context['ble_sstream'] is not None:
+            context['ble_sstream'].close()
             del context['ble_sstream']
 
         tcat_devices = await ble_scanner.scan_tcat_devices()
@@ -379,3 +384,17 @@ class ScanCommand(Command):
         else:
             print('Secure channel not established.')
             await ble_stream.disconnect()
+        return CommandResultNone()
+
+
+class DisconnectCommand(Command):
+
+    def get_help_string(self) -> str:
+        return 'Disconnect client from TCAT device'
+
+    async def execute_default(self, args, context):
+        if 'ble_sstream' not in context or context['ble_sstream'] is None:
+            print("TCAT Device not connected.")
+            return CommandResultNone()
+        await context['ble_sstream'].close()
+        return CommandResultNone()

tools/tcat_ble_client/cli/cli.py

@@ -29,9 +29,10 @@ import readline
 import shlex
 from argparse import ArgumentParser
 from ble.ble_stream_secure import BleStreamSecure
-from cli.base_commands import (HelpCommand, HelloCommand, CommissionCommand, DecommissionCommand, GetDeviceIdCommand,
-                               GetPskdHash, GetExtPanIDCommand, GetNetworkNameCommand, GetProvisioningUrlCommand,
-                               PingCommand, GetRandomNumberChallenge, ThreadStateCommand, ScanCommand, PresentHash)
+from cli.base_commands import (DisconnectCommand, HelpCommand, HelloCommand, CommissionCommand, DecommissionCommand,
+                               GetDeviceIdCommand, GetPskdHash, GetExtPanIDCommand, GetNetworkNameCommand,
+                               GetProvisioningUrlCommand, PingCommand, GetRandomNumberChallenge, ThreadStateCommand,
+                               ScanCommand, PresentHash)
 from cli.dataset_commands import (DatasetCommand)
 from dataset.dataset import ThreadDataset
 from typing import Optional
@@ -48,6 +49,7 @@ class CLI:
             'hello': HelloCommand(),
             'commission': CommissionCommand(),
             'decommission': DecommissionCommand(),
+            'disconnect': DisconnectCommand(),
             'device_id': GetDeviceIdCommand(),
             'ext_panid': GetExtPanIDCommand(),
             'provisioning_url': GetProvisioningUrlCommand(),

tools/tcat_ble_client/dataset/dataset_entries.py

@@ -306,7 +306,7 @@ class Pskc(DatasetEntry):
             args[0] = args[0][2:]
         pskc = args[0]
         if (len(pskc) > self.maxlen * 2):
-            raise ValueError('Invalid length of Pskc. Can be max ' f'{self.length * 2} hex characters.')
+            raise ValueError(f'Invalid length of PSKc. Can be max {self.length * 2} hex characters.')
         self.data = pskc
 
     def set_from_tlv(self, tlv: TLV):