Increased pcied unit test coverage to > 80% (#243)

sonic-pcied/setup.py

@@ -21,11 +21,10 @@
         'enum34; python_version < "3.4"',
         'sonic-py-common',
     ],
-    tests_requires=[
+    tests_require=[
         'mock>=2.0.0; python_version < "3.3"',
         'pytest',
         'pytest-cov',
-        'sonic-platform-common'
     ],
     classifiers=[
         'Development Status :: 4 - Beta',

sonic-pcied/tests/mocked_libs/sonic_platform/__init__.py

@@ -0,0 +1,6 @@
+"""
+    Mock implementation of sonic_platform package for unit testing
+"""
+
+from . import pcie
+

sonic-pcied/tests/mocked_libs/sonic_platform/pcie.py

@@ -0,0 +1,13 @@
+"""
+    Mock implementation of sonic_platform package for unit testing
+"""
+
+from sonic_platform_base.pcie_base import PcieBase
+
+
+class Pcie(PcieBase):
+    def __init__(self):
+        self.platform_pcieutil = "/tmp/Pcie"
+
+    def __str__(self):
+        return self.platform_pcieutil

sonic-pcied/tests/mocked_libs/sonic_platform_base/__init__.py

@@ -0,0 +1 @@
+

sonic-pcied/tests/mocked_libs/sonic_platform_base/sonic_pcie/pcie_base.py

@@ -0,0 +1,65 @@
+#
+# pcie_base.py
+#
+# Abstract base class for implementing platform-specific
+#  PCIE functionality for SONiC
+#
+
+try:
+    import abc
+except ImportError as e:
+    raise ImportError(str(e) + " - required module not found")
+
+
+class PcieBase(object):
+    def __init__(self, path):
+        """
+         Constructor
+
+         Args:
+           pcieutil file and config file path
+        """
+
+    @abc.abstractmethod
+    def get_pcie_device(self):
+        """
+         get current device pcie info
+
+          Returns:
+            A list including pcie device info
+        """
+        return []
+
+    @abc.abstractmethod
+    def get_pcie_check(self):
+        """
+         Check Pcie device with config file
+
+         Returns:
+            A list including pcie device and test result info
+        """
+        return []
+
+    @abc.abstractmethod
+    def get_pcie_aer_stats(self, domain, bus, dev, fn):
+        """
+        Returns a nested dictionary containing the AER stats belonging to a
+        PCIe device
+
+        Args:
+            domain, bus, dev, fn: Domain, bus, device, function of the PCIe
+            device respectively
+
+        Returns:
+            A nested dictionary where key is severity 'correctable', 'fatal' or
+            'non_fatal', value is a dictionary of key, value pairs in the format:
+                {'AER Error type': Error count}
+
+            Ex. {'correctable': {'BadDLLP': 0, 'BadTLP': 0},
+                 'fatal': {'RxOF': 0, 'MalfTLP': 0},
+                 'non_fatal': {'RxOF': 0, 'MalfTLP': 0}}
+
+            For PCIe devices that do not support AER, the value for each
+            severity key is an empty dictionary.
+        """
+        return {}

sonic-pcied/tests/mocked_libs/sonic_platform_base/sonic_pcie/pcie_common.py

@@ -0,0 +1,146 @@
+# pcie_common.py
+# Common PCIE check interfaces for SONIC
+#
+
+import os
+import yaml
+import subprocess
+import re
+import sys
+from copy import deepcopy
+try:
+    from .pcie_base import PcieBase
+except ImportError as e:
+    raise ImportError(str(e) + "- required module not found")
+
+
+class PcieUtil(PcieBase):
+    """Platform-specific PCIEutil class"""
+    # got the config file path
+    def __init__(self, path):
+        self.config_path = path
+        self._conf_rev = None
+
+    # load the config file
+    def load_config_file(self):
+        conf_rev = "_{}".format(self._conf_rev) if self._conf_rev else ""
+        config_file = "{}/pcie{}.yaml".format(self.config_path, conf_rev)
+        try:
+            with open(config_file) as conf_file:
+                self.confInfo = yaml.safe_load(conf_file)
+        except IOError as e:
+            print("Error: {}".format(str(e)))
+            print("Not found config file, please add a config file manually, or generate it by running [pcieutil pcie_generate]")
+            sys.exit()
+
+    # load current PCIe device
+    def get_pcie_device(self):
+        pciDict = {}
+        pciList = []
+        p1 = "^(\w+):(\w+)\.(\w)\s(.*)\s*\(*.*\)*"
+        p2 = "^.*:.*:.*:(\w+)\s*\(*.*\)*"
+        command1 = "sudo lspci"
+        command2 = "sudo lspci -n"
+        # run command 1
+        proc1 = subprocess.Popen(command1, shell=True, universal_newlines=True, stdout=subprocess.PIPE)
+        output1 = proc1.stdout.readlines()
+        (out, err) = proc1.communicate()
+        # run command 2
+        proc2 = subprocess.Popen(command2, shell=True, universal_newlines=True, stdout=subprocess.PIPE)
+        output2 = proc2.stdout.readlines()
+        (out, err) = proc2.communicate()
+
+        if proc1.returncode > 0:
+            for line1 in output1:
+                print(line1.strip())
+            return
+        elif proc2.returncode > 0:
+            for line2 in output2:
+                print(line2.strip())
+            return
+        else:
+            for (line1, line2) in zip(output1, output2):
+                pciDict.clear()
+                match1 = re.search(p1, line1.strip())
+                match2 = re.search(p2, line2.strip())
+                if match1 and match2:
+                    Bus = match1.group(1)
+                    Dev = match1.group(2)
+                    Fn = match1.group(3)
+                    Name = match1.group(4)
+                    Id = match2.group(1)
+                    pciDict["name"] = Name
+                    pciDict["bus"] = Bus
+                    pciDict["dev"] = Dev
+                    pciDict["fn"] = Fn
+                    pciDict["id"] = Id
+                    pciList.append(pciDict)
+                    pciDict = deepcopy(pciDict)
+                else:
+                    print("CAN NOT MATCH PCIe DEVICE")
+        return pciList
+
+    # check the sysfs tree for each PCIe device
+    def check_pcie_sysfs(self, domain=0, bus=0, device=0, func=0):
+        dev_path = os.path.join('/sys/bus/pci/devices', '%04x:%02x:%02x.%d' % (domain, bus, device, func))
+        if os.path.exists(dev_path):
+            return True
+        return False
+
+    # check the current PCIe device with config file and return the result
+    def get_pcie_check(self):
+        self.load_config_file()
+        for item_conf in self.confInfo:
+            bus_conf = item_conf["bus"]
+            dev_conf = item_conf["dev"]
+            fn_conf = item_conf["fn"]
+            if self.check_pcie_sysfs(bus=int(bus_conf, base=16), device=int(dev_conf, base=16), func=int(fn_conf, base=16)):
+                item_conf["result"] = "Passed"
+            else:
+                item_conf["result"] = "Failed"
+        return self.confInfo
+
+    # return AER stats of PCIe device
+    def get_pcie_aer_stats(self, domain=0, bus=0, dev=0, func=0):
+        aer_stats = {'correctable': {}, 'fatal': {}, 'non_fatal': {}}
+        dev_path = os.path.join('/sys/bus/pci/devices', '%04x:%02x:%02x.%d' % (domain, bus, dev, func))
+
+        # construct AER sysfs filepath
+        correctable_path = os.path.join(dev_path, "aer_dev_correctable")
+        fatal_path = os.path.join(dev_path, "aer_dev_fatal")
+        non_fatal_path = os.path.join(dev_path, "aer_dev_nonfatal")
+
+        # update AER-correctable fields
+        if os.path.isfile(correctable_path):
+            with open(correctable_path, 'r') as fh:
+                lines = fh.readlines()
+            for line in lines:
+                correctable_field, value = line.split()
+                aer_stats['correctable'][correctable_field] = value
+
+        # update AER-Fatal fields
+        if os.path.isfile(fatal_path):
+            with open(fatal_path, 'r') as fh:
+                lines = fh.readlines()
+            for line in lines:
+                fatal_field, value = line.split()
+                aer_stats['fatal'][fatal_field] = value
+
+        # update AER-Non Fatal fields
+        if os.path.isfile(non_fatal_path):
+            with open(non_fatal_path, 'r') as fh:
+                lines = fh.readlines()
+            for line in lines:
+                non_fatal_field, value = line.split()
+                aer_stats['non_fatal'][non_fatal_field] = value
+
+        return aer_stats
+
+    # generate the config file with current pci device
+    def dump_conf_yaml(self):
+        curInfo = self.get_pcie_device()
+        conf_rev = "_{}".format(self._conf_rev) if self._conf_rev else ""
+        config_file = "{}/pcie{}.yaml".format(self.config_path, conf_rev)
+        with open(config_file, "w") as conf_file:
+            yaml.dump(curInfo, conf_file, default_flow_style=False)
+        return

sonic-pcied/tests/test_DaemonPcied.py

@@ -38,12 +38,10 @@
 {'correctable': {}, 'fatal': {}, 'non_fatal': {}}
 """
 
-pcie_aer_stats_no_err = \
-"""
-{'correctable': {'field1': '0', 'field2': '0'},
+pcie_aer_stats_no_err = {'correctable': {'field1': '0', 'field2': '0'},
  'fatal': {'field3': '0', 'field4': '0'},
  'non_fatal': {'field5': '0', 'field6': '0'}}
-"""
+
 
 pcie_aer_stats_err = \
 """
@@ -207,9 +205,9 @@ def test_update_aer_to_statedb(self):
         daemon_pcied.log_debug = mock.MagicMock()
         daemon_pcied.device_table = mock.MagicMock()
         daemon_pcied.device_name = mock.MagicMock()
-        daemon_pcied.aer_stats = mock.MagicMock()
-
+        daemon_pcied.aer_stats = pcie_aer_stats_no_err
 
+        """
         mocked_expected_fvp = pcied.swsscommon.FieldValuePairs(
             [("correctable|field1", '0'),
              ("correctable|field2", '0'),
@@ -218,9 +216,10 @@ def test_update_aer_to_statedb(self):
              ("non_fatal|field5", '0'),
              ("non_fatal|field6", '0'),
              ])
+        """
 
         daemon_pcied.update_aer_to_statedb() 
-        assert daemon_pcied.log_debug.call_count == 1
-        assert daemon_pcied.device_table.set.call_count == 0
+        assert daemon_pcied.log_debug.call_count == 0
+        assert daemon_pcied.device_table.set.call_count == 1
 
         daemon_pcied.device_table.set.reset_mock()

sonic-pcied/tests/test_pcied.py

@@ -5,12 +5,12 @@
 import pytest
 
 # TODO: Clean this up once we no longer need to support Python 2
-if sys.version_info.major == 3:
-    from unittest import mock
+if sys.version_info >= (3, 3):
+    from unittest.mock import MagicMock, patch, mock_open
 else:
-    import mock
-from sonic_py_common import daemon_base, device_info
+    from mock import MagicMock, patch, mock_open
 
+from sonic_py_common import daemon_base, device_info
 from .mock_platform import MockPcieUtil
 
 tests_path = os.path.dirname(os.path.abspath(__file__))
@@ -27,17 +27,33 @@
 import pcied
 
 
-daemon_base.db_connect = mock.MagicMock()
-
+daemon_base.db_connect = MagicMock()
 
 SYSLOG_IDENTIFIER = 'pcied_test'
 NOT_AVAILABLE = 'N/A'
 
 
-@mock.patch('pcied.load_platform_pcieutil', mock.MagicMock())
-@mock.patch('pcied.DaemonPcied.run')
+@patch('pcied.load_platform_pcieutil', MagicMock())
+@patch('pcied.DaemonPcied.run')
 def test_main(mock_run):
     mock_run.return_value = False
 
     pcied.main()
     assert mock_run.call_count == 1
+
+
+@patch('pcied.os.path.exists', MagicMock(return_value=True))
+def test_read_id_file():
+
+    device_name = "test"
+
+    with patch('builtins.open', new_callable=mock_open, read_data='15') as mock_fd:
+        rc = pcied.read_id_file(device_name)
+        assert rc == "15"
+
+@patch('pcied.device_info.get_paths_to_platform_and_hwsku_dirs', MagicMock(return_value=('/tmp', None)))
+def test_load_platform_pcieutil():
+    from sonic_platform_base.sonic_pcie.pcie_common import PcieUtil
+    rc = pcied.load_platform_pcieutil()
+
+    assert isinstance(rc, PcieUtil)