fix TLS credentials IF

Signed-off-by: Maximilian Deubel <maximilian.deubel@nordicsemi.no>

Author: maxd-nordic

src/nrfcredstore/command_interface.py

@@ -235,29 +235,27 @@ def write_credential(self, sectag, cred_type, cred_text):
         for c in range(chunks):
             chunk = encoded[c*TLS_CRED_CHUNK_SIZE:(c+1)*TLS_CRED_CHUNK_SIZE]
             self.write_raw(f"cred buf {chunk}")
-            self.comms.expect_response("Stored")
+            self.comms.expect_response(ok_pattern="Stored")
 
         # Store the buffered credential
         self.write_raw(f"cred add {sectag} {TLS_CRED_TYPES[cred_type]} DEFAULT bint")
-        result, _ = self.comms.expect_response("Added TLS credential")
-        time.sleep(1)
+        result, _ = self.comms.expect_response(ok_pattern="Added TLS credential")
         return result
 
     def delete_credential(self, sectag: int, cred_type: int):
         self.write_raw(f'cred del {sectag} {TLS_CRED_TYPES[cred_type]}')
-        result, _ = self.comms.expect_response("Deleted TLS credential", "There is no TLS credential")
-        time.sleep(2)
+        result, _ = self.comms.expect_response(ok_pattern="Deleted TLS credential", error_pattern="There is no TLS credential")
         return result
 
     def check_credential_exists(self, sectag: int, cred_type: int, get_hash=True):
         self.write_raw(f'cred list {sectag} {TLS_CRED_TYPES[cred_type]}')
 
         # This will capture the list dump for the credential if it exists.
-        result, output = self.comms.expect_response("1 credentials found.",
-                                                    "0 credentials found.",
-                                                    f"{sectag},{TLS_CRED_TYPES[cred_type]}")
+        result, output = self.comms.expect_response(ok_pattern="1 credentials found.",
+                                                    error_pattern="0 credentials found.",
+                                                    store_str=f"{sectag},{TLS_CRED_TYPES[cred_type]}")
 
-        if not output:
+        if not result:
             return False, None
 
         if not get_hash:

src/nrfcredstore/comms.py

@@ -322,7 +322,7 @@ def close(self):
             self.serial_api.close()
             self.serial_api = None
 
-    def expect_response(self, ok_str=None, error_str=None, store_str=None, timeout=15, suppress_errors=False):
+    def expect_response(self, ok_str=None, error_str=None, store_str=None, timeout=15, suppress_errors=False, error_pattern=None, ok_pattern=None):
         '''
         Read lines until either ok_str or error_str is found or timeout (seconds) is reached.
         If store_str is in one of the lines, it will be returned as the output.
@@ -339,8 +339,12 @@ def expect_response(self, ok_str=None, error_str=None, store_str=None, timeout=1
                 line = ansi_escape.sub('', line)
                 if ok_str and ok_str == line:
                     return (True, output)
+                if ok_pattern and ok_pattern in line:
+                    return (True, output)
                 if error_str and error_str == line:
                     return (False, output)
+                if error_pattern and error_pattern in line:
+                    return (False, output)
                 if line.startswith('+CME ERROR'):
                     code = int(line.replace('+CME ERROR: ', ''))
                     if not suppress_errors:

tests/test_command_interface.py

@@ -49,6 +49,50 @@ def test_delete_credential_at(at_command_interface):
     at_command_interface.delete_credential(sectag=42, cred_type=CredType.CLIENT_CERT.value)
     at_command_interface.comms.write_line.assert_called_once_with('AT%CMNG=3,42,1')
 
+def test_delete_credential_tls(tls_cred_shell_interface):
+    """Test deleting a credential that exists in TLS shell"""
+    tls_cred_shell_interface.comms.write_line = Mock()
+    tls_cred_shell_interface.comms.expect_response.return_value = (True, '')
+    tls_cred_shell_interface.delete_credential(sectag=42, cred_type=CredType.ROOT_CA_CERT.value)
+    tls_cred_shell_interface.comms.write_line.assert_called_once_with('cred del 42 CA')
+
+def test_delete_credential_tls_not_found(tls_cred_shell_interface):
+    """Test deleting a credential that does not exist in TLS shell"""
+    tls_cred_shell_interface.comms.write_line = Mock()
+    tls_cred_shell_interface.comms.expect_response.return_value = (False, '')
+    tls_cred_shell_interface.delete_credential(sectag=42, cred_type=CredType.ROOT_CA_CERT.value)
+    tls_cred_shell_interface.comms.write_line.assert_called_once_with('cred del 42 CA')
+
+def test_check_credential_exists_tls(tls_cred_shell_interface):
+    """Test checking a credential that exists in TLS shell"""
+    tls_cred_shell_interface.comms.write_line = Mock()
+    tls_cred_shell_interface.comms.expect_response.return_value = (True, '42,CA,qLqL2kIyuj7FF9aFXQuhmc8kbMOHNYsM451gElqNyVQ=,0\r\n')
+    result, hash = tls_cred_shell_interface.check_credential_exists(sectag=42, cred_type=CredType.ROOT_CA_CERT.value, get_hash=True)
+    tls_cred_shell_interface.comms.write_line.assert_called_once_with('cred list 42 CA')
+    assert hash == 'qLqL2kIyuj7FF9aFXQuhmc8kbMOHNYsM451gElqNyVQ='
+    assert result == True
+
+def test_check_credential_exists_tls_not_found(tls_cred_shell_interface):
+    """Test checking a credential that does not exist in TLS shell"""
+    tls_cred_shell_interface.comms.write_line = Mock()
+    tls_cred_shell_interface.comms.expect_response.return_value = (False, '')
+    result, hash = tls_cred_shell_interface.check_credential_exists(sectag=42, cred_type=CredType.ROOT_CA_CERT.value, get_hash=True)
+    tls_cred_shell_interface.comms.write_line.assert_called_once_with('cred list 42 CA')
+    assert hash == None
+    assert result == False
+
+def test_write_credential_tls(tls_cred_shell_interface):
+    """Test writing a credential using TLSCredShellInterface"""
+    tls_cred_shell_interface.comms.write_line = Mock()
+    tls_cred_shell_interface.comms.expect_response.side_effect = [
+        (True, ""),
+        (True, "")
+    ]
+    result = tls_cred_shell_interface.write_credential(sectag=42, cred_type=CredType.CLIENT_CERT.value, cred_text='test_value')
+    assert result == True
+    assert tls_cred_shell_interface.comms.write_line.call_count >= 3
+
+
 def test_check_credential_exists_at(at_command_interface):
     """Test checking if a credential exists using ATCommandInterface"""
     at_command_interface.comms.write_line = Mock()

tests/test_comms.py

@@ -457,3 +457,22 @@ def test_expect_response_store(mock_serial):
         mock_select.assert_called_once()
         assert result is True
         assert output.strip() == '%ATTESTTOKEN: "foo.bar"'
+
+def test_expect_response_pattern(mock_serial):
+    with patch("nrfcredstore.comms.select_device", return_value=(Mock(), "123456789")) as mock_select:
+        comms = Comms()
+        comms.read_line = Mock(side_effect=[
+            "Stored 48 bytes.",
+            "16842753,CA,qLqL2kIyuj7FF9aFXQuhmc8kbMOHNYsM451gElqNyVQ=,0",
+            "1 credentials found.",
+            "0 credentials found."
+        ])
+        assert comms.expect_response(ok_pattern="Stored") == (True, '')
+        assert comms.expect_response(
+            ok_pattern="1 credentials found.",
+            error_pattern="0 credentials found.",
+            store_str="16842753,CA") == (True, "16842753,CA,qLqL2kIyuj7FF9aFXQuhmc8kbMOHNYsM451gElqNyVQ=,0\r\n")
+        assert comms.expect_response(
+            ok_pattern="1 credentials found.",
+            error_pattern="0 credentials found.",
+            store_str="16842753,CA") == (False, '')