fix: remove unnecessary ECC lock and improve bus check (#234)

* fix: remove unnecessary ECC lock and improve bus check

- ECC lock not required on this check because all we are doing is checking the i2c address of the attached device.
- persist a generated ECC location (from the i2c bus) to file to make get_ecc_location() function more efficient and to remove unnecessary calls to the i2c bus
- additional tests for SWARM_KEY_URI_OVERRIDE and new "persist to file" feature

Ref: NebraLtd/helium-syncrobit#10
Ref: NebraLtd/helium-syncrobit#2

* Update miner_param.py

* Update miner_param.py

hm_pyhelper/miner_param.py

@@ -115,8 +115,22 @@ def get_gateway_mfr_version() -> Version:
 def get_ecc_location() -> str:
     ecc_list = get_variant_attribute(os.getenv('VARIANT'), 'SWARM_KEY_URI')
 
+    try:
+        with open("/var/nebra/ecc_file", 'r') as data:
+            generated_ecc_location = str(data.read()).rstrip('\n')
+
+        if len(generated_ecc_location) < 10:
+            generated_ecc_location = None
+        else:
+            LOGGER.info("Generated ECC location file found: " + generated_ecc_location)
+    except FileNotFoundError:
+        # No ECC location file found, create one with value None
+        generated_ecc_location = None
+
     if os.getenv('SWARM_KEY_URI_OVERRIDE'):
         ecc_location = os.getenv('SWARM_KEY_URI_OVERRIDE')
+    elif generated_ecc_location is not None:
+        ecc_location = generated_ecc_location
     elif len(ecc_list) == 1:
         ecc_location = ecc_list[0]
     else:
@@ -129,8 +143,14 @@ def get_ecc_location() -> str:
 
             if config_search_param(command, parameter):
                 ecc_location = location
+                with open("/var/nebra/ecc_file", "w") as file:
+                    file.write(ecc_location)
                 return ecc_location
 
+    if not ecc_location:
+        ecc_location = None
+        LOGGER.info("Can't find ECC. Ensure SWARM_KEY_URI is correct in hardware definitions.")
+
     return ecc_location
 
 
@@ -354,7 +374,6 @@ def await_spi_available(spi_bus):
         raise SPIUnavailableException("SPI bus %s not found!" % spi_bus)
 
 
-@lock_ecc()
 def config_search_param(command, param):
     """
     input:

hm_pyhelper/tests/test_miner_param.py

@@ -9,7 +9,7 @@
     GatewayMFRFileNotFoundException, UnsupportedGatewayMfrVersion
 from hm_pyhelper.lock_singleton import ResourceBusyError
 from hm_pyhelper.miner_param import retry_get_region, await_spi_available, \
-    provision_key, run_gateway_mfr, get_gateway_mfr_path, config_search_param, \
+    provision_key, run_gateway_mfr, get_gateway_mfr_path, config_search_param, get_ecc_location, \
     did_gateway_mfr_test_result_include_miner_key_pass, parse_i2c_address, parse_i2c_bus, \
     get_mac_address, get_public_keys_rust, get_gateway_mfr_version, get_gateway_mfr_command
 
@@ -76,6 +76,9 @@
         }
     }
 
+ECC_FILE_DATA = 'ecc://i2c-Y:96?slot=1'
+ECC_FILE_DATA_BLANK = None
+
 
 class SubprocessResult(object):
     def __init__(self, stdout=None, stderr=None):
@@ -164,6 +167,20 @@ def test_get_gateway_mfr_command_v021_no_SWARM_KEY_URI(self, mocked_get_gateway_
         expected_result = [ANY, 'test']
         self.assertListEqual(actual_result, expected_result)
 
+    @patch.dict('os.environ', {"SWARM_KEY_URI_OVERRIDE": "override-test"})
+    @patch('hm_pyhelper.miner_param.get_gateway_mfr_version',
+           return_value=Version('0.2.1'))
+    def test_get_gateway_mfr_command_v021_override(self, mocked_get_gateway_mfr_version):
+        actual_result = get_gateway_mfr_command('key')
+        expected_result = [ANY, '--device', 'override-test', 'key']
+        self.assertListEqual(actual_result, expected_result)
+        mocked_get_gateway_mfr_version.assert_called_once()
+
+        actual_result = get_gateway_mfr_command('test')
+        expected_result = [ANY, '--device', 'override-test', 'test']
+        self.assertListEqual(actual_result, expected_result)
+
+    @patch("builtins.open", mock_open(read_data=ECC_FILE_DATA_BLANK))
     @patch.dict('os.environ', {"VARIANT": "NEBHNT-MULTIPLE-ECC-ADDRESS"})
     @patch('subprocess.Popen')
     @patch('hm_pyhelper.miner_param.get_gateway_mfr_version',
@@ -180,6 +197,13 @@ def test_get_gateway_mfr_command_v021_multi_SWARM_KEY_URI(self, mocked_get_gatew
         self.assertListEqual(actual_result, expected_result)
         mocked_get_gateway_mfr_version.assert_called_once()
 
+    @patch("builtins.open", mock_open(read_data=ECC_FILE_DATA_BLANK))
+    @patch.dict('os.environ', {"VARIANT": "NEBHNT-MULTIPLE-ECC-ADDRESS"})
+    @patch('subprocess.Popen')
+    @patch('hm_pyhelper.miner_param.get_gateway_mfr_version',
+           return_value=Version('0.2.1'))
+    def test_get_gateway_mfr_command_v021_multi_SWARM_KEY_58(self, mocked_get_gateway_mfr_version,
+                                                             mock_subproc_popen):
         process_mock = Mock()
         attrs = {'communicate.return_value': (str.encode("58 --"), 'error')}
         process_mock.configure_mock(**attrs)
@@ -232,6 +256,12 @@ def test_get_gateway_mfr_command_unsupported_version(self, mocked_get_gateway_mf
             get_gateway_mfr_command('key')
         mocked_get_gateway_mfr_version.assert_called_once()
 
+    @patch("builtins.open", mock_open(read_data=ECC_FILE_DATA))
+    def test_get_ecc_location_generated_ecc(self):
+        actual_result = get_ecc_location()
+        expected_result = 'ecc://i2c-Y:96?slot=1'
+        self.assertEqual(actual_result, expected_result)
+
     @patch('hm_pyhelper.miner_param.get_gateway_mfr_command',
            return_value=['gateway_mfr', 'arg1', 'arg2'])
     @patch('subprocess.run', side_effect=FileNotFoundError())

setup.py

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