Merge pull request #192 from NebraLtd/kashifpk/pm-180-more-robust-key…

…-provisioning

Support for performing key provisioning on any slot and backup key generation logic.

hm_pyhelper/miner_param.py

@@ -25,8 +25,8 @@
 
 
 @lock_ecc()
-def run_gateway_mfr(sub_command: str) -> None:
-    command = get_gateway_mfr_command(sub_command)
+def run_gateway_mfr(sub_command: str, slot: int = 0) -> dict:
+    command = get_gateway_mfr_command(sub_command, slot=slot)
 
     try:
         run_gateway_mfr_result = subprocess.run(
@@ -101,7 +101,7 @@ def get_gateway_mfr_version() -> Version:
         raise GatewayMFRInvalidVersion(err_str).with_traceback(e.__traceback__)
 
 
-def get_gateway_mfr_command(sub_command: str) -> list:
+def get_gateway_mfr_command(sub_command: str, slot: int = 0) -> list:
     gateway_mfr_path = get_gateway_mfr_path()
     command = [gateway_mfr_path]
 
@@ -132,6 +132,9 @@ def get_gateway_mfr_command(sub_command: str) -> list:
         except (UnknownVariantException, UnknownVariantAttributeException) as e:
             LOGGER.warning(str(e) + ' Omitting --device arg.')
 
+        slot_str = f'slot={slot}'
+        command[-1] = command[-1].replace('slot=0', slot_str)
+
         if ' ' in sub_command:
             command += sub_command.split(' ')
         else:
@@ -164,30 +167,48 @@ def get_gateway_mfr_test_result():
     return run_gateway_mfr("test")
 
 
-def provision_key():
+def provision_key(slot: int, force: bool = False):
     """
     Attempt to provision key.
+
+    :param slot: The ECC key slot to use
+    :param force: If set to True then try `key --generate` if `provision` action fails.
+
+    :return: A 2 element tuple, first one specifying provisioning success (True/False) and
+             second element contains gateway mfr output or error response.
     """
-    test_results = get_gateway_mfr_test_result()
-    if did_gateway_mfr_test_result_include_miner_key_pass(test_results):
-        return True
 
     provisioning_successful = False
+    response = ''
 
     try:
-        gateway_mfr_result = run_gateway_mfr("provision")
+        gateway_mfr_result = run_gateway_mfr("provision", slot=slot)
         LOGGER.info("[ECC Provisioning] %s", gateway_mfr_result)
         provisioning_successful = True
+        response = gateway_mfr_result
 
-    except subprocess.CalledProcessError:
+    except subprocess.CalledProcessError as exp:
         LOGGER.error("[ECC Provisioning] Exited with a non-zero status")
         provisioning_successful = False
+        response = str(exp)
 
     except Exception as exp:
         LOGGER.error("[ECC Provisioning] Error during provisioning. %s" % str(exp))
         provisioning_successful = False
+        response = str(exp)
+
+    # Try key generation.
+    if provisioning_successful is False and force is True:
+        try:
+            gateway_mfr_result = run_gateway_mfr("key --generate", slot=slot)
+            provisioning_successful = True
+            response = gateway_mfr_result
+
+        except Exception as exp:
+            LOGGER.error("[ECC Provisioning] key --generate failed: %s" % str(exp))
+            response = str(exp)
 
-    return provisioning_successful
+    return provisioning_successful, response
 
 
 def did_gateway_mfr_test_result_include_miner_key_pass(

hm_pyhelper/tests/test_miner_param.py

@@ -235,7 +235,7 @@ def test_run_gateway_mfr_GatewayMFRFileNotFoundException(
         with self.assertRaises(GatewayMFRFileNotFoundException):
             run_gateway_mfr("unittest")
 
-        mocked_get_gateway_mfr_command.assert_called_once_with('unittest')
+        mocked_get_gateway_mfr_command.assert_called_once_with('unittest', slot=0)
         mocked_subprocess_run.assert_called_once_with(
             ['gateway_mfr', 'arg1', 'arg2'],
             capture_output=True, check=True)
@@ -251,7 +251,7 @@ def test_run_gateway_mfr_ECCMalfunctionException(
         with self.assertRaises(ECCMalfunctionException):
             run_gateway_mfr("unittest")
 
-        mocked_get_gateway_mfr_command.assert_called_once_with('unittest')
+        mocked_get_gateway_mfr_command.assert_called_once_with('unittest', slot=0)
         mocked_subprocess_run.assert_called_once_with(
             ['gateway_mfr', 'arg1', 'arg2'],
             capture_output=True, check=True)
@@ -267,7 +267,7 @@ def test_run_gateway_mfr_ResourceBusyError(
         with self.assertRaises(ResourceBusyError):
             run_gateway_mfr("unittest")
 
-        mocked_get_gateway_mfr_command.assert_called_once_with('unittest')
+        mocked_get_gateway_mfr_command.assert_called_once_with('unittest', slot=0)
         mocked_subprocess_run.assert_called_once_with(
             ['gateway_mfr', 'arg1', 'arg2'],
             capture_output=True, check=True)
@@ -296,41 +296,18 @@ def test_get_public_keys_rust(
 
         self.assertDictEqual(actual_result, expected_result)
 
-        mocked_get_gateway_mfr_command.assert_called_once_with('key')
+        mocked_get_gateway_mfr_command.assert_called_once_with('key', slot=0)
         mocked_subprocess_run.assert_called_once_with(
             ['gateway_mfr', 'arg1', 'arg2'],
             capture_output=True, check=True)
 
-    @patch(
-            'hm_pyhelper.miner_param.get_gateway_mfr_test_result',
-            return_value={
-                "result": "pass",
-                "tests": ALL_PASS_GATEWAY_MFR_TESTS
-                }
-    )
-    def test_provision_key_all_passed(
-            self,
-            mocked_get_gateway_mfr_test_result
-    ):
-        self.assertTrue(provision_key())
-        mocked_get_gateway_mfr_test_result.assert_called_once()
+    def test_provision_key_all_passed(self):
+        self.assertTrue(provision_key(slot=0))
 
-    @patch(
-            'hm_pyhelper.miner_param.get_gateway_mfr_test_result',
-            return_value={
-                "result": "fail",
-                "tests": NONE_PASS_GATEWAY_MFR_TESTS
-            }
-    )
     @patch('hm_pyhelper.miner_param.run_gateway_mfr')
-    def test_provision_key_none_passed(
-            self,
-            mocked_get_gateway_mfr_test_result,
-            mocked_run_gateway_mfr
-    ):
-        self.assertTrue(provision_key())
+    def test_provision_key_none_passed(self, mocked_run_gateway_mfr):
+        self.assertTrue(provision_key(slot=0))
 
-        mocked_get_gateway_mfr_test_result.assert_called_once()
         mocked_run_gateway_mfr.assert_called_once()
 
     @patch(

setup.py

@@ -3,7 +3,7 @@
 
 setup(
     name='hm_pyhelper',
-    version='0.13.36',
+    version='0.13.37',
     author="Nebra Ltd",
     author_email="[email protected]",
     description="Helium Python Helper",