From d5ea8f3bdb2dff2c85b688f10e4267f6686f620b Mon Sep 17 00:00:00 2001 From: croy Date: Fri, 22 Aug 2025 15:01:34 -0400 Subject: [PATCH] TestScanner now inserts new capabilities in database. test_scanner_uploader will now insert new capabilities in the database. Change-Id: I7a054b73ee8958c08c67dc2b01a6a3b83eff0e83 Signed-off-by: croy --- .../operations/capability_operation.py | 35 +++++++++++++++---- .../scanning/objects/test_scanner_uploader.py | 20 ++++++----- 2 files changed, 39 insertions(+), 16 deletions(-) diff --git a/framework/database/operations/capability_operation.py b/framework/database/operations/capability_operation.py index 2615dd18..2313ec5e 100644 --- a/framework/database/operations/capability_operation.py +++ b/framework/database/operations/capability_operation.py @@ -1,7 +1,8 @@ +from psycopg2.extras import RealDictCursor + from framework.database.connection.database_operation_manager import DatabaseOperationManager from framework.database.objects.capability import Capability from framework.logging.automation_logger import get_logger -from psycopg2.extras import RealDictCursor class CapabilityOperation: @@ -12,16 +13,36 @@ class CapabilityOperation: def __init__(self): self.database_operation_manager = DatabaseOperationManager() + def insert_capability(self, capability_name: str, capability_marker: str) -> Capability: + """ + Inserts into the database a new capability with capability_name and capability_marker passed in. + + Args: + capability_name (str): the name of the new capability + capability_marker (str): the marker associated with the new capability + + Returns: + Capability: A Capability object with the capability_id generated from the database. + """ + insert_query = f"INSERT INTO capability (capability_name, capability_marker) " f"VALUES ('{capability_name}', '{capability_marker}') " f"returning capability_id" + + results = self.database_operation_manager.execute_query(insert_query, cursor_factory=RealDictCursor) + if results: # can only ever be 1 result + capability = Capability(results[0]["capability_id"], capability_name, capability_marker) + return capability + + raise ValueError("Unable to insert the capability and get a capability_id.") + def get_capability_by_marker(self, capability_marker: str) -> Capability: """ Getter for capability given a capability name + Args: - capability_marker (): the marker - - Returns: The capability + capability_marker (str): The marker used to identify the capability. + Returns: + Capability: The capability object if found, -1 if no capability is found. """ - get_capability_query = f"SELECT * FROM capability where capability_marker='{capability_marker}'" results = self.database_operation_manager.execute_query(get_capability_query, cursor_factory=RealDictCursor) @@ -29,9 +50,9 @@ class CapabilityOperation: if len(results) > 1: get_logger().log_info(f"WARNING: We have found more than one result matching the Capability Name: {capability_marker}") result = results[0] - capability = Capability(result['capability_id'], result['capability_name'], result['capability_marker']) + capability = Capability(result["capability_id"], result["capability_name"], result["capability_marker"]) return capability else: - get_logger().log_error(f"ERROR: no capability with the name {capability_marker}") + get_logger().log_warning(f"There is no capability with the name {capability_marker}") return -1 diff --git a/framework/scanning/objects/test_scanner_uploader.py b/framework/scanning/objects/test_scanner_uploader.py index 15e1251e..54caee42 100644 --- a/framework/scanning/objects/test_scanner_uploader.py +++ b/framework/scanning/objects/test_scanner_uploader.py @@ -6,6 +6,7 @@ from framework.database.objects.testcase import TestCase from framework.database.operations.capability_operation import CapabilityOperation from framework.database.operations.test_capability_operation import TestCapabilityOperation from framework.database.operations.test_info_operation import TestInfoOperation +from framework.logging.automation_logger import get_logger from framework.pytest_plugins.collection_plugin import CollectionPlugin @@ -120,14 +121,15 @@ class TestScannerUploader: capability_test_operation = TestCapabilityOperation() # get all the capabilities associated with this test - capabilities = test.get_markers() - for capability in capabilities: + capability_markers = test.get_markers() + for capability_marker in capability_markers: - capability_object = capability_operation.get_capability_by_marker(capability) + capability_object = capability_operation.get_capability_by_marker(capability_marker) - # If capability does not exist, raise an exception + # If capability does not exist, insert it in the database and update the capability_object with the returned capability_id. if capability_object == -1: - raise ValueError(f"No capability with name {capability} exists") + get_logger().log_info(f"Inserting a new Capability in the database for marker: {capability_marker}") + capability_object = capability_operation.insert_capability(capability_marker, capability_marker) capability_id = capability_object.get_capability_id() @@ -136,15 +138,15 @@ class TestScannerUploader: if mapping_id == -1: capability_test_operation.create_new_mapping(capability_id, test_info_id) - self.check_for_capabilities_to_remove(test_info_id, capabilities) + self.check_for_capabilities_to_remove(test_info_id, capability_markers) - def check_for_capabilities_to_remove(self, test_info_id: int, capabilities: [str]): + def check_for_capabilities_to_remove(self, test_info_id: int, capability_markers: [str]): """ Checks for capabilities in the db that no longer exist on the test Args: test_info_id (int): the test_info_id - capabilities ([str]): the capabilities on the test + capability_markers ([str]): the capability markers on the test """ # next we need to remove capabilities that are in the database but no longer on the test @@ -155,7 +157,7 @@ class TestScannerUploader: db_marker_names = map(lambda db_capability: db_capability.get_capability_marker(), db_capabilities) # this will get the capabilities that exist in the db but not on the test itself - capabilities_to_be_removed = list(set(db_marker_names).difference(capabilities)) + capabilities_to_be_removed = list(set(db_marker_names).difference(capability_markers)) for marker_name in capabilities_to_be_removed: # find the correct db_capability