utils: Add PCI device, drivers, modules, dmesg related avocado utilities #6225
Conversation
This patch introduces a utility to verify the presence of a given string pattern in kernel logs. The function first searches through `dmesg` output and, if not found, falls back to `journalctl -k -b`. The helper returns True if the pattern is detected in either source, otherwise False. This provides a simple and reusable interface for tests that need to validate kernel messages without duplicating log parsing logic. Signed-off-by: Dheeraj Kumar Srivastava <dheerajkumar.srivastava@amd.com>
This patch adds a helper to validate a given kernel configuration option and load the corresponding module if available. The utility performs the following: * checks if the specified config is set as built-in, module, or not set * skips loading if built-in (already present in the kernel) * attempts to load the module if configured as loadable * raises an error if the config is not set or the module fails to load This provides a cleaner interface for conditional module loading in tests and utilities. Signed-off-by: Dheeraj Kumar Srivastava <dheerajkumar.srivastava@amd.com>
📝 WalkthroughWalkthroughAdds utilities for searching kernel logs, configuring kernel modules, and managing PCI drivers/MSI-X checks; updates the selftests count and adds unit tests for the new utilities. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Pre-merge checks❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 Pylint (4.0.0)selftests/check.pyavocado/utils/pci.pyselftests/unit/utils/pci.pyThanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (12)
avocado/utils/linux_modules.py (1)
279-285: Minor: exception message style (TRY003)Static analysis suggests avoiding long messages in generic exceptions. Optional: shorten messages or define a specific exception (e.g., ModuleLoadError) to carry details.
avocado/utils/pci.py (4)
379-400: Use errno to detect EEXIST instead of matching error textString-matching "File exists" is brittle and locale-dependent. Prefer checking the wrapped OSError’s errno.
Apply this diff:
def add_vendor_id(full_pci_address, driver): @@ try: genio.write_file_or_fail(f"/sys/bus/pci/drivers/{driver}/new_id", f"{vid}\n") except genio.GenIOError as e: - if "File exists" in str(e): - return + cause = getattr(e, "__cause__", None) + # Ignore if the ID is already present + if isinstance(cause, OSError) and getattr(cause, "errno", None) == errno.EEXIST: + return raise ValueError( f"Failed to add vendor ID '{vid}' to driver '{driver}': {e}" ) from eAlso add the missing import at the top of this module:
import errno
414-433: attach_driver(): add rollback on failure and validate driver name
- If bind() fails after unbinding, consider best-effort rollback to the previous driver.
- Optionally validate driver against a safe pattern to avoid path traversal into sysfs.
Example improvement:
def attach_driver(full_pci_address, driver): @@ - try: - if not driver or not full_pci_address: - raise ValueError("'driver' or 'full_pci_address' inputs are None") + cur_driver = None + try: + if not driver or not full_pci_address: + raise ValueError("'driver' or 'full_pci_address' inputs are None") + if not re.fullmatch(r"[A-Za-z0-9_\-]+", driver): + raise ValueError(f"Invalid driver name: {driver}") @@ - cur_driver = get_driver(full_pci_address) + cur_driver = get_driver(full_pci_address) if cur_driver is not None: unbind(cur_driver, full_pci_address) @@ - except OSError as e: + except OSError as e: + # Best-effort rollback + if cur_driver: + try: + bind(cur_driver, full_pci_address) + except Exception: + pass raise ValueError( f"Not able to attach {driver} to {full_pci_address}. Reason: {e}" ) from e
436-446: Docstring mismatch: capability vs enabled stateCode returns True if “MSI-X:” capability exists, irrespective of “Enable+/-”. Docstring says “and if it is currently enabled.”
- Either check for “MSI-X: Enable+” explicitly, or adjust the docstring to “supports MSI-X (capability present)”.
Apply this minimal docstring fix:
def check_msix_capability(full_pci_address): """ - Check whether the PCI device supports Extended Message Signaled Interrupts - and if it is currently enabled. + Check whether the PCI device reports MSI-X capability. @@ - :return: True if supported, False otherwise + :return: True if MSI-X capability is present, False otherwise """
443-445: Input validation before shell=True (S604)Since shell=True is used, validate the PCI address to mitigate injection risk. A strict regex guard is simple and cheap.
Example:
def check_msix_capability(full_pci_address): + if not re.fullmatch(r"[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]", full_pci_address): + raise ValueError(f"Invalid PCI address: {full_pci_address}") out = process.run(f"lspci -vvs {full_pci_address}", ignore_status=True, shell=True) @@ def device_supports_irqs(full_pci_address, count): + if not re.fullmatch(r"[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]", full_pci_address): + raise ValueError(f"Invalid PCI address: {full_pci_address}") out = process.run( f"lspci -vv -s {full_pci_address}", ignore_status=True, shell=True )Also applies to: 457-461
selftests/unit/utils/pci.py (3)
45-66: Add negative-path test for non-EEXIST errorsPlease add a test where write_file_or_fail raises GenIOError with a different errno/message and assert add_vendor_id raises ValueError.
67-77: Add input validation tests for attach_driverConsider tests verifying:
- ValueError when driver or full_pci_address is None/empty.
- Best-effort rollback if bind() fails (if you adopt the rollback suggestion).
78-107: Cover lspci failure pathsAdd tests asserting ValueError is raised when process.run returns a non-zero exit_status for:
- check_msix_capability()
- device_supports_irqs()
avocado/utils/dmesg.py (1)
193-200: Consider refactoring to reduce code duplication.The dmesg and journalctl checks follow identical logic. This duplication could be reduced by extracting the common pattern.
Consider refactoring to a helper function:
def check_kernel_logs(pattern): """ Check if "pattern" is present in kernel logs. :param pattern: string to check in kernel logs :return: True if pattern found, False otherwise """ commands = ["dmesg", "journalctl -k -b"] for cmd in commands: out = process.run(cmd, ignore_status=True, verbose=False, sudo=True) if pattern in out.stdout_text: return True return Falseselftests/unit/utils/dmesg.py (3)
10-19: Add assertion for journalctl call verification.The test verifies the dmesg call but doesn't verify that journalctl is NOT called when the pattern is found in dmesg. This would ensure early return optimization is working correctly.
Apply this diff to improve test coverage:
def test_pattern_found_in_dmesg(self, mock_run): # Simulate `dmesg` output containing the pattern mock_run.return_value.stdout_text = "kernel: test_pattern detected" mock_run.return_value.exit_status = 0 result = dmesg.check_kernel_logs("test_pattern") self.assertTrue(result) mock_run.assert_called_with( "dmesg", ignore_status=True, verbose=False, sudo=True ) + # Verify journalctl was not called since pattern was found in dmesg + self.assertEqual(mock_run.call_count, 1)
21-34: Add assertion for journalctl call verification.The test should verify that journalctl is called with the correct arguments to ensure complete test coverage.
Apply this diff:
def test_pattern_found_in_journalctl(self, mock_run): dmesg_out = mock.Mock() dmesg_out.stdout_text = "kernel: something else" dmesg_out.exit_status = 0 journal_out = mock.Mock() journal_out.stdout_text = "kernel: test_pattern here" journal_out.exit_status = 0 mock_run.side_effect = [dmesg_out, journal_out] result = dmesg.check_kernel_logs("test_pattern") self.assertTrue(result) + # Verify both commands were called + self.assertEqual(mock_run.call_count, 2) + mock_run.assert_any_call( + "dmesg", ignore_status=True, verbose=False, sudo=True + ) + mock_run.assert_any_call( + "journalctl -k -b", ignore_status=True, verbose=False, sudo=True + )
8-53: Expand test coverage for edge cases and error conditions.The test suite is missing coverage for several important scenarios:
- Edge cases: empty pattern, None pattern, non-string pattern
- Error conditions: stdout_text raises TypeError/AttributeError
- Boundary cases: pattern at start/end of output, case sensitivity
Consider adding these test cases:
@mock.patch("avocado.utils.dmesg.process.run") def test_empty_pattern(self, mock_run): """Test behavior with empty pattern.""" with self.assertRaises(ValueError): dmesg.check_kernel_logs("") @mock.patch("avocado.utils.dmesg.process.run") def test_none_pattern(self, mock_run): """Test behavior with None pattern.""" with self.assertRaises(ValueError): dmesg.check_kernel_logs(None) @mock.patch("avocado.utils.dmesg.process.run") def test_stdout_text_error(self, mock_run): """Test handling of stdout_text TypeError.""" mock_run.return_value.stdout_text = mock.PropertyMock( side_effect=TypeError("Unable to decode") ) result = dmesg.check_kernel_logs("test_pattern") self.assertFalse(result) @mock.patch("avocado.utils.dmesg.process.run") def test_pattern_case_sensitivity(self, mock_run): """Test that pattern matching is case-sensitive.""" mock_run.return_value.stdout_text = "kernel: TEST_PATTERN detected" result = dmesg.check_kernel_logs("test_pattern") self.assertFalse(result)Note: These tests assume the implementation includes the error handling and validation suggested in the previous review comments.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
avocado/utils/dmesg.py(1 hunks)avocado/utils/linux_modules.py(1 hunks)avocado/utils/pci.py(1 hunks)selftests/check.py(1 hunks)selftests/unit/utils/dmesg.py(1 hunks)selftests/unit/utils/linux_modules.py(1 hunks)selftests/unit/utils/pci.py(2 hunks)
🧰 Additional context used
🧬 Code graph analysis (5)
avocado/utils/dmesg.py (1)
avocado/utils/process.py (3)
run(926-948)run(952-1019)stdout_text(404-409)
avocado/utils/pci.py (2)
avocado/utils/genio.py (2)
write_file_or_fail(163-177)GenIOError(32-35)avocado/utils/process.py (4)
run(926-948)run(952-1019)stderr_text(412-417)stdout_text(404-409)
selftests/unit/utils/dmesg.py (2)
avocado/utils/process.py (1)
stdout_text(404-409)avocado/utils/dmesg.py (1)
check_kernel_logs(186-202)
selftests/unit/utils/linux_modules.py (1)
avocado/utils/linux_modules.py (2)
ModuleConfig(37-43)configure_module(266-284)
selftests/unit/utils/pci.py (3)
avocado/utils/pci.py (4)
add_vendor_id(379-399)attach_driver(402-432)check_msix_capability(435-446)device_supports_irqs(449-466)avocado/utils/genio.py (1)
GenIOError(32-35)avocado/utils/process.py (1)
stdout_text(404-409)
🪛 Ruff (0.14.0)
avocado/utils/linux_modules.py
282-282: Avoid specifying long messages outside the exception class
(TRY003)
284-284: Avoid specifying long messages outside the exception class
(TRY003)
avocado/utils/pci.py
397-399: Avoid specifying long messages outside the exception class
(TRY003)
416-416: Avoid specifying long messages outside the exception class
(TRY003)
430-432: Avoid specifying long messages outside the exception class
(TRY003)
443-443: Function call with shell=True parameter identified, security issue
(S604)
445-445: Avoid specifying long messages outside the exception class
(TRY003)
457-457: Function call with shell=True parameter identified, security issue
(S604)
461-461: Avoid specifying long messages outside the exception class
(TRY003)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (47)
- GitHub Check: rpm-build:centos-stream-9-x86_64
- GitHub Check: rpm-build:fedora-41-aarch64
- GitHub Check: rpm-build:fedora-41-x86_64
- GitHub Check: rpm-build:fedora-41-s390x
- GitHub Check: rpm-build:fedora-43-x86_64
- GitHub Check: rpm-build:fedora-41-aarch64
- GitHub Check: rpm-build:fedora-rawhide-x86_64
- GitHub Check: rpm-build:fedora-41-x86_64
- GitHub Check: rpm-build:epel-9-x86_64
- GitHub Check: rpm-build:fedora-42-x86_64
- GitHub Check: rpm-build:fedora-41-ppc64le
- GitHub Check: rpm-build:centos-stream-9-x86_64
- GitHub Check: rpm-build:fedora-42-x86_64
- GitHub Check: rpm-build:centos-stream-9-x86_64
- GitHub Check: rpm-build:epel-9-x86_64
- GitHub Check: rpm-build:fedora-41-s390x
- GitHub Check: rpm-build:fedora-rawhide-x86_64
- GitHub Check: rpm-build:fedora-41-ppc64le
- GitHub Check: rpm-build:fedora-41-aarch64
- GitHub Check: rpm-build:fedora-41-x86_64
- GitHub Check: rpm-build:fedora-43-x86_64
- GitHub Check: rpm-build:centos-stream-9-x86_64
- GitHub Check: rpm-build:fedora-43-x86_64
- GitHub Check: rpm-build:fedora-rawhide-x86_64
- GitHub Check: rpm-build:fedora-41-s390x
- GitHub Check: rpm-build:fedora-41-x86_64
- GitHub Check: rpm-build:fedora-42-x86_64
- GitHub Check: rpm-build:epel-9-x86_64
- GitHub Check: rpm-build:fedora-41-ppc64le
- GitHub Check: rpm-build:fedora-41-aarch64
- GitHub Check: Fedora develop install/uninstall task
- GitHub Check: Fedora selftests
- GitHub Check: Egg task ubuntu:24.04
- GitHub Check: Podman spawner with 3rd party runner plugin
- GitHub Check: Egg task ubuntu:22.04
- GitHub Check: Egg task fedora:40
- GitHub Check: macOS with Python 3.11
- GitHub Check: Version task ubuntu:22.04
- GitHub Check: Build Package (wheel/tarball) for Python 3.13
- GitHub Check: Windows with Python 3.11
- GitHub Check: Windows with Python 3.9
- GitHub Check: Version task debian:12.4
- GitHub Check: Smokecheck on Linux with Python 3.11
- GitHub Check: Smokecheck on Linux with Python 3.10
- GitHub Check: Version task ubi:8.8
- GitHub Check: Static checks
- GitHub Check: Code Coverage (3.11)
🔇 Additional comments (3)
selftests/check.py (1)
30-30: Update unit count to 889: request confirmationPlease confirm this matches the actual discovered tests on CI for selftests/unit (including newly added dmesg, pci, linux_modules tests). If any unit tests get skipped/gated by environment, this number may need adjustment.
avocado/utils/linux_modules.py (1)
266-285: configure_module(): Logic looks good and matches testsBehavior across NOT_SET/BUILTIN/MODULE is correct and uses existing helpers cleanly.
selftests/unit/utils/linux_modules.py (1)
66-107: configure_module tests cover key scenariosGood coverage for NOT_SET, BUILTIN, MODULE success and failure; also assert no unnecessary calls in BUILTIN case.
| def check_kernel_logs(pattern): | ||
| """ | ||
| Check if "pattern" is present in kernel logs. | ||
| :param pattern: string to check in kernel logs | ||
| :return: True if pattern found, False otherwise | ||
| """ | ||
| out = process.run("dmesg", ignore_status=True, verbose=False, sudo=True) | ||
| if pattern in out.stdout_text: | ||
| return True | ||
|
|
||
| cmd = "journalctl -k -b" | ||
| out = process.run(cmd, ignore_status=True, verbose=False, sudo=True) | ||
| if pattern in out.stdout_text: | ||
| return True | ||
|
|
||
| return False |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Enhance error handling and add input validation.
The function lacks error handling for potential exceptions and input validation:
- No validation for
patternparameter (could be None, empty string, or non-string type). - The
stdout_textproperty can raiseTypeErrorif unable to decode (seeavocado/utils/process.pylines 403-408), but this exception is not handled. - Simple substring matching with
inoperator may produce false positives for partial matches.
Consider applying this diff to add input validation and improve robustness:
def check_kernel_logs(pattern):
"""
Check if "pattern" is present in kernel logs.
:param pattern: string to check in kernel logs
+ :type pattern: str
:return: True if pattern found, False otherwise
+ :rtype: bool
+ :raises ValueError: if pattern is not a valid string
"""
+ if not pattern or not isinstance(pattern, str):
+ raise ValueError("pattern must be a non-empty string")
+
+ try:
- out = process.run("dmesg", ignore_status=True, verbose=False, sudo=True)
- if pattern in out.stdout_text:
- return True
-
- cmd = "journalctl -k -b"
- out = process.run(cmd, ignore_status=True, verbose=False, sudo=True)
- if pattern in out.stdout_text:
- return True
-
- return False
+ out = process.run("dmesg", ignore_status=True, verbose=False, sudo=True)
+ if pattern in out.stdout_text:
+ return True
+
+ cmd = "journalctl -k -b"
+ out = process.run(cmd, ignore_status=True, verbose=False, sudo=True)
+ if pattern in out.stdout_text:
+ return True
+
+ return False
+ except (TypeError, AttributeError) as e:
+ LOGGER.error("Error accessing kernel logs: %s", e)
+ return FalseCommittable suggestion skipped: line range outside the PR's diff.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #6225 +/- ##
==========================================
+ Coverage 73.46% 73.58% +0.11%
==========================================
Files 206 206
Lines 22487 22539 +52
==========================================
+ Hits 16521 16585 +64
+ Misses 5966 5954 -12 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
…ability checks This patch introduces a set of PCI utility functions to simplify device management and capability checks: * add_vendor_id() – retrieves the vendor ID of a PCI device and associate it with the specified driver * attach_driver() – unbinds a device from its current driver and rebinds it to a given driver * check_msix_capability() – verifies whether the device supports and has MSI-X capability enabled * device_supports_irqs() – checks if the device supports at least a specified number of IRQs These helpers provide a cleaner and reusable interface for driver binding and interrupt capability validation. Signed-off-by: Dheeraj Kumar Srivastava <dheerajkumar.srivastava@amd.com>
There was a problem hiding this comment.
Actionable comments posted: 4
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
avocado/utils/pci.py(1 hunks)selftests/check.py(1 hunks)selftests/unit/utils/pci.py(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- selftests/check.py
- selftests/unit/utils/pci.py
🧰 Additional context used
🧬 Code graph analysis (1)
avocado/utils/pci.py (2)
avocado/utils/genio.py (2)
write_file_or_fail(163-177)GenIOError(32-35)avocado/utils/process.py (4)
run(926-948)run(952-1019)stderr_text(412-417)stdout_text(404-409)
🪛 Ruff (0.14.0)
avocado/utils/pci.py
397-399: Avoid specifying long messages outside the exception class
(TRY003)
416-416: Avoid specifying long messages outside the exception class
(TRY003)
430-432: Avoid specifying long messages outside the exception class
(TRY003)
443-443: Function call with shell=True parameter identified, security issue
(S604)
445-445: Avoid specifying long messages outside the exception class
(TRY003)
457-457: Function call with shell=True parameter identified, security issue
(S604)
461-461: Avoid specifying long messages outside the exception class
(TRY003)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: rpm-build:centos-stream-9-x86_64
- GitHub Check: rpm-build:fedora-43-x86_64
- GitHub Check: rpm-build:fedora-rawhide-x86_64
- GitHub Check: rpm-build:fedora-41-s390x
- GitHub Check: rpm-build:fedora-41-x86_64
- GitHub Check: rpm-build:fedora-42-x86_64
- GitHub Check: rpm-build:epel-9-x86_64
- GitHub Check: rpm-build:fedora-41-ppc64le
- GitHub Check: rpm-build:fedora-41-aarch64
- GitHub Check: Fedora selftests
- GitHub Check: Egg task ubuntu:24.04
- GitHub Check: Egg task ubuntu:22.04
- GitHub Check: Podman spawner with 3rd party runner plugin
- GitHub Check: Version task debian:12.4
- GitHub Check: Build Package (wheel/tarball) for Python 3.13
- GitHub Check: Build Package (wheel/tarball) for Python 3.11
- GitHub Check: Version task ubi:8.8
- GitHub Check: Version task ubuntu:22.04
- GitHub Check: Build Package (wheel/tarball) for Python 3.12
- GitHub Check: Build Package (wheel/tarball) for Python 3.10
- GitHub Check: Smokecheck on Linux with Python 3.10
- GitHub Check: Smokecheck on Linux with Python 3.11
- GitHub Check: macOS with Python 3.11
- GitHub Check: Static checks
- GitHub Check: Code Coverage (3.11)
| def add_vendor_id(full_pci_address, driver): | ||
| """ | ||
| Retrieve and add the vendor ID of a PCI device to the specified driver. | ||
| :param full_pci_address: Full PCI device address, including domain | ||
| (e.g., 0000:03:00.0). | ||
| :param driver: Driver to associate with the vendor ID of the PCI device. | ||
| """ | ||
| # Get vendor id of pci device | ||
| output = get_vendor_id(full_pci_address) | ||
| vid = output.replace(":", " ") | ||
|
|
||
| # Add device vendor id to the driver | ||
| try: | ||
| genio.write_file_or_fail(f"/sys/bus/pci/drivers/{driver}/new_id", f"{vid}\n") | ||
| except genio.GenIOError as e: | ||
| if "File exists" in str(e): | ||
| return | ||
| raise ValueError( | ||
| f"Failed to add vendor ID '{vid}' to driver '{driver}': {e}" | ||
| ) from e | ||
|
|
There was a problem hiding this comment.
Add input validation and document idempotent behavior.
The function lacks input validation for driver and full_pci_address parameters, which could lead to confusing errors downstream. Additionally, the idempotent behavior (silently returning when the vendor ID already exists) is not documented in the docstring.
Consider applying these improvements:
- Add input validation:
def add_vendor_id(full_pci_address, driver):
"""
Retrieve and add the vendor ID of a PCI device to the specified driver.
:param full_pci_address: Full PCI device address, including domain
(e.g., 0000:03:00.0).
:param driver: Driver to associate with the vendor ID of the PCI device.
+ :raises ValueError: If inputs are invalid or if adding the vendor ID fails.
+ :note: This function is idempotent. If the vendor ID is already associated
+ with the driver, the function returns successfully without error.
"""
+ if not driver or not full_pci_address:
+ raise ValueError("'driver' and 'full_pci_address' must not be None or empty")
+
# Get vendor id of pci device
output = get_vendor_id(full_pci_address)- The "File exists" check on line 395 is fragile as it relies on string matching in the exception message. Consider checking for a more specific exception type or error code if available.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def add_vendor_id(full_pci_address, driver): | |
| """ | |
| Retrieve and add the vendor ID of a PCI device to the specified driver. | |
| :param full_pci_address: Full PCI device address, including domain | |
| (e.g., 0000:03:00.0). | |
| :param driver: Driver to associate with the vendor ID of the PCI device. | |
| """ | |
| # Get vendor id of pci device | |
| output = get_vendor_id(full_pci_address) | |
| vid = output.replace(":", " ") | |
| # Add device vendor id to the driver | |
| try: | |
| genio.write_file_or_fail(f"/sys/bus/pci/drivers/{driver}/new_id", f"{vid}\n") | |
| except genio.GenIOError as e: | |
| if "File exists" in str(e): | |
| return | |
| raise ValueError( | |
| f"Failed to add vendor ID '{vid}' to driver '{driver}': {e}" | |
| ) from e | |
| def add_vendor_id(full_pci_address, driver): | |
| """ | |
| Retrieve and add the vendor ID of a PCI device to the specified driver. | |
| :param full_pci_address: Full PCI device address, including domain | |
| (e.g., 0000:03:00.0). | |
| :param driver: Driver to associate with the vendor ID of the PCI device. | |
| :raises ValueError: If inputs are invalid or if adding the vendor ID fails. | |
| :note: This function is idempotent. If the vendor ID is already associated | |
| with the driver, the function returns successfully without error. | |
| """ | |
| if not driver or not full_pci_address: | |
| raise ValueError("'driver' and 'full_pci_address' must not be None or empty") | |
| # Get vendor id of pci device | |
| output = get_vendor_id(full_pci_address) | |
| vid = output.replace(":", " ") | |
| # Add device vendor id to the driver | |
| try: | |
| genio.write_file_or_fail(f"/sys/bus/pci/drivers/{driver}/new_id", f"{vid}\n") | |
| except genio.GenIOError as e: | |
| if "File exists" in str(e): | |
| return | |
| raise ValueError( | |
| f"Failed to add vendor ID '{vid}' to driver '{driver}': {e}" | |
| ) from e |
🧰 Tools
🪛 Ruff (0.14.0)
397-399: Avoid specifying long messages outside the exception class
(TRY003)
🤖 Prompt for AI Agents
In avocado/utils/pci.py around lines 379 to 400, validate inputs and document
idempotence: ensure driver and full_pci_address are non-empty strings and
validate full_pci_address matches expected PCI BDF format (e.g.,
domain:bus:slot.func) before calling get_vendor_id; update the docstring to
mention the function is idempotent and will silently return if the vendor ID is
already present; replace the fragile string match on GenIOError with a robust
check by inspecting the underlying exception (e.g., check for an errno attribute
equal to EEXIST or the wrapped OSError/IOError indicating “File exists”), and
only treat that specific case as a no-op while re-raising other errors with a
clear ValueError including the original exception.
| def attach_driver(full_pci_address, driver): | ||
| """ | ||
| Unbind the device from its existing driver and bind it to the given driver. | ||
| :param full_pci_address: Full PCI device address (e.g., 0000:03:00.0) | ||
| :param driver: Driver to be attached to the specified PCI device | ||
| :raises ValueError: If `driver` or `full_pci_address` is None, or if the driver | ||
| could not be attached successfully due to an OS error. | ||
| :warning: This function may unbind the device from its current driver, which can | ||
| temporarily make the device unavailable until it is reattached. | ||
| """ | ||
|
|
||
| try: | ||
| if not driver or not full_pci_address: | ||
| raise ValueError("'driver' or 'full_pci_address' inputs are None") | ||
|
|
||
| # add vendor id of device to driver | ||
| add_vendor_id(full_pci_address, driver) | ||
|
|
||
| # unbind the device from its initial driver | ||
| cur_driver = get_driver(full_pci_address) | ||
| if cur_driver is not None: | ||
| unbind(cur_driver, full_pci_address) | ||
|
|
||
| # Bind device to driver | ||
| bind(driver, full_pci_address) | ||
|
|
||
| except OSError as e: | ||
| raise ValueError( | ||
| f"Not able to attach {driver} to {full_pci_address}. Reason: {e}" | ||
| ) from e | ||
|
|
There was a problem hiding this comment.
Fix incomplete exception handling.
The function only catches OSError (line 429), but the called functions raise different exception types:
add_vendor_idraisesValueErrorand can propagateGenIOErrorunbindandbindraiseValueErrorandGenIOError(fromwrite_file_or_fail)
This means most exceptions will propagate uncaught, bypassing the intended error handling and the descriptive error message.
Apply this diff to fix the exception handling:
- except OSError as e:
+ except (OSError, ValueError, genio.GenIOError) as e:
raise ValueError(
f"Not able to attach {driver} to {full_pci_address}. Reason: {e}"
) from eAdditionally, the error message on line 416 says "inputs are None" but the check if not driver or not full_pci_address also catches empty strings. Update the message for accuracy:
if not driver or not full_pci_address:
- raise ValueError("'driver' or 'full_pci_address' inputs are None")
+ raise ValueError("'driver' and 'full_pci_address' must not be None or empty")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def attach_driver(full_pci_address, driver): | |
| """ | |
| Unbind the device from its existing driver and bind it to the given driver. | |
| :param full_pci_address: Full PCI device address (e.g., 0000:03:00.0) | |
| :param driver: Driver to be attached to the specified PCI device | |
| :raises ValueError: If `driver` or `full_pci_address` is None, or if the driver | |
| could not be attached successfully due to an OS error. | |
| :warning: This function may unbind the device from its current driver, which can | |
| temporarily make the device unavailable until it is reattached. | |
| """ | |
| try: | |
| if not driver or not full_pci_address: | |
| raise ValueError("'driver' or 'full_pci_address' inputs are None") | |
| # add vendor id of device to driver | |
| add_vendor_id(full_pci_address, driver) | |
| # unbind the device from its initial driver | |
| cur_driver = get_driver(full_pci_address) | |
| if cur_driver is not None: | |
| unbind(cur_driver, full_pci_address) | |
| # Bind device to driver | |
| bind(driver, full_pci_address) | |
| except OSError as e: | |
| raise ValueError( | |
| f"Not able to attach {driver} to {full_pci_address}. Reason: {e}" | |
| ) from e | |
| def attach_driver(full_pci_address, driver): | |
| """ | |
| Unbind the device from its existing driver and bind it to the given driver. | |
| :param full_pci_address: Full PCI device address (e.g., 0000:03:00.0) | |
| :param driver: Driver to be attached to the specified PCI device | |
| :raises ValueError: If `driver` or `full_pci_address` is None or empty, or if the driver | |
| could not be attached successfully due to an OS error. | |
| :warning: This function may unbind the device from its current driver, which can | |
| temporarily make the device unavailable until it is reattached. | |
| """ | |
| try: | |
| if not driver or not full_pci_address: | |
| raise ValueError("'driver' and 'full_pci_address' must not be None or empty") | |
| # add vendor id of device to driver | |
| add_vendor_id(full_pci_address, driver) | |
| # unbind the device from its initial driver | |
| cur_driver = get_driver(full_pci_address) | |
| if cur_driver is not None: | |
| unbind(cur_driver, full_pci_address) | |
| # Bind device to driver | |
| bind(driver, full_pci_address) | |
| except (OSError, ValueError, genio.GenIOError) as e: | |
| raise ValueError( | |
| f"Not able to attach {driver} to {full_pci_address}. Reason: {e}" | |
| ) from e |
🧰 Tools
🪛 Ruff (0.14.0)
416-416: Avoid specifying long messages outside the exception class
(TRY003)
430-432: Avoid specifying long messages outside the exception class
(TRY003)
🤖 Prompt for AI Agents
In avocado/utils/pci.py around lines 402 to 433, the try/except only catches
OSError but called functions can raise ValueError and GenIOError; update the
except to catch OSError, ValueError, and GenIOError (e.g., except (OSError,
ValueError, GenIOError) as e) and re-raise a ValueError with the same
descriptive message and "from e" to preserve traceback; also change the initial
input-check error text to reflect both None and empty values (e.g., "'driver' or
'full_pci_address' is missing or empty") so the message accurately describes the
condition.
| def check_msix_capability(full_pci_address): | ||
| """ | ||
| Check whether the PCI device supports Extended Message Signaled Interrupts | ||
| and if it is currently enabled. | ||
| :param full_pci_address: Full PCI address including domain (0000:03:00.0) | ||
| :return: True if supported, False otherwise | ||
| """ | ||
| out = process.run(f"lspci -vvs {full_pci_address}", ignore_status=True, shell=True) | ||
| if out.exit_status: | ||
| raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}") | ||
| return any("MSI-X:" in line for line in out.stdout_text.splitlines()) | ||
|
|
There was a problem hiding this comment.
Docstring is misleading about enabled state verification.
The docstring states the function checks "if it is currently enabled" (line 438), but the implementation only verifies the presence of MSI-X capability, not whether it's enabled. The lspci output contains "MSI-X: Enable+" (enabled) or "MSI-X: Enable-" (disabled), but line 446 searches for any line containing "MSI-X:", matching both states.
If the function should check whether MSI-X is enabled (as the docstring claims), apply this diff:
def check_msix_capability(full_pci_address):
"""
Check whether the PCI device supports Extended Message Signaled Interrupts
and if it is currently enabled.
:param full_pci_address: Full PCI address including domain (0000:03:00.0)
- :return: True if supported, False otherwise
+ :return: True if MSI-X is supported and enabled, False otherwise
"""
+ if not full_pci_address:
+ raise ValueError("'full_pci_address' must not be None or empty")
+
out = process.run(f"lspci -vvs {full_pci_address}", ignore_status=True, shell=True)
if out.exit_status:
raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}")
- return any("MSI-X:" in line for line in out.stdout_text.splitlines())
+ return any("MSI-X: Enable+" in line for line in out.stdout_text.splitlines())Or, if the function should only check for MSI-X capability support (not enabled state), update the docstring:
"""
- Check whether the PCI device supports Extended Message Signaled Interrupts
- and if it is currently enabled.
+ Check whether the PCI device supports Extended Message Signaled Interrupts (MSI-X).Note: Input validation has been added above to mitigate shell injection risk (S604) since full_pci_address is passed to a shell command.
As per static analysis hints
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def check_msix_capability(full_pci_address): | |
| """ | |
| Check whether the PCI device supports Extended Message Signaled Interrupts | |
| and if it is currently enabled. | |
| :param full_pci_address: Full PCI address including domain (0000:03:00.0) | |
| :return: True if supported, False otherwise | |
| """ | |
| out = process.run(f"lspci -vvs {full_pci_address}", ignore_status=True, shell=True) | |
| if out.exit_status: | |
| raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}") | |
| return any("MSI-X:" in line for line in out.stdout_text.splitlines()) | |
| def check_msix_capability(full_pci_address): | |
| """ | |
| Check whether the PCI device supports Extended Message Signaled Interrupts | |
| and if it is currently enabled. | |
| :param full_pci_address: Full PCI address including domain (0000:03:00.0) | |
| :return: True if MSI-X is supported and enabled, False otherwise | |
| """ | |
| if not full_pci_address: | |
| raise ValueError("'full_pci_address' must not be None or empty") | |
| out = process.run( | |
| f"lspci -vvs {full_pci_address}", | |
| ignore_status=True, | |
| shell=True | |
| ) | |
| if out.exit_status: | |
| raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}") | |
| return any("MSI-X: Enable+" in line for line in out.stdout_text.splitlines()) |
🧰 Tools
🪛 Ruff (0.14.0)
443-443: Function call with shell=True parameter identified, security issue
(S604)
445-445: Avoid specifying long messages outside the exception class
(TRY003)
🤖 Prompt for AI Agents
In avocado/utils/pci.py around lines 435 to 447, the docstring claims we check
whether MSI-X is "currently enabled" but the code only looks for any "MSI-X:"
token; update the implementation to match the docstring by verifying the enabled
state (search for "MSI-X: Enable+" in lspci output) and add input validation for
full_pci_address (e.g., assert it matches a PCI address regex like
r'^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]$' or similar) or avoid
shell=True by passing the command as a list to process.run to mitigate shell
injection, so the function both accurately detects enabled MSI-X and is safe
from injection.
| def device_supports_irqs(full_pci_address, count): | ||
| """ | ||
| Check if the device supports at least the specified number of interrupts. | ||
| :param full_pci_address: Full PCI device address including domain (e.g., 0000:03:00.0) | ||
| :param count: Number of IRQs the device should support | ||
| :return: True if supported, False otherwise | ||
| """ | ||
| out = process.run( | ||
| f"lspci -vv -s {full_pci_address}", ignore_status=True, shell=True | ||
| ) | ||
| if out.exit_status: | ||
| raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}") | ||
| match = re.search(r"MSI-X:.*?Count=(\d+)", out.stdout_text, re.S) | ||
| if match: | ||
| nirq = int(match.group(1)) | ||
| return nirq >= count | ||
| return False | ||
|
|
There was a problem hiding this comment.
Add input validation to mitigate injection risk.
The function passes full_pci_address to a shell command without validation, creating a potential injection vector. Additionally, the count parameter is not validated.
Apply this diff to add input validation:
def device_supports_irqs(full_pci_address, count):
"""
Check if the device supports at least the specified number of interrupts.
:param full_pci_address: Full PCI device address including domain (e.g., 0000:03:00.0)
:param count: Number of IRQs the device should support
:return: True if supported, False otherwise
+ :note: This function checks MSI-X interrupt count. Devices without MSI-X will return False.
"""
+ if not full_pci_address:
+ raise ValueError("'full_pci_address' must not be None or empty")
+ if not isinstance(count, int) or count < 0:
+ raise ValueError("'count' must be a non-negative integer")
+
out = process.run(
f"lspci -vv -s {full_pci_address}", ignore_status=True, shell=True
)As per static analysis hints
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def device_supports_irqs(full_pci_address, count): | |
| """ | |
| Check if the device supports at least the specified number of interrupts. | |
| :param full_pci_address: Full PCI device address including domain (e.g., 0000:03:00.0) | |
| :param count: Number of IRQs the device should support | |
| :return: True if supported, False otherwise | |
| """ | |
| out = process.run( | |
| f"lspci -vv -s {full_pci_address}", ignore_status=True, shell=True | |
| ) | |
| if out.exit_status: | |
| raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}") | |
| match = re.search(r"MSI-X:.*?Count=(\d+)", out.stdout_text, re.S) | |
| if match: | |
| nirq = int(match.group(1)) | |
| return nirq >= count | |
| return False | |
| def device_supports_irqs(full_pci_address, count): | |
| """ | |
| Check if the device supports at least the specified number of interrupts. | |
| :param full_pci_address: Full PCI device address including domain (e.g., 0000:03:00.0) | |
| :param count: Number of IRQs the device should support | |
| :return: True if supported, False otherwise | |
| :note: This function checks MSI-X interrupt count. Devices without MSI-X will return False. | |
| """ | |
| if not full_pci_address: | |
| raise ValueError("'full_pci_address' must not be None or empty") | |
| if not isinstance(count, int) or count < 0: | |
| raise ValueError("'count' must be a non-negative integer") | |
| out = process.run( | |
| f"lspci -vv -s {full_pci_address}", ignore_status=True, shell=True | |
| ) | |
| if out.exit_status: | |
| raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}") | |
| match = re.search(r"MSI-X:.*?Count=(\d+)", out.stdout_text, re.S) | |
| if match: | |
| nirq = int(match.group(1)) | |
| return nirq >= count | |
| return False |
🧰 Tools
🪛 Ruff (0.14.0)
457-457: Function call with shell=True parameter identified, security issue
(S604)
461-461: Avoid specifying long messages outside the exception class
(TRY003)
🤖 Prompt for AI Agents
In avocado/utils/pci.py around lines 449 to 467, the function
device_supports_irqs accepts unvalidated full_pci_address and count and
interpolates the address into a shell command, creating an injection risk and
allowing invalid counts; validate inputs and avoid shell interpolation: ensure
full_pci_address matches the expected PCI address pattern (domain:bus:slot.func,
e.g. r'^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]$') and raise
ValueError on mismatch, ensure count is an integer >= 0 (coerce or raise if
not), and call lspci without shell interpolation (pass arguments as a list to
process.run or use shlex.quote if constrained) so the address is not evaluated
by a shell; keep existing error handling for command failures and unchanged
parsing logic.
Adds functionality to test interrupt remapping support for a PCI device using VFIO ioctls. - Verifies if the PCI device supports "count" MSI-X interrupts. - Binds the input PCI device to the vfio-pci driver. - Attempts to allocate "count" number of IRQs with IOMMU Interrupt remapping enabled; the test fails if allocation is unsuccessful. This patch is depended on 1. avocado-framework/avocado#6225 2. avocado-framework/avocado#6226 Signed-off-by: Dheeraj Kumar Srivastava <dheerajkumar.srivastava@amd.com>
1 participant
Add utilities to
Summary by CodeRabbit
New Features
Tests