wip command_interface tests

Author: maxd-nordic

src/nrfcredstore/command_interface.py

@@ -27,32 +27,32 @@ def __init__(self, comms: Comms):
         """
         self.comms = comms
 
-    def write_raw(self, command):
+    def write_raw(self, command: str):
         """Write a raw line directly to the serial interface."""
         self.comms.write_line(command)
 
     @abstractmethod
-    def write_credential(self, sectag, cred_type, cred_text):
+    def write_credential(self, sectag: int, cred_type: int, cred_text: str):
         """Write a credential string to the command interface"""
         return
 
     @abstractmethod
-    def delete_credential(self, sectag, cred_type):
+    def delete_credential(self, sectag: int, cred_type: int):
         """Delete a credential using command interface"""
         return
 
     @abstractmethod
-    def check_credential_exists(self, sectag, cred_type, get_hash=True):
+    def check_credential_exists(self, sectag: int, cred_type: int, get_hash=True):
         """Verify that a credential is installed. If check_hash is true, retrieve the SHA hash."""
         return
 
     @abstractmethod
-    def calculate_expected_hash(self, cred_text):
+    def calculate_expected_hash(self, cred_text: str):
         """Returns the expected digest/hash for a given credential as a string"""
         return
 
     @abstractmethod
-    def get_csr(self, sectag = 0, cn = ""):
+    def get_csr(self, sectag: int, attributes: str):
         """Generate a private/public keypair and a corresponding Certificate Signing Request.
 
         Returns:
@@ -78,7 +78,7 @@ def get_mfw_version(self):
 class ATCommandInterface(CredentialCommandInterface):
     shell = False
 
-    def _parse_sha(self, cmng_result_str):
+    def _parse_sha(self, cmng_result_str: str):
         # Example AT%CMNG response:
         #   %CMNG: 123,0,"2C43952EE9E000FF2ACC4E2ED0897C0A72AD5FA72C3D934E81741CBD54F05BD1"
         # The first item in " is the SHA.
@@ -88,7 +88,7 @@ def _parse_sha(self, cmng_result_str):
             logger.error(f'Could not parse credential hash: {cmng_result_str}')
             return None
 
-    def set_shell_mode(self, shell):
+    def set_shell_mode(self, shell: bool):
         self.shell = shell
 
     def detect_shell_mode(self):
@@ -107,7 +107,7 @@ def enable_error_codes(self):
         if not self.at_command('AT+CMEE=1', wait_for_result=True):
             logger.error("Failed to enable error codes.")
 
-    def at_command(self, at_command, wait_for_result=False, suppress_errors=False):
+    def at_command(self, at_command: str, wait_for_result=False, suppress_errors=False):
         """Write an AT command to the command interface. Optionally wait for OK"""
 
         if self.shell:
@@ -124,17 +124,13 @@ def at_command(self, at_command, wait_for_result=False, suppress_errors=False):
         else:
             return True
 
-    def write_credential(self, sectag, cred_type, cred_text):
-        result = self.at_command(f'AT%CMNG=0,{sectag},{cred_type},"{cred_text}"',
-                                 wait_for_result=True)
-        time.sleep(1)
-        return result
+    def write_credential(self, sectag: int, cred_type: int, cred_text: str):
+        return self.at_command(f'AT%CMNG=0,{sectag},{cred_type},"{cred_text}"', wait_for_result=True)
 
-    def delete_credential(self, sectag, cred_type):
-        # No output is expected beyond OK/ERROR in this case
+    def delete_credential(self, sectag: int, cred_type: int):
         return self.at_command(f'AT%CMNG=3,{sectag},{cred_type}', wait_for_result=True)
 
-    def check_credential_exists(self, sectag, cred_type, get_hash=True):
+    def check_credential_exists(self, sectag: int, cred_type: int, get_hash=True):
         self.at_command(f'AT%CMNG=1,{sectag},{cred_type}')
         retval, res = self.comms.expect_response("OK", "ERROR", "%CMNG")
         if retval and res:
@@ -145,7 +141,7 @@ def check_credential_exists(self, sectag, cred_type, get_hash=True):
 
         return False, None
 
-    def calculate_expected_hash(self, cred_text):
+    def calculate_expected_hash(self, cred_text: str):
         # AT Command host returns hex of SHA256 hash of credential plaintext
         return hashlib.sha256(cred_text.encode('utf-8')).hexdigest().upper()
 
@@ -274,7 +270,7 @@ def calculate_expected_hash(self, cred_text):
         hash = hashlib.sha256(cred_text.encode('utf-8') + b'\x00')
         return base64.b64encode(hash.digest()).decode()
 
-    def get_csr(self, sectag = 0, cn = ""):
+    def get_csr(self, sectag=0, attributes=""):
         raise RuntimeError("The TLS Credentials Shell does not support CSR generation")
 
     def go_offline(self):

tests/test_command_interface.py

@@ -3,6 +3,7 @@
 import pytest
 
 from nrfcredstore.command_interface import CredentialCommandInterface, ATCommandInterface, TLSCredShellInterface
+from nrfcredstore.credstore import CredType
 
 @pytest.fixture
 def comms():
@@ -13,11 +14,141 @@ def comms():
 @pytest.fixture
 def at_command_interface(comms):
     """Mock command interface"""
-    command_interface = ATCommandInterface(comms)
-    return command_interface
+    interface = ATCommandInterface(comms)
+    return interface
 
-def test_write_raw(at_command_interface):
+@pytest.fixture
+def tls_cred_shell_interface(comms):
+    """Mock TLSCredShellInterface"""
+    interface = TLSCredShellInterface(comms)
+    return interface
+
+def test_write_raw_at(at_command_interface):
     """Test writing raw AT commands"""
     at_command_interface.comms.write_line = Mock()
     at_command_interface.write_raw('AT+TEST=1')
     at_command_interface.comms.write_line.assert_called_once_with('AT+TEST=1')
+
+def test_write_raw_tls(tls_cred_shell_interface):
+    """Test writing raw TLS commands"""
+    tls_cred_shell_interface.comms.write_line = Mock()
+    tls_cred_shell_interface.write_raw('AT+TEST=1')
+    tls_cred_shell_interface.comms.write_line.assert_called_once_with('AT+TEST=1')
+
+def test_write_credential_at(at_command_interface):
+    """Test writing a credential using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, "")
+    at_command_interface.write_credential(sectag=42, cred_type=CredType.CLIENT_CERT.value, cred_text='test_value')
+    at_command_interface.comms.write_line.assert_called_once_with('AT%CMNG=0,42,1,"test_value"')
+
+def test_delete_credential_at(at_command_interface):
+    """Test deleting a credential using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, "")
+    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_check_credential_exists_at(at_command_interface):
+    """Test checking if a credential exists using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, '%CMNG: 42,1,"8CEA57609B0F95C0D0F80383A7A21ECD1C6E102FDCC3CDCEB1948B0EA828601D"')
+    exists, sha = at_command_interface.check_credential_exists(sectag=42, cred_type=CredType.CLIENT_CERT.value)
+    assert exists is True
+    assert sha == '8CEA57609B0F95C0D0F80383A7A21ECD1C6E102FDCC3CDCEB1948B0EA828601D'
+
+def test_get_csr_at(at_command_interface):
+    """Test getting a CSR using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, '%KEYGEN: "foo.bar"')
+    csr = at_command_interface.get_csr(sectag=42, attributes='O=Test,CN=Device')
+    assert csr == 'foo.bar'
+    at_command_interface.comms.write_line.assert_called_once_with('AT%KEYGEN=42,2,0,"O=Test,CN=Device"')
+
+def test_get_csr_no_attributes(at_command_interface):
+    """Test getting a CSR without attributes using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, '%KEYGEN: "foo.bar"')
+    csr = at_command_interface.get_csr(sectag=42)
+    assert csr == 'foo.bar'
+    at_command_interface.comms.write_line.assert_called_once_with('AT%KEYGEN=42,2,0')
+
+def test_get_imei(at_command_interface):
+    """Test getting IMEI using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, '123456789012345')
+    imei = at_command_interface.get_imei()
+    assert imei == '123456789012345'
+    at_command_interface.comms.write_line.assert_called_once_with('AT+CGSN')
+
+def test_go_offline(at_command_interface):
+    """Test going offline using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, '')
+    at_command_interface.go_offline()
+    at_command_interface.comms.write_line.assert_called_once_with('AT+CFUN=4')
+
+def test_get_model_id(at_command_interface):
+    """Test getting model ID using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, 'nRF9151-LACA')
+    model_id = at_command_interface.get_model_id()
+    assert model_id == 'nRF9151-LACA'
+    at_command_interface.comms.write_line.assert_called_once_with('AT+CGMM')
+
+def test_get_mfw_version(at_command_interface):
+    """Test getting MFW version using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, 'mfw_nrf91x1_2.0.2')
+    mfw_version = at_command_interface.get_mfw_version()
+    assert mfw_version == 'mfw_nrf91x1_2.0.2'
+    at_command_interface.comms.write_line.assert_called_once_with('AT+CGMR')
+
+def test_get_attestation_token(at_command_interface):
+    """Test getting attestation token using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, '%ATTESTTOKEN: "foo.bar"')
+    attestation_token = at_command_interface.get_attestation_token()
+    assert attestation_token == 'foo.bar'
+    at_command_interface.comms.write_line.assert_called_once_with('AT%ATTESTTOKEN')
+
+def test_enable_error_codes(at_command_interface):
+    """Test enabling error codes using ATCommandInterface"""
+    at_command_interface.comms.write_line = Mock()
+    at_command_interface.comms.expect_response.return_value = (True, '')
+    at_command_interface.enable_error_codes()
+    at_command_interface.comms.write_line.assert_called_once_with('AT+CMEE=1')
+
+class MockCommsAT:
+    """Mock comms for ATCommandInterface"""
+    def write_line(self, line):
+        self.last_written = line
+
+    def expect_response(self, ok_str=None, error_str=None, store_str=None, timeout=15, suppress_errors=False):
+        if self.last_written == 'AT+CGSN':
+            return True, '123456789012345'
+        else:
+            return False, ''
+
+def test_detect_shell_mode_at(at_command_interface):
+    """Test detecting shell mode using ATCommandInterface (AT Host)"""
+    at_command_interface.comms = MockCommsAT()
+    at_command_interface.detect_shell_mode()
+    assert at_command_interface.shell == False
+
+class MockCommsATShell:
+    """Mock comms for TLSCredShellInterface"""
+    def write_line(self, line):
+        self.last_written = line
+
+    def expect_response(self, ok_str=None, error_str=None, store_str=None, timeout=15, suppress_errors=False):
+        if self.last_written == 'at AT+CGSN':
+            return True, '123456789012345'
+        else:
+            return False, ''
+
+def test_detect_shell_mode_shell(at_command_interface):
+    """Test detecting shell mode using ATCommandInterface (AT Shell)"""
+    at_command_interface.comms = MockCommsATShell()
+    at_command_interface.detect_shell_mode()
+    assert at_command_interface.shell == True