refactor: initFile generation

There were a few places where the initFile endpoint was previously
returning 500. The primary goal was to ensure a response is always
returned, even when the miner is not fully operational.

To make this usage explicit, a DiagnosticsReport class has
been created that will never throw an exception. Exceptions
are stringified.

In the process, each type of test has been refactored into a
distinct Diagnostic. A DiagnosticReport is constructed with
an arbitrary set of Diagnostics. The result of each is appended
to the DiagnosticsReport dict.

The current keys used for the initFile are terse and confusing,
eg PF, OK, etc. In an attempt to migrate towards human friendly
names, each Diagnostic uses the legacy keys in addition
to new verbose_human_friendly_keys.

README.md

@@ -76,6 +76,14 @@ Because the stack is tightly intertwined with Balena, the easiest way to test th
 
 If you are on the same network as the Raspberry Pi, enter `LOCAL IP ADDRESS` from Balena into the browser.
 
+### Testing
+
+```
+pip install -r test-requirements.txt
+pytest
+flake8 hw_diag
+```
+
 ### Deprecated deployment
 This is no longer [the recommended way](https://www.balena.io/docs/learn/deploy/deployment/#overview) of doing Balena deployments.
 

hw_diag/diagnostics/bt_lte_diagnostic.py

@@ -0,0 +1,71 @@
+import os
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+
+
+DEVICE_TYPE_ADDRESS_MAPPINGS = [
+    {
+        'dev_type': 'BT',
+        'dev_addr': '0a12'
+    },
+    {
+        'dev_type': 'LTE',  # Quectel
+        'dev_addr': '2c7c'
+    },
+    {
+        'dev_type': 'LTE',  # Sierra Wireless MC7700
+        'dev_addr': '68a2'
+    },
+    {
+        'dev_type': 'LTE',  # Telit / Reyax
+        'dev_addr': '1bc7'
+    },
+    {
+        'dev_type': 'LTE',  # SimCom SIM7100E
+        'dev_addr': '1e0e'
+    },
+    {
+        'dev_type': 'LTE',  # Huawei ME909s-120
+        'dev_addr': '12d1'
+    },
+    {
+        'dev_type': 'LTE',  # MikroTik R11e-LTE6
+        'dev_addr': '2cd2'
+    }
+]
+
+
+class BtLteDiagnostic(Diagnostic):
+    def __init__(self, dev_type, dev_addr):
+        super(BtLteDiagnostic, self).__init__(dev_type, dev_type)
+        self.dev_addr = dev_addr
+
+    def perform_test(self, diagnostics_report):
+        resp = os.popen(
+            'grep %s /sys/bus/usb/devices/*/idVendor' % self.dev_addr
+        ).read()
+
+        if self.dev_addr in resp:
+            print("setting %s to %s" % (self.key, True))
+            diagnostics_report.record_result(True, self)
+        else:
+            # Only set the dev_type to False if it is not already true
+            if self.key not in diagnostics_report:
+                diagnostics_report.record_result(False, self)
+
+            dev_type_already_true = diagnostics_report[self.key]
+            if not dev_type_already_true:
+                diagnostics_report.record_result(False, self)
+
+
+class BtLteDiagnostics():
+    def __init__(self):
+        def get_diagnostic_for_dev(bt_lte_mapping):
+            return BtLteDiagnostic(bt_lte_mapping['dev_type'],
+                                   bt_lte_mapping['dev_addr'])
+
+        self.bt_lte_diagnostics = map(get_diagnostic_for_dev,
+                                      DEVICE_TYPE_ADDRESS_MAPPINGS)
+
+    def perform_test(self, diagnostics_report):
+        for bt_lte_diagnostic in self.bt_lte_diagnostics:
+            bt_lte_diagnostic.perform_test(diagnostics_report)

hw_diag/diagnostics/ecc_diagnostic.py

@@ -0,0 +1,26 @@
+import json
+from hm_pyhelper.miner_param import get_gateway_mfr_test_result
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+from hm_pyhelper.exceptions import ECCMalfunctionException
+
+KEY = 'ECC'
+FRIENDLY_NAME = "ECC"
+
+
+class EccDiagnostic(Diagnostic):
+    def __init__(self):
+        super(EccDiagnostic, self).__init__(KEY, FRIENDLY_NAME)
+
+    def perform_test(self, diagnostics_report):
+        try:
+            ecc_tests = get_gateway_mfr_test_result()
+
+            if ecc_tests['result'] == 'pass':
+                diagnostics_report.record_result(True, self)
+            else:
+                msg = "gateway_mfr test finished with error, %s" % \
+                      str(json.dumps(ecc_tests))
+                diagnostics_report.record_failure(msg, self)
+
+        except ECCMalfunctionException as e:
+            diagnostics_report.record_failure(str(e), self)

hw_diag/diagnostics/env_var_diagnostics.py

@@ -0,0 +1,49 @@
+import os
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+
+ENV_VARS_MAPPING = [{
+    'ENV_VAR': 'BALENA_DEVICE_NAME_AT_INIT',
+    'DIAGNOSTIC_KEY': 'BN'
+}, {
+    'ENV_VAR': 'BALENA_DEVICE_UUID',
+    'DIAGNOSTIC_KEY': 'ID'
+}, {
+    'ENV_VAR': 'BALENA_APP_NAME',
+    'DIAGNOSTIC_KEY': 'BA'
+}, {
+    'ENV_VAR': 'FREQ',
+    'DIAGNOSTIC_KEY': 'FR'
+}, {
+    'ENV_VAR': 'FIRMWARE_VERSION',
+    'DIAGNOSTIC_KEY': 'FW'
+}, {
+    'ENV_VAR': 'VARIANT',
+    'DIAGNOSTIC_KEY': 'VA'
+}]
+
+
+class EnvVarDiagnostic(Diagnostic):
+    def __init__(self, key, friendly_key):
+        super(EnvVarDiagnostic, self).__init__(key, friendly_key)
+
+    def perform_test(self, diagnostics_report):
+        env_value = os.getenv(self.friendly_key)
+        if not env_value:
+            diagnostics_report.record_failure(
+                "Env var %s not set" % self.friendly_key, self)
+        else:
+            diagnostics_report.record_result(env_value, self)
+
+
+class EnvVarDiagnostics():
+    def __init__(self):
+        def get_diagnostic_for_env_var(env_var_mapping):
+            return EnvVarDiagnostic(env_var_mapping['DIAGNOSTIC_KEY'],
+                                    env_var_mapping['ENV_VAR'])
+
+        self.env_var_diagnostics = map(get_diagnostic_for_env_var,
+                                       ENV_VARS_MAPPING)
+
+    def perform_test(self, diagnostics_report):
+        for env_var_diagnostic in self.env_var_diagnostics:
+            env_var_diagnostic.perform_test(diagnostics_report)

hw_diag/diagnostics/key_diagnostics.py

@@ -0,0 +1,47 @@
+from hm_pyhelper.miner_param import get_public_keys_rust
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+
+
+KEY_MAPPINGS = [
+    {
+        'key': 'OK',
+        'friendly_key': 'onboarding_key',
+        'key_path': 'key'
+    },
+    {
+        'key': 'PK',
+        'friendly_key': 'public_key',
+        'key_path': 'key'
+    }
+]
+
+
+class KeyDiagnostic(Diagnostic):
+    def __init__(self, key, friendly_key, key_path):
+        super(KeyDiagnostic, self).__init__(key, friendly_key)
+        self.key_path = key_path
+
+    def perform_test(self, diagnostics_report):
+        public_keys = get_public_keys_rust()
+        try:
+            diagnostics_report.record_result(public_keys[self.key_path], self)
+
+        except KeyError:
+            err_msg = "Key %s not found" % self.key_path
+            diagnostics_report.record_failure(err_msg, self)
+
+
+class KeyDiagnostics:
+    def __init__(self):
+        def get_diagnostic_for_key(key_mapping):
+            return KeyDiagnostic(
+              key_mapping['key'],
+              key_mapping['friendly_key'],
+              key_mapping['key_path'])
+
+        self.key_diagnostics = map(get_diagnostic_for_key,
+                                   KEY_MAPPINGS)
+
+    def perform_test(self, diagnostics_report):
+        for key_diagnostic in self.key_diagnostics:
+            key_diagnostic.perform_test(diagnostics_report)

hw_diag/diagnostics/lora_diagnostic.py

@@ -0,0 +1,16 @@
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+from hw_diag.utilities.hardware import lora_module_test
+
+KEY = 'LOR'
+FRIENDLY_NAME = "lora"
+
+
+class LoraDiagnostic(Diagnostic):
+    def __init__(self):
+        super(LoraDiagnostic, self).__init__(KEY, FRIENDLY_NAME)
+
+    def perform_test(self, diagnostics_report):
+        if lora_module_test():
+            diagnostics_report.record_result(True, self)
+        else:
+            diagnostics_report.record_failure(False, self)

hw_diag/diagnostics/mac_diagnostics.py

@@ -0,0 +1,45 @@
+from hm_pyhelper.miner_param import get_mac_address
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+
+
+INTERFACE_MAPPINGS = [
+  {
+    'key': 'E0',
+    'friendly_key': 'eth_mac_address',
+    'mac_filepath': '/sys/class/net/eth0/address'
+  },
+  {
+    'key': 'W0',
+    'friendly_key': 'wifi_mac_address',
+    'mac_filepath': '/sys/class/net/wlan0/address'
+  }
+]
+
+
+class MacDiagnostic(Diagnostic):
+    def __init__(self, key, friendly_key, mac_filepath):
+        super(MacDiagnostic, self).__init__(key, friendly_key)
+        self.mac_filepath = mac_filepath
+
+    def perform_test(self, diagnostics_report):
+        try:
+            mac_address = get_mac_address(self.mac_filepath)
+            diagnostics_report.record_result(mac_address, self)
+        except Exception as e:
+            diagnostics_report.record_failure(str(e), self)
+
+
+class MacDiagnostics():
+    def __init__(self):
+        def get_diagnostic_for_interface(interface_info):
+            return MacDiagnostic(
+              interface_info['key'],
+              interface_info['friendly_key'],
+              interface_info['mac_filepath'])
+
+        self.mac_diagnostics = map(get_diagnostic_for_interface,
+                                   INTERFACE_MAPPINGS)
+
+    def perform_test(self, diagnostics_report):
+        for mac_diagnostic in self.mac_diagnostics:
+            mac_diagnostic.perform_test(diagnostics_report)

hw_diag/diagnostics/pf_diagnostic.py

@@ -0,0 +1,22 @@
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+
+KEY = 'PF'
+FRIENDLY_NAME = "legacy_pass_fail"
+CHECK_KEYS = ["ECC", "E0", "W0", "BT", "LOR"]
+
+
+class PfDiagnostic(Diagnostic):
+    def __init__(self):
+        super(PfDiagnostic, self).__init__(KEY, FRIENDLY_NAME)
+
+    def perform_test(self, diagnostics_report):
+        def get_result(key):
+            return key in diagnostics_report and \
+                   diagnostics_report[key]
+
+        all_passed = all(map(get_result, CHECK_KEYS))
+
+        if all_passed:
+            diagnostics_report.record_result(True, self)
+        else:
+            diagnostics_report.record_failure(False, self)

hw_diag/diagnostics/rpi_serial_diagnostic.py

@@ -0,0 +1,23 @@
+
+from hm_pyhelper.diagnostics.diagnostic import Diagnostic
+
+KEY = 'RPI'
+FRIENDLY_NAME = "raspberry_pi_serial_number"
+SERIAL_FILEPATH = "/proc/cpuinfo"
+
+
+class RpiSerialDiagnostic(Diagnostic):
+    def __init__(self):
+        super(RpiSerialDiagnostic, self). \
+            __init__(KEY, FRIENDLY_NAME)
+
+    def perform_test(self, diagnostics_report):
+        try:
+            rpi_serial = open(SERIAL_FILEPATH).readlines()[-2].strip()[10:]
+            diagnostics_report.record_result(rpi_serial, self)
+
+        except FileNotFoundError as e:
+            diagnostics_report.record_failure(e, self)
+
+        except PermissionError as e:
+            diagnostics_report.record_failure(e, self)

hw_diag/tests/diagnostics/test_bt_lte_diagnostics.py

@@ -0,0 +1,51 @@
+import unittest
+from unittest.mock import patch
+
+from hm_pyhelper.diagnostics.diagnostics_report import \
+    DIAGNOSTICS_PASSED_KEY, DIAGNOSTICS_ERRORS_KEY, DiagnosticsReport
+
+from hw_diag.diagnostics.bt_lte_diagnostic import \
+    BtLteDiagnostic, BtLteDiagnostics
+
+
+class TestBtLteDiagnostics(unittest.TestCase):
+    @patch('os.popen')
+    def test_success(self, mock):
+        mock().read.return_value = 'dev_addr'
+        diagnostic = BtLteDiagnostic('DEV_TYPE', 'dev_addr')
+        diagnostics_report = DiagnosticsReport([diagnostic])
+        diagnostics_report.perform_diagnostics()
+
+        self.assertDictEqual(diagnostics_report, {
+            DIAGNOSTICS_PASSED_KEY: True,
+            DIAGNOSTICS_ERRORS_KEY: [],
+            'DEV_TYPE': True
+        })
+
+    @patch('os.popen')
+    def test_error(self, mock):
+        mock().read.return_value = 'INVALID'
+        diagnostic = BtLteDiagnostic('DEV_TYPE', 'dev_addr')
+        diagnostics_report = DiagnosticsReport([diagnostic])
+        diagnostics_report.perform_diagnostics()
+
+        self.assertDictEqual(diagnostics_report, {
+            DIAGNOSTICS_PASSED_KEY: True,
+            DIAGNOSTICS_ERRORS_KEY: [],
+            'DEV_TYPE': False
+        })
+
+    @patch('os.popen')
+    def test_all_success(self, mock):
+        mock().read.return_value = '0a12, 2c7c, 68a2, 1bc7, 1e0e, 12d1, 2cd2'
+
+        diagnostic = BtLteDiagnostics()
+        diagnostics_report = DiagnosticsReport([diagnostic])
+        diagnostics_report.perform_diagnostics()
+
+        self.assertDictEqual(diagnostics_report, {
+            DIAGNOSTICS_PASSED_KEY: True,
+            DIAGNOSTICS_ERRORS_KEY: [],
+            'BT': True,
+            'LTE': True,
+        })