starlingx/test
Code Issues Proposed changes

ResourceFinder and CollectionPlugin fixes

Browse Source 
Fixing issues with the ResourceFinder and the CollectionPlugin
not finding files if test-case-level pytest paths were getting
used.

Change-Id: I317749a0d5f5908105c525f4f33ee786b0ff1a1b
Signed-off-by: croy <Christian.Roy@windriver.com>
This commit is contained in:
croy
2025-08-19 14:19:01 -04:00
parent fb9262dab8
commit e7b0c1b554
6 changed files with 161 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
framework/pytest_plugins/collection_plugin.py
View File

@@ -1,54 +1,84 @@
import os
from pathlib import Path
from typing import Any
from framework.database.objects.testcase import TestCase
class CollectionPlugin:
"""
Plugin that allows us to get all tests after running a collect only in pytest
Plugin that allows us to get all tests after running a collect only in pytest.
"""
PRIORITIES = ['p0', 'p1', 'p2', 'p3']
PRIORITIES = ["p0", "p1", "p2", "p3"]
def __init__(self):
def __init__(self, repo_root: str):
"""
Constructor.
Args:
repo_root (str): The Absolute path to the root of the repo.
"""
self.repo_root: str = repo_root
self.tests: [TestCase] = []
def pytest_report_collectionfinish(self, items):
def _get_full_nodeid(self, test: Any) -> str:
"""
Run after collection is finished
Ensures the nodeid is relative to the repository root.
Args:
items (): list of test items
test (Any): The pytest test item.
Returns:
str: Full nodeid relative to repository root.
"""
# Get the absolute path of the test file
abs_path = Path(test.path).absolute()
repo_root_path = Path(self.repo_root)
# Make it relative to repo root
rel_path = abs_path.relative_to(repo_root_path).as_posix()
# Replace the file path portion of nodeid with the repository-relative path
parts = test.nodeid.split("::")
return "::".join([rel_path] + parts[1:])
def pytest_report_collectionfinish(self, items: Any):
"""
Run after collection is finished.
Args:
items (Any): list of Pytest test items.
"""
for test in items:
markers = list(map(lambda marker: marker.name, test.own_markers))
priority = self.get_testcase_priority(markers)
if priority:
markers.remove(priority)
testcase = TestCase(test.name, os.path.basename(test.location[0]), priority, test.location[0], test.nodeid)
full_node_id = self._get_full_nodeid(test)
testcase = TestCase(test.name, os.path.basename(test.location[0]), priority, test.location[0], full_node_id)
testcase.set_markers(markers)
self.tests.append(testcase)
def get_tests(self) -> [TestCase]:
def get_tests(self) -> list[TestCase]:
"""
Returns the tests
Returns:
Returns the tests.
Returns:
list[TestCase]: List of test cases collected during pytest collection.
"""
return self.tests
def get_testcase_priority(self, markers):
def get_testcase_priority(self, markers: Any) -> str:
"""
Gets the testcase priority from the list of markers
Gets the testcase priority from the list of markers.
Args:
markers: the markers to find the priority from
Returns: testcase priority
markers (Any): the pytest markers to find the priority from.
Returns:
str: Testcase priority.
"""
for mark in markers:
if mark in self.PRIORITIES:

32
framework/resources/resource_finder.py
View File

@@ -4,28 +4,46 @@ import os.path
from pathlib import Path
def get_stx_repo_root() -> str:
"""
Find the full path to the repository root.
Uses the position of the current file relative to the root of the repo.
Returns:
str: The absolute path to the repository root.
Example:
>>> get_repo_root()
will return /home/user/repo/starlingx
"""
path_of_this_file = Path(__file__)
# This file is in framework/resources/resource_finder.py, so go up 3 levels
root_folder = path_of_this_file.parent.parent.parent
return root_folder.as_posix()
def get_stx_resource_path(relative_path: str) -> str:
"""
This function will get the full path to the resource from the relative_path provided.
This will allow projects that use StarlingX as a submodule to still find resource files using the relative path.
Args:
relative_path: The relative path to the resource.
relative_path (str): The relative path to the resource.
Returns: The full path to the resource
Returns:
str: The full path to the resource.
Example:
>>> get_resource_path("framework/resources/resource_finder.py")
will return /home/user/repo/starlingx/framework/resources/resource_finder.py
"""
# check to see if the path really is relative, if not just return the path
if os.path.isabs(relative_path):
return relative_path
path_of_this_file = Path(__file__)
root_folder_of_stx = path_of_this_file.parent.parent.parent
root_folder_of_stx = get_stx_repo_root()
path_to_resource = Path(root_folder_of_stx, relative_path).as_posix()
return path_to_resource

67
framework/runner/objects/test_capability_matcher.py
View File

@@ -1,34 +1,56 @@
import pytest
from config.lab.objects.lab_config import LabConfig
from framework.database.objects.testcase import TestCase
from framework.database.operations.lab_capability_operation import LabCapabilityOperation
from framework.database.operations.lab_operation import LabOperation
from framework.database.operations.run_content_operation import RunContentOperation
from framework.pytest_plugins.collection_plugin import CollectionPlugin
from framework.resources.resource_finder import get_stx_repo_root
class TestCapabilityMatcher:
"""
Class to hold matches for a set of tests given a lab config
Class to hold matches for a set of tests given a lab config.
"""
priority_marker_list = ['p0', 'p1', 'p2', 'p3']
priority_marker_list = ["p0", "p1", "p2", "p3"]
def __init__(self, lab_config: LabConfig):
"""
Constructor
Args:
lab_config (LabConfig): Config for the lab
"""
self.lab_config = lab_config
def get_list_of_tests(self, test_case_folder: str) -> []:
def get_list_of_tests(self, test_case_folder: str) -> list[TestCase]:
"""
Getter for the list of tests that this lab can run
Returns: the list of tests
Getter for the list of tests that this lab can run.
Args:
test_case_folder (str): Path to the folder containing test cases.
Returns:
list[TestCase]: List of tests that can be run.
"""
tests = self._get_all_tests_in_folder(test_case_folder)
capabilities = self.lab_config.get_lab_capabilities()
return self._filter_tests(tests, capabilities)
def get_list_of_tests_from_db(self, run_id: int) -> []:
def get_list_of_tests_from_db(self, run_id: int) -> list[TestCase]:
"""
This function will return the list of test cases matching the run_id from the database.
Args:
run_id (int): Run Id
Returns:
list[TestCase]:
"""
run_content_operation = RunContentOperation()
tests = run_content_operation.get_tests_from_run_content(run_id)
@@ -40,15 +62,16 @@ class TestCapabilityMatcher:
return self._filter_tests(tests, capabilities)
def _filter_tests(self, tests: [TestCase], capabilities: [str]) -> [TestCase]:
def _filter_tests(self, tests: list[TestCase], capabilities: list[str]) -> list[TestCase]:
"""
Filters out the tests that can run on the given lab
Filters out the tests that can run on the given lab.
Args:
tests (): the list of tests
capabilities (): the capabilities
tests (list[TestCase]): The list of tests.
capabilities (list[str]): The capabilities.
Returns:
list[TestCase]: List of tests that can run on the lab based on capabilities.
"""
tests_to_run = []
for test in tests:
@@ -57,24 +80,30 @@ class TestCapabilityMatcher:
tests_to_run.append(test)
return tests_to_run
def _get_all_tests_in_folder(self, test_case_folder: str) -> [TestCase]:
def _get_all_tests_in_folder(self, test_case_folder: str) -> list[TestCase]:
"""
Gerts all tests in the testcase folder
Returns:
Gets all tests in the testcase folder.
Args:
test_case_folder (str): Path to the folder containing test cases.
Returns:
list[TestCase]: List of test cases found in the folder.
"""
collection_plugin = CollectionPlugin()
repo_root = get_stx_repo_root()
collection_plugin = CollectionPlugin(repo_root)
pytest.main(["--collect-only", test_case_folder], plugins=[collection_plugin])
return collection_plugin.get_tests()
def _get_markers(self, test: TestCase):
def _get_markers(self, test: TestCase) -> list[str]:
"""
Gets the markers for the given test
Gets the markers for the given test.
Args:
test (): the test
test (TestCase): The test case to get markers from.
Returns:
list[str]: List of markers associated with the test.
"""
markers = []
for marker in test.get_markers():

17
framework/runner/scripts/test_executor.py
View File

@@ -1,12 +1,9 @@
import os
from optparse import OptionParser
from typing import Optional
import pytest
from config.configuration_file_locations_manager import (
ConfigurationFileLocationsManager,
)
from config.configuration_file_locations_manager import ConfigurationFileLocationsManager
from config.configuration_manager import ConfigurationManager
from framework.database.objects.testcase import TestCase
from framework.database.operations.run_content_operation import RunContentOperation
@@ -14,7 +11,6 @@ from framework.database.operations.run_operation import RunOperation
from framework.database.operations.test_plan_operation import TestPlanOperation
from framework.logging.automation_logger import get_logger
from framework.pytest_plugins.result_collector import ResultCollector
from framework.resources.resource_finder import get_stx_resource_path
from framework.runner.objects.test_capability_matcher import TestCapabilityMatcher
from framework.runner.objects.test_executor_summary import TestExecutorSummary
from testcases.conftest import log_configuration
@@ -32,16 +28,7 @@ def execute_test(test: TestCase, test_executor_summary: TestExecutorSummary, tes
"""
result_collector = ResultCollector(test_executor_summary, test, test_case_result_id)
pytest_args = ConfigurationManager.get_config_pytest_args()
node_id = test.get_pytest_node_id().lstrip("/") # Normalize node_id
# Ensure we do not prepend "testcases/" for unit tests
if node_id.startswith("unit_tests/"):
resolved_path = get_stx_resource_path(node_id)
else:
resolved_path = get_stx_resource_path(os.path.join("testcases", node_id))
pytest_args.append(resolved_path)
pytest_args.append(test.get_pytest_node_id())
pytest.main(pytest_args, plugins=[result_collector])

50
framework/scanning/objects/test_scanner_uploader.py
View File

@@ -1,6 +1,7 @@
from typing import List
import pytest
from framework.database.objects.testcase import TestCase
from framework.database.operations.capability_operation import CapabilityOperation
from framework.database.operations.test_capability_operation import TestCapabilityOperation
@@ -16,15 +17,17 @@ class TestScannerUploader:
def __init__(self, test_folders: List[str]):
self.test_folders = test_folders
def scan_and_upload_tests(self):
def scan_and_upload_tests(self, repo_root: str):
"""
Scan code base and upload/update tests
Returns:
Scans the repo and uploads the new tests to the database.
Args:
repo_root (str): The full path to the root of the repo.
"""
test_info_operation = TestInfoOperation()
scanned_tests = self.scan_for_tests()
scanned_tests = self.scan_for_tests(repo_root)
# Filter to find only the test cases in the desired folders.
filtered_test_cases = []
@@ -47,22 +50,28 @@ class TestScannerUploader:
self.update_pytest_node_id(test, database_testcase)
self.update_capability(test, database_testcase.get_test_info_id())
def scan_for_tests(self) -> [TestCase]:
def scan_for_tests(self, repo_root: str) -> [TestCase]:
"""
Scan for tests
Returns: list of Testcases
Args:
repo_root (str): The full path to the root of the repo.
Returns:
[TestCase]: list of Testcases
"""
collection_plugin = CollectionPlugin()
collection_plugin = CollectionPlugin(repo_root)
pytest.main(["--collect-only"], plugins=[collection_plugin])
return collection_plugin.get_tests()
def update_priority(self, test: TestCase, database_testcase: TestCase):
"""
Checks the current priority of the test, if it's changed, update it
Args:
test: the Test in the Repo Scan
database_testcase: the Test in the Database
test (TestCase): the Test in the Repo Scan
database_testcase (TestCase): the Test in the Database
"""
database_priority = database_testcase.get_priority()
@@ -74,9 +83,10 @@ class TestScannerUploader:
def update_test_path(self, test: TestCase, database_testcase: TestCase):
"""
Checks the current test_path of the test, if it's changed, update it
Args:
test: the Test in the Repo Scan
database_testcase: the Test in the Database
test (TestCase): the Test in the Repo Scan
database_testcase (TestCase): the Test in the Database
"""
database_test_path = database_testcase.get_test_path()
actual_test_path = test.get_test_path().replace("\\", "/")
@@ -88,9 +98,10 @@ class TestScannerUploader:
def update_pytest_node_id(self, test: TestCase, database_testcase: TestCase):
"""
Checks the current pytest_node_id of the test, if it's changed, update it
Args:
test: the Test in the Repo Scan
database_testcase: the Test in the Database
test (TestCase): the Test in the Repo Scan
database_testcase (TestCase): the Test in the Database
"""
current_pytest_node_id = database_testcase.get_pytest_node_id()
if not current_pytest_node_id or current_pytest_node_id is not test.get_pytest_node_id():
@@ -100,9 +111,10 @@ class TestScannerUploader:
def update_capability(self, test: TestCase, test_info_id: int):
"""
Updates the test in the db with any capabilities it has
Args:
test: the test
test_info_id: the id of the test to check.
test (TestCase): the test
test_info_id (int): the id of the test to check.
"""
capability_operation = CapabilityOperation()
capability_test_operation = TestCapabilityOperation()
@@ -126,13 +138,13 @@ class TestScannerUploader:
self.check_for_capabilities_to_remove(test_info_id, capabilities)
def check_for_capabilities_to_remove(self, test_info_id, capabilities):
def check_for_capabilities_to_remove(self, test_info_id: int, capabilities: [str]):
"""
Checks for capabilities in the db that no longer exist on the test
Args:
test_info_id: the test_info_id
capabilities: the capabilities on the test
test_info_id (int): the test_info_id
capabilities ([str]): the capabilities on the test
v
"""
capability_test_operation = TestCapabilityOperation()
# next we need to remove capabilities that are in the database but no longer on the test

6
scripts/test_case_scanner.py
View File

@@ -2,9 +2,10 @@ from optparse import OptionParser
from config.configuration_file_locations_manager import ConfigurationFileLocationsManager
from config.configuration_manager import ConfigurationManager
from framework.resources.resource_finder import get_stx_repo_root
from framework.scanning.objects.test_scanner_uploader import TestScannerUploader
if __name__ == '__main__':
if __name__ == "__main__":
"""
This Function will scan the repository for all test cases and update the database.
@@ -18,6 +19,7 @@ if __name__ == '__main__':
configuration_locations_manager.set_configs_from_options_parser(parser)
ConfigurationManager.load_configs(configuration_locations_manager)
repo_root = get_stx_repo_root()
folders_to_scan = ["testcases"]
test_scanner_uploader = TestScannerUploader(folders_to_scan)
test_scanner_uploader.scan_and_upload_tests()
test_scanner_uploader.scan_and_upload_tests(repo_root)