Added a procfs check for pcie peripherals to verify device ID.

sonic_platform_base/sonic_pcie/pcie_common.py

@@ -10,16 +10,22 @@
 from copy import deepcopy
 try:
     from .pcie_base import PcieBase
+    import binascii
+    from sonic_py_common import logger
 except ImportError as e:
     raise ImportError(str(e) + "- required module not found")
 
+# Enable logging by classes and helper functions
+SYSLOG_IDENTIFIER = "PcieUtil"
+log = logger.Logger(SYSLOG_IDENTIFIER)
 
 class PcieUtil(PcieBase):
     """Platform-specific PCIEutil class"""
     # got the config file path
     def __init__(self, path):
         self.config_path = path
         self._conf_rev = None
+        self.procfs_path = "/proc/bus/pci/"
 
     # load the config file
     def load_config_file(self):
@@ -29,8 +35,8 @@ def load_config_file(self):
             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]")
+            log.log_error("Error: {}".format(str(e)))
+            log.log_error("Not found config file, please add a config file manually, or generate it by running [pcieutil pcie_generate]")
             sys.exit()
 
     # load current PCIe device
@@ -77,9 +83,38 @@ def get_pcie_device(self):
                     pciList.append(pciDict)
                     pciDict = deepcopy(pciDict)
                 else:
-                    print("CAN NOT MATCH PCIe DEVICE")
+                    log.log_warning("CAN NOT MATCH PCIe DEVICE")
         return pciList
 
+    # Check device ID from procfs file check for each PCI device
+    def check_pcie_deviceid(self, bus="", device="", fn="", id=""):
+        current_file = os.path.join(self.procfs_path, bus, "{}.{}".format(device, fn))
+        pcie_device = ("{}:{}.{}".format(bus, device, fn))
+        transaction_check = ['setpci', '-s', pcie_device, '2.w']
+        try:
+            with open(current_file, 'rb') as f:
+                f.seek(2)
+                data = f.read(2)
+                hexstring = binascii.hexlify(data).decode('ascii')
+                procfs_id = str(hexstring[-2:] + hexstring[:2])
+                if id != procfs_id:
+                    log.log_notice("PCIe YAML file device ID mismatch for {}:{}.{} - expected {}, got {}".format(bus, device, fn, id, procfs_id))
+
+                output = subprocess.check_output(transaction_check)
+                transaction_check_result = output.decode('utf8').strip()
+
+                if procfs_id == transaction_check_result:
+                    return True
+                else:
+                    log.log_warning("PCIe transaction check device ID mismatch for {}:{}.{} - expected {}, got {}".format(bus, device, fn, procfs_id, transaction_check_result))
+
+                return False
+        except OSError as osex:
+            log.log_warning("Ecountered {} while trying to open {}".format(str(osex), current_file))
+            return False
+
+
+
     # 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))
@@ -93,8 +128,13 @@ def get_pcie_check(self):
         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)):
+            fn_conf  = item_conf["fn"]
+            id_conf  = item_conf["id"]
+
+            sysfs_check = self.check_pcie_sysfs(bus=int(bus_conf, base=16), device=int(dev_conf, base=16), func=int(fn_conf, base=16))
+            deviceid_check = self.check_pcie_deviceid(bus_conf, dev_conf, fn_conf, id_conf)
+
+            if sysfs_check and deviceid_check:
                 item_conf["result"] = "Passed"
             else:
                 item_conf["result"] = "Failed"

tests/pcie_common_test.py

@@ -150,6 +150,35 @@ def subprocess_popen_side_effect(*args, **kwargs):
         result = pcieutil.get_pcie_device()
         assert result == pcie_device_list
 
+    @mock.patch('subprocess.check_output')
+    def test_check_pcie_deviceid(self, subprocess_check_output_mock):
+        bus = "00"
+        dev = "01"
+        fn  = "1"
+        id  = "0001"
+        test_binary_file = b'\x01\x00\x00\x00'
+
+        def subprocess_check_output_side_effect(*args, **kwargs):
+            return ("0001".encode("utf-8"))
+
+        subprocess_check_output_mock.side_effect = subprocess_check_output_side_effect
+
+        pcieutil = PcieUtil(tests_dir)
+        with mock.patch('builtins.open', new_callable=mock.mock_open, read_data=test_binary_file) as mock_fd:
+            result = pcieutil.check_pcie_deviceid(bus, dev, fn, id)
+            assert result == True
+
+    def test_check_pcie_deviceid_mismatch(self):
+        bus = "00"
+        dev = "01"
+        fn  = "1"
+        id  = "0001"
+        test_binary_file = b'\x02\x03\x00\x00'
+        pcieutil = PcieUtil(tests_dir)
+        with mock.patch('builtins.open', new_callable=mock.mock_open, read_data=test_binary_file) as mock_fd:
+            result = pcieutil.check_pcie_deviceid(bus, dev, fn, id)
+            assert result == False
+
     @mock.patch('os.path.exists')
     def test_get_pcie_check(self, os_path_exists_mock):
 
@@ -159,6 +188,7 @@ def os_path_exists_side_effect(*args):
         os_path_exists_mock.side_effect = os_path_exists_side_effect
         pcieutil = PcieUtil(tests_dir)
         sample_pcie_config = yaml.dump(pcie_device_list)
+        pcieutil.check_pcie_deviceid = mock.MagicMock()
 
         open_mock = mock.mock_open(read_data=sample_pcie_config)
         with mock.patch('{}.open'.format(BUILTINS), open_mock):