From bdbd507fb7f0c796799f6642bf4c43be81e38958 Mon Sep 17 00:00:00 2001 From: saraNoigershel Date: Thu, 19 Sep 2024 14:11:52 +0300 Subject: [PATCH 1/4] tamar_k_parametergroup --- .../DBClusterParameterGroupController.py | 18 + .../DBClusterParameterGroupManager.py | 39 ++ .../Models/DBClusterParameterGroupModel.py | 82 ++++ .../Classes/DBClusterParameterGroupService.py | 201 ++++++++++ .../Test/DBClusterParameterGroupTests.py | 374 ++++++++++++++++++ DB/NEW_KT_DB/Test/GeneralTests.py | 29 +- 6 files changed, 737 insertions(+), 6 deletions(-) create mode 100644 DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py create mode 100644 DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py create mode 100644 DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py create mode 100644 DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py create mode 100644 DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py diff --git a/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py b/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py new file mode 100644 index 00000000..cac47782 --- /dev/null +++ b/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py @@ -0,0 +1,18 @@ +from NEW_KT_DB.Service.Classes.DBClusterParameterGroupService import DBClusterParameterGroupService +from typing import Optional, Dict + +class DBClusterParameterGroupController: + def __init__(self, service: DBClusterParameterGroupService): + self.service = service + + def create_db_cluster_parameter_group(self, group_name: str, group_family: str, description: Optional[str]=None): + return self.service.create(group_name, group_family, description) + + def delete_db_cluste_parameter_group(self, group_name: str): + self.service.delete(group_name) + + def describe_db_cluste_parameter_group(self, group_name: str = None, max_records: int = 100, marker: str = None) -> Dict: + return self.service.describe_group('DBClusterParameterGroup', group_name, max_records, marker) + + def modify_db_cluste_parameter_group(self, group_name: str, parameters: list[Dict[str, any]]): + self.service.modify('DBClusterParameterGroup', group_name, parameters) \ No newline at end of file diff --git a/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py new file mode 100644 index 00000000..a0ce1b93 --- /dev/null +++ b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py @@ -0,0 +1,39 @@ +from typing import Dict, Any +import json +import sqlite3 +from NEW_KT_DB.DataAccess.ObjectManager import ObjectManager +from NEW_KT_DB.Models.DBClusterParameterGroupModel import DBClusterParameterGroup + +class DBClusterParameterGroupManager: + def __init__(self, db_file: str): + '''Initialize ObjectManager with the database connection.''' + self.object_manager = ObjectManager(db_file) + self.object_manager.create_management_table( + DBClusterParameterGroup.get_object_name(), DBClusterParameterGroup.table_schema, 'TEXT') + + + def createInMemoryDBCluster(self, data): + self.object_manager.save_in_memory(self.__class__.__name__[:-len("Manager")], data) + + + def deleteInMemoryDBCluster(self, group_name): + self.object_manager.delete_from_memory_by_pk(self.__class__.__name__[:-len("Manager")], pk_column=DBClusterParameterGroup.pk_column, pk_value=group_name) + + def modifyDBCluster(self, group_name, data): + self.object_manager.update_in_memory(self.__class__.__name__[:-len("Manager")], updates=data, criteria=f'{DBClusterParameterGroup.pk_column} = "{group_name}"') + + def get(self, group_name): + return self.object_manager.get_from_memory(self.__class__.__name__[:-len("Manager")], columns='*', criteria=f'{DBClusterParameterGroup.pk_column} = "{group_name}"') + + def get_all_groups(self): + return self.object_manager.get_all_objects_from_memory(self.__class__.__name__[:-len("Manager")]) + + def is_identifier_exist(self, group_name): + result= self.object_manager.get_from_memory(self.__class__.__name__[:-len("Manager")], columns='*', criteria=f'{DBClusterParameterGroup.pk_column} = "{group_name}"') + if result !=[]: + return True + return False + + + + diff --git a/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py new file mode 100644 index 00000000..c8389164 --- /dev/null +++ b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py @@ -0,0 +1,82 @@ +from abc import abstractmethod +from typing import Dict, Optional, List +from NEW_KT_DB.DataAccess.ObjectManager import ObjectManager + + +class DBClusterParameterGroup: + + pk_column = 'group_name' + table_schema = """ + group_name TEXT PRIMARY KEY, + group_family TEXT, + description TEXT, + parameters TEXT + """ + def __init__(self, group_name: str, group_family: str, description: Optional[str] = None, tags: Optional[List[str]] = None, pk_column: str='DBClusterParameterGroupName', pk_value: str= None ): + self.group_name = group_name + self.group_family = group_family + self.description = description + self.parameters = self.load_default_parameters() + self.tags = tags + self.pk_column = pk_column + self.pk_value = pk_value + + def load_default_parameters(self): + """ + Loads default parameters for the DB parameter group. + + Returns: + list: Default parameters for the DB parameter group + """ + # Loading default parameters - can be replaced with actual parameters + parameters = [] + parameters.append(Parameter('backup_retention_period', 7).to_dict()) + parameters.append(Parameter('preferred_backup_window', '03:00-03:30').to_dict()) + parameters.append(Parameter('preferred_maintenance_window', 'Mon:00:00-Mon:00:30').to_dict()) + return parameters + + + def to_dict(self) -> Dict: + return ObjectManager.convert_object_attributes_to_dictionary( + group_name= self.group_name, + group_family= self.group_family, + description= self.description, + parameters= self.parameters, + # tags= self.tags, + # pk_column=self.pk_column, + # pk_value=self.pk_value + ) + @staticmethod + def get_object_name(): + return DBClusterParameterGroup.__name__ + +from typing import Optional, List, Dict + +class Parameter: + def __init__(self, parameter_name: str, parameter_value: str, description: str = '', source: str = 'engine-default', apply_method: str = '', is_modifiable: bool = True):#, apply_type: str = '', data_type: str = '', allowed_values: str = '', is_modifiable: bool = True, minimum_engine_version: str = '', apply_method: str = '', supported_engine_modes: Optional[List[str]] = None): + self.parameter_name = parameter_name + self.parameter_value = parameter_value + self.description = description + # self.source = source + # self.apply_type = apply_type + # self.data_type = data_type + # self.allowed_values = allowed_values + self.is_modifiable = is_modifiable + # self.minimum_engine_version = minimum_engine_version + self.apply_method = apply_method + # self.supported_engine_modes = supported_engine_modes + + def to_dict(self) -> Dict: + return ObjectManager.convert_object_attributes_to_dictionary( + parameter_name= self.parameter_name, + parameter_value= self.parameter_value, + description= self.description, + # source= self.source, + # apply_type=self.apply_type, + # data_type= self.data_type, + # allowed_values=self.allowed_values, + is_modifiable= self.is_modifiable, + # minimum_engine_version= self.minimum_engine_version, + apply_method= self.apply_method + # supported_engine_modes= self.supported_engine_modes + ) diff --git a/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py new file mode 100644 index 00000000..65af796b --- /dev/null +++ b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py @@ -0,0 +1,201 @@ +from abc import abstractmethod +import json +import os +import sys +from typing import Optional, Dict +# from DataAccess import ObjectManager +from NEW_KT_DB.Service.Abc.DBO import DBO +# from DB.KT_DB.Models.ParameterGroupModel import ParameterGroupModel +from NEW_KT_DB.Validation.GeneralValidations import is_valid_user_group_name, is_valid +from NEW_KT_DB.Models.DBClusterParameterGroupModel import DBClusterParameterGroup +from NEW_KT_DB.DataAccess import DBClusterManager#, DBClusterParameterGroupManager +from NEW_KT_DB.DataAccess import DBClusterParameterGroupManager +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../..'))) +from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager + +class DBClusterParameterGroupService(DBO): + """ + Service class for managing generic parameter groups. + """ + column_index_mapping = { + 'group_name': 0, + 'group_family': 1, + 'description': 2, + 'parameters': 3 + } + + def __init__(self, dal:DBClusterParameterGroupManager, dal_cluster: DBClusterManager, storage_manager: StorageManager): + """ + Initialize the service with a ObjectManager instance. + + :param dal: instance to interact with the database. + :param dal_cluster: ClusterManager instance to handle cluster-related operations. + """ + self.dal = dal + self.dal_cluster = dal_cluster + self.storage_manager=storage_manager + + def create(self, group_name: str, group_family: str, description: Optional[str] = None): + """ + Create a new parameter group. + + :param group_name: The name of the parameter group. + :param group_family: The family to which the parameter group belongs. + :param description: An optional description for the parameter group. + :param is_cluster: Indicates if the group is a DBCluster parameter group. Defaults to True. + :return: A dictionary containing details about the created parameter group. + """ + if not is_valid_user_group_name(group_name): + raise ValueError(f"group_name {group_name} is not valid") + if self.dal.is_identifier_exist(group_name): + raise ValueError(f"ParameterGroup with NAME '{group_name}' already exists.") + group = DBClusterParameterGroup(group_name, group_family, description) + parameter_group_dict=group.to_dict() + data_tuple = ( + parameter_group_dict['group_name'], + parameter_group_dict['group_family'], + parameter_group_dict.get('description', None), + json.dumps(parameter_group_dict['parameters']) + ) + self.dal.createInMemoryDBCluster(data_tuple) + file_name=f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + self.storage_manager.create_directory('db_cluster_parameter_groups') + self.storage_manager.create_file(file_name, json.dumps(parameter_group_dict)) + print(f"Creating parameter group '{group_name}' in family '{group_family}' with description '{description}'") + group_tuple=self.get(group_name) + return self.describe(group_tuple) + + def delete(self, group_name: str): + """ + Delete an existing parameter group. + + :param group_name: The name of the parameter group to delete. + :param class_name: The class name of the parameter group. + """ + if group_name == "default": + raise ValueError("You can't delete a default parameter group") + if not self.dal.is_identifier_exist(group_name): + raise ValueError(f"Parameter Group '{group_name}' does not exist.") + clusters = self.dal_cluster.get_all_clusters() + for c in clusters: + if c[6] == group_name: + raise ValueError("Can't delete parameter group associated with any DB clusters") + self.dal.deleteInMemoryDBCluster(group_name) + file_name = f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + self.storage_manager.delete_file(file_name) + print(f"Deleting parameter group '{group_name}'") + + def describe_group(self, title: str, parameter_group_name: str = None, max_records: int = 100, marker: str = None) -> Dict: + """ + Describe a specific parameter group. + + :param title: The title for the output data. + :param parameter_group_name: The name of the parameter group to describe. Optional. + :param max_records: The maximum number of records to return. + :param marker: The marker to start listing from. Used for pagination. + :return: A dictionary containing details about the parameter group(s). + """ + parameter_groups_local = [] + if parameter_group_name is not None: + data = self.get(parameter_group_name) + parameter_groups_local.append(self.describe(data)) + else: + parameter_groups = self.dal.get_all_groups() + count = 0 + for p in parameter_groups: + if p[DBClusterParameterGroupService.column_index_mapping['group_name']] == marker or marker is None: + marker = None + count += 1 + if count <= max_records: + parameter_groups_local.append(self.describe(p)) + else: + marker = p[DBClusterParameterGroupService.column_index_mapping['group_name']] + if marker is None: + return {title: parameter_groups_local} + return {'Marker': marker, title: parameter_groups_local} + + def camel_to_snake_case(self, name: str) -> str: + """ + Convert a CamelCase string to snake_case. + + :param name: The CamelCase string to convert. + :return: The snake_case version of the input string. + """ + return ''.join(['_' + c.lower() if c.isupper() else c for c in name]).lstrip('_') + + def convert_dict_keys_to_snake_case(self, input_dict: Dict) -> Dict: + """ + Convert all keys in a dictionary from CamelCase to snake_case. + + :param input_dict: The input dictionary with CamelCase keys. + :return: A new dictionary with snake_case keys. + """ + return {self.camel_to_snake_case(key): value for key, value in input_dict.items()} + + def modify(self, title: str, group_name: str, parameters: Optional[list[Dict[str, any]]] = None): + """ + Modify an existing parameter group. + + :param title: The title for the output data. + :param group_name: The name of the parameter group to modify. + :param parameters: A list of dictionaries with updates to apply to the parameter group. + :return: A dictionary containing details about the modified parameter group. + """ + parameter_group = self.get(group_name) + parameters_in_parameter_group=parameter_group[DBClusterParameterGroupService.column_index_mapping['parameters']] + parameters_in_parameter_group=json.loads(parameters_in_parameter_group) + # print(f"parameter_group{parameter_group}") + + for new_parameter in parameters: + is_valid(new_parameter['IsModifiable'], [True, False], 'IsModifiable') + is_valid(new_parameter['ApplyMethod'], ['immediate', 'pending-reboot'], 'ApplyMethod') + for idx, old_parameter in enumerate(parameters_in_parameter_group): + if new_parameter['ParameterName'] == old_parameter['parameter_name']: + if old_parameter['is_modifiable'] == False: + raise ValueError(f"You can't modify the parameter {old_parameter['parameter_name']}") + new_parameter_updates = self.convert_dict_keys_to_snake_case(new_parameter) + updated_parameter = {**old_parameter, **new_parameter_updates} + parameters_in_parameter_group[idx] = updated_parameter + self.dal.modifyDBCluster(group_name, f"parameters='{json.dumps(parameters_in_parameter_group)}'") + file_name=f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + group_family=parameter_group[DBClusterParameterGroupService.column_index_mapping['group_family']] + description=parameter_group[DBClusterParameterGroupService.column_index_mapping['description']] + group = DBClusterParameterGroup(group_name, group_family, description) + parameter_group_dict=group.to_dict() + parameter_group_dict['parameters']=parameters_in_parameter_group + self.storage_manager.write_to_file(file_name, json.dumps(parameter_group_dict)) + return {title: group_name} + + def describe(self, data: tuple) -> Dict: + """ + Abstract method to describe a parameter group. + + :param name: The name of the parameter group. + :param arn: The Amazon Resource Name (ARN) for the parameter group. + :param data: The data for the parameter group. + :return: A dictionary containing the description of the parameter group. + """ + describe = { + 'DBClusterParameterGroupName': data[DBClusterParameterGroupService.column_index_mapping['group_name']], + 'DBParameterGroupFamily': data[DBClusterParameterGroupService.column_index_mapping['group_family']], + 'Description': data[DBClusterParameterGroupService.column_index_mapping['description']], + 'DBClusterParameterGroupArn': f'arn:aws:rds:region:account:dbcluster-parameter_group/{data[DBClusterParameterGroupService.column_index_mapping["group_name"]]}' + } + return describe + + def get(self, group_name: str) -> Dict: + """ + Retrieve a parameter group by its name. + + :param group_name: The name of the parameter group to retrieve. + :return: A dictionary representing the parameter group. + :raises ValueError: If the parameter group does not exist. + + This method queries the data access layer (DAL) to retrieve the parameter group with the specified name. + If no parameter group is found, it raises a ValueError indicating that the parameter group does not exist. + Otherwise, it returns the first result as a dictionary. + """ + result = self.dal.get(group_name) + if result == []: + raise ValueError(f"Parameter Group '{group_name}' does not exist.") + return result[0] \ No newline at end of file diff --git a/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py new file mode 100644 index 00000000..88bb6825 --- /dev/null +++ b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py @@ -0,0 +1,374 @@ +import json +import os +import sys +import pytest +from unittest.mock import Mock, patch +from unittest.mock import MagicMock +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) +from NEW_KT_DB.DataAccess.DBClusterManager import DBClusterManager +from NEW_KT_DB.Controller.DBClusterParameterGroupController import DBClusterParameterGroupController +from NEW_KT_DB.Service.Classes.DBClusterParameterGroupService import DBClusterParameterGroupService +from NEW_KT_DB.DataAccess.DBClusterParameterGroupManager import DBClusterParameterGroupManager +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../..'))) +from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager +from GeneralTests import * + +@pytest.fixture +def parameter_group_manager(): + return DBClusterParameterGroupManager(':memory:') + +# @pytest.fixture +# def cluster_manager(): +# return DBClusterManager(':memory:') + +@pytest.fixture +def cluster_manager(): + # Create a mock for DBClusterManager and its method get_all_clusters + mock_cluster_manager = Mock(spec=DBClusterManager) + # Set the return value of get_all_clusters + mock_cluster_manager.get_all_clusters.return_value = {} + return mock_cluster_manager + +@pytest.fixture +def parameter_group_service(parameter_group_manager, cluster_manager, storage_manager): + return DBClusterParameterGroupService(parameter_group_manager, cluster_manager, storage_manager) + +@pytest.fixture +def parameter_group_controller(parameter_group_service): + return DBClusterParameterGroupController(parameter_group_service) + +# Generic function to create a parameter group +def create_parameter_group(controller, group_name, group_family, description): + return controller.create_db_cluster_parameter_group(group_name, group_family, description) + +# Generic function for file name +def generate_file_name_for_group (group_name): + return f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + +# Generic function to assert the parameter group's details +def assert_parameter_group_details(result, index, expected_group_name, expected_family, expected_description): + """ + Assert the details of a specific DBClusterParameterGroup in the result. + + :param result: The result dictionary returned from the describe_db_cluste_parameter_group function. + :param index: The index of the parameter group in the result list to check. + :param expected_group_name: The expected DBClusterParameterGroupName value. + :param expected_family: The expected DBParameterGroupFamily value. + :param expected_description: The expected Description value. + """ + parameter_group = result['DBClusterParameterGroup'][index] + + assert parameter_group['DBClusterParameterGroupName'] == expected_group_name, \ + f"Expected DBClusterParameterGroupName to be '{expected_group_name}' but got '{parameter_group['DBClusterParameterGroupName']}'" + assert parameter_group['DBParameterGroupFamily'] == expected_family, \ + f"Expected DBParameterGroupFamily to be '{expected_family}' but got '{parameter_group['DBParameterGroupFamily']}'" + assert parameter_group['Description'] == expected_description, \ + f"Expected Description to be '{expected_description}' but got '{parameter_group['Description']}'" + +def test_create_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + group_family = "TestFamily" + description = "Test Description" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + result = create_parameter_group(parameter_group_controller, group_name, group_family, description) + assert result['DBClusterParameterGroupName'] == group_name + assert result['DBParameterGroupFamily'] == group_family + assert result['Description'] == description + full_path = os.path.abspath(file_name) + print(f"Full path of the file: {full_path}") + # Check if the correct file was created + assert_file_exists(storage_manager, file_name) + + # Check if the file content matches the expected result + expected_data = {'group_name': group_name, 'group_family': group_family, 'description': description} + assert_json_content(storage_manager, file_name, expected_data) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_create_existing_parameter_group(parameter_group_controller): + group_name = "TestGroup" + group_family = "TestFamily" + + # Ensure the group exists + create_parameter_group(parameter_group_controller, group_name, group_family, "Test Description") + + # Test if exception is raised when trying to create an existing group + with pytest.raises(ValueError, match=f"ParameterGroup with NAME '{group_name}' already exists."): + create_parameter_group(parameter_group_controller, group_name, group_family, "Another Description") + +def test_create_parameter_group_with_invalid_name(parameter_group_controller): + invalid_group_name = "InvalidGroupName!" + + # Test if exception is raised when trying to create a group with invalid_name + with pytest.raises(ValueError, match=f"group_name {invalid_group_name} is not valid"): + create_parameter_group(parameter_group_controller, invalid_group_name, "ValidFamily", "Valid Description") + +# def test_delete_parameter_group(parameter_group_controller): +# group_name = "TestGroup" +# file_name = generate_file_name_for_group(group_name) + +# # Create the parameter group +# create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + +# # Ensure the file exists before deletion +# assert_file_exists(file_name) + +# # Delete the parameter group +# parameter_group_controller.delete_db_cluste_parameter_group(group_name) + +# # Check if the file was deleted +# assert not os.path.exists(file_name), f"Expected file {file_name} was not deleted." + +def test_delete_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Ensure the file exists before deletion + assert_file_exists(storage_manager, file_name) + + # Delete the parameter group + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + + # Check if the file was deleted + assert not os.path.exists(file_name), f"Expected file {file_name} was not deleted." + +def test_delete_parameter_group_with_associated_cluster(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Mock get_all_clusters to return a cluster associated with the parameter group + parameter_group_controller.service.dal_cluster.get_all_clusters.return_value =[("","","","","","",group_name)] #{"TestCluster": {"group_name": group_name}} + + # Attempt to delete the parameter group, expect an exception due to association with cluster + with pytest.raises(ValueError, match="Can't delete parameter group associated with any DB clusters"): + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_delete_nonexistent_parameter_group(parameter_group_controller): + group_name = "NonExistentGroup" + + # Test if exception is raised when trying to delete a non-existent group + with pytest.raises(ValueError, match=f"Parameter Group '{group_name}' does not exist."): + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + +# def test_delete_parameter_group_with_associated_cluster(parameter_group_controller, cluster_manager): +# group_name = "TestGroup" +# file_name = generate_file_name_for_group(group_name) + +# # Create the parameter group +# create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + +# # Associate the parameter group with a cluster +# cluster_manager.create_cluster({"cluster_id": "TestCluster", "group_name": group_name}) + +# # Attempt to delete the parameter group, expect an exception due to association with cluster +# with pytest.raises(ValueError, match="Can't delete parameter group associated with any DB clusters"): +# parameter_group_controller.delete_db_cluste_parameter_group(group_name) + +# # Cleanup +# delete_file_if_exists(file_name) + +def test_delete_default_parameter_group(parameter_group_controller, storage_manager): + group_name = "default" + file_name = generate_file_name_for_group(group_name) + + # Create the default parameter group + create_parameter_group(parameter_group_controller, group_name, "DefaultFamily", "Default group description") + + # Test if exception is raised when trying to delete the default group + with pytest.raises(ValueError, match="You can't delete a default parameter group"): + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create a parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Modify the parameter group with new parameters + parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': 14, 'IsModifiable': True, 'ApplyMethod': 'immediate'} + ] + parameter_group_controller.modify_db_cluste_parameter_group(group_name, parameters) + + # Check if the modifications were applied + expected_parameters = {'parameters': [{'parameter_name': 'backup_retention_period', 'parameter_value': 14, 'description': '', + 'is_modifiable': True, 'apply_method': 'immediate'}, {'parameter_name': 'preferred_backup_window', 'parameter_value': '03:00-03:30', + 'description': '', 'is_modifiable': True, 'apply_method': ''}, {'parameter_name': 'preferred_maintenance_window', + 'parameter_value': 'Mon:00:00-Mon:00:30', 'description': '', 'is_modifiable': True, 'apply_method': ''}]} + + assert_json_content(storage_manager, file_name, expected_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_nonexistent_parameter_group(parameter_group_controller): + group_name = "NonExistentGroup" + parameters = [{'ParameterName': 'backup_retention_period', 'ParameterValue': 14, 'IsModifiable': True, 'ApplyMethod': 'immediate'}] + + # Test if exception is raised when trying to modify a non-existent group + with pytest.raises(ValueError, match=f"Parameter Group '{group_name}' does not exist."): + parameter_group_controller.modify_db_cluste_parameter_group(group_name, parameters) + +def test_modify_non_modifiable_parameter(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Define a non-modifiable parameter + parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': 5, 'IsModifiable': False, 'ApplyMethod': 'immediate'} + ] + parameter_group_controller.modify_db_cluste_parameter_group( group_name, parameters) + + # Attempt to change a non-modifiable parameter, expect an exception + with pytest.raises(ValueError, match="You can't modify the parameter backup_retention_period"): + new_parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': 14, 'IsModifiable': False, 'ApplyMethod': 'immediate'} + ] + parameter_group_controller.modify_db_cluste_parameter_group(group_name, new_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_with_invalid_is_modifiable(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + + + invalid_parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': '14', 'IsModifiable': 'invalid_value', 'ApplyMethod': 'immediate'} + ] + + with pytest.raises(ValueError, match="value invalid_value is invalid for IsModifiable"): + parameter_group_controller.modify_db_cluste_parameter_group(group_name, invalid_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_with_invalid_apply_method(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + + invalid_parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': '14', 'IsModifiable': True, 'ApplyMethod': 'invalid_value'} + ] + + with pytest.raises(ValueError, match="value invalid_value is invalid for ApplyMethod"): + parameter_group_controller.modify_db_cluste_parameter_group(group_name, invalid_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_describe_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + family="TestFamily" + description="Test Description" + file_name = generate_file_name_for_group(group_name) + + # Create a parameter group + create_parameter_group(parameter_group_controller, group_name, family, description) + + # Describe the parameter group + result = parameter_group_controller.describe_db_cluste_parameter_group(group_name) + # Check the result contains the correct description + assert_parameter_group_details(result, 0, group_name, family, description) + result = parameter_group_controller.describe_db_cluste_parameter_group() + # Check the result contains the correct description + assert_parameter_group_details(result, 0, group_name, family, description) + + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_describe_nonexistent_parameter_group(parameter_group_controller): + group_name = "NonExistentGroup" + + # Test if exception is raised when trying to describe a non-existent group + with pytest.raises(ValueError, match=f"Parameter Group '{group_name}' does not exist."): + parameter_group_controller.describe_db_cluste_parameter_group(group_name) + +def test_describe_group_without_parameter_group_name(parameter_group_controller, storage_manager): + max_records = 2 + marker = None + + + # Mock the return of get_all_groups method to simulate multiple parameter groups + mock_parameter_groups = { + "Group1": {"group_name": "Group1", "family": "TestFamily1", "description": "Description 1"}, + "Group2": {"group_name": "Group2", "family": "TestFamily2", "description": "Description 2"}, + "Group3": {"group_name": "Group3", "family": "TestFamily3", "description": "Description 3"}, + } + for p in mock_parameter_groups.values(): + create_parameter_group(parameter_group_controller, p['group_name'], p['family'], p['description']) + # with patch.object(parameter_group_controller.service.dal, 'get_all_groups', return_value=mock_parameter_groups): + # result = parameter_group_controller.describe_db_cluste_parameter_group() + # parameter_group_controller.dal.get_all_groups = lambda: mock_parameter_groups + + # Call the describe_group without a parameter_group_name + result = parameter_group_controller.describe_db_cluste_parameter_group(max_records=max_records, marker=marker) + + # Check that the correct number of parameter groups are returned based on max_records + assert len(result["DBClusterParameterGroup"]) == max_records + for idx, p in enumerate(mock_parameter_groups.values()): + if idx >= max_records: + break + assert_parameter_group_details(result, idx, p['group_name'], p['family'], p['description']) + + + + # Check if pagination marker is returned + assert 'Marker' in result + assert result['Marker'] == "Group3" + for p in mock_parameter_groups.values(): + file_name=generate_file_name_for_group(p['group_name']) + delete_file_if_exists(storage_manager, file_name) + + +# def test_describe_group_without_any_parameters(parameter_group_controller): +# # title = "Default Title" + +# # Mock the return of get_all_groups method to simulate multiple parameter groups +# mock_parameter_groups = { +# "Group1": {"group_name": "Group1", "family": "TestFamily1", "description": "Description 1"}, +# "Group2": {"group_name": "Group2", "family": "TestFamily2", "description": "Description 2"}, +# } +# # for p in mock_parameter_groups.values(): +# # create_parameter_group(parameter_group_controller, p['group_name'], p['family'], p['description']) + +# parameter_group_controller.get_all_groups = lambda: mock_parameter_groups + +# # Call describe_group without any parameters (using default values) +# result = parameter_group_controller.describe_db_cluste_parameter_group() + +# # Verify that all parameter groups are returned (up to max_records default which is 100) +# assert len(result["DBClusterParameterGroup"]) == len(mock_parameter_groups) +# assert result["DBClusterParameterGroup"][0]['DBClusterParameterGroupName'] == "Group1" +# assert result["DBClusterParameterGroup"][1]['DBClusterParameterGroupName'] == "Group2" + +# # Check that marker is not returned since there are less than 100 records +# assert 'Marker' not in result diff --git a/DB/NEW_KT_DB/Test/GeneralTests.py b/DB/NEW_KT_DB/Test/GeneralTests.py index a97eadb4..641d447b 100644 --- a/DB/NEW_KT_DB/Test/GeneralTests.py +++ b/DB/NEW_KT_DB/Test/GeneralTests.py @@ -1,10 +1,27 @@ +import json +import os +import sys import pytest -from KT_STORAGE import StorageManager - +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../..'))) +from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager +@pytest.fixture def storage_manager(): - """Fixture to create an instance of OptionGroup.""" - return StorageManager() + """Fixture to create an instance of StorageManager.""" + return StorageManager('test') + +def assert_file_exists(storage_manager, file_name): + assert storage_manager.is_file_exist(file_name), f"Expected file {file_name} was not created." + +# Generic function to delete a file +def delete_file_if_exists(storage_manager, file_name): + storage_manager.delete_file(file_name) -def test_file_exists(file_name): - assert storage_manager.is_file_exist(file_name) \ No newline at end of file +# Generic function to load JSON file and assert its content +def assert_json_content(storage_manager, file_name, expected_data): + full_path = os.path.join(storage_manager.base_directory, file_name) + with open(full_path, 'r') as json_file: + data = json.load(json_file) + for key, value in expected_data.items(): + print(value) + assert data[key] == value, f"Expected {key} to be {value}, but got {data[key]}" \ No newline at end of file From 8cfb107ca0f1da5dc9dec14144c62fbc2459ee63 Mon Sep 17 00:00:00 2001 From: tamar koledetzky Date: Thu, 19 Sep 2024 15:47:32 +0300 Subject: [PATCH 2/4] DBClusterParameterGroup --- .../DBClusterParameterGroupController.py | 18 + .../DBClusterParameterGroupManager.py | 39 ++ .../Models/DBClusterParameterGroupModel.py | 82 ++++ .../Classes/DBClusterParameterGroupService.py | 201 ++++++++++ .../Test/DBClusterParameterGroupTests.py | 374 ++++++++++++++++++ DB/NEW_KT_DB/Test/GeneralTests.py | 29 +- 6 files changed, 737 insertions(+), 6 deletions(-) create mode 100644 DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py create mode 100644 DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py create mode 100644 DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py create mode 100644 DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py create mode 100644 DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py diff --git a/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py b/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py new file mode 100644 index 00000000..cac47782 --- /dev/null +++ b/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py @@ -0,0 +1,18 @@ +from NEW_KT_DB.Service.Classes.DBClusterParameterGroupService import DBClusterParameterGroupService +from typing import Optional, Dict + +class DBClusterParameterGroupController: + def __init__(self, service: DBClusterParameterGroupService): + self.service = service + + def create_db_cluster_parameter_group(self, group_name: str, group_family: str, description: Optional[str]=None): + return self.service.create(group_name, group_family, description) + + def delete_db_cluste_parameter_group(self, group_name: str): + self.service.delete(group_name) + + def describe_db_cluste_parameter_group(self, group_name: str = None, max_records: int = 100, marker: str = None) -> Dict: + return self.service.describe_group('DBClusterParameterGroup', group_name, max_records, marker) + + def modify_db_cluste_parameter_group(self, group_name: str, parameters: list[Dict[str, any]]): + self.service.modify('DBClusterParameterGroup', group_name, parameters) \ No newline at end of file diff --git a/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py new file mode 100644 index 00000000..a0ce1b93 --- /dev/null +++ b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py @@ -0,0 +1,39 @@ +from typing import Dict, Any +import json +import sqlite3 +from NEW_KT_DB.DataAccess.ObjectManager import ObjectManager +from NEW_KT_DB.Models.DBClusterParameterGroupModel import DBClusterParameterGroup + +class DBClusterParameterGroupManager: + def __init__(self, db_file: str): + '''Initialize ObjectManager with the database connection.''' + self.object_manager = ObjectManager(db_file) + self.object_manager.create_management_table( + DBClusterParameterGroup.get_object_name(), DBClusterParameterGroup.table_schema, 'TEXT') + + + def createInMemoryDBCluster(self, data): + self.object_manager.save_in_memory(self.__class__.__name__[:-len("Manager")], data) + + + def deleteInMemoryDBCluster(self, group_name): + self.object_manager.delete_from_memory_by_pk(self.__class__.__name__[:-len("Manager")], pk_column=DBClusterParameterGroup.pk_column, pk_value=group_name) + + def modifyDBCluster(self, group_name, data): + self.object_manager.update_in_memory(self.__class__.__name__[:-len("Manager")], updates=data, criteria=f'{DBClusterParameterGroup.pk_column} = "{group_name}"') + + def get(self, group_name): + return self.object_manager.get_from_memory(self.__class__.__name__[:-len("Manager")], columns='*', criteria=f'{DBClusterParameterGroup.pk_column} = "{group_name}"') + + def get_all_groups(self): + return self.object_manager.get_all_objects_from_memory(self.__class__.__name__[:-len("Manager")]) + + def is_identifier_exist(self, group_name): + result= self.object_manager.get_from_memory(self.__class__.__name__[:-len("Manager")], columns='*', criteria=f'{DBClusterParameterGroup.pk_column} = "{group_name}"') + if result !=[]: + return True + return False + + + + diff --git a/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py new file mode 100644 index 00000000..c8389164 --- /dev/null +++ b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py @@ -0,0 +1,82 @@ +from abc import abstractmethod +from typing import Dict, Optional, List +from NEW_KT_DB.DataAccess.ObjectManager import ObjectManager + + +class DBClusterParameterGroup: + + pk_column = 'group_name' + table_schema = """ + group_name TEXT PRIMARY KEY, + group_family TEXT, + description TEXT, + parameters TEXT + """ + def __init__(self, group_name: str, group_family: str, description: Optional[str] = None, tags: Optional[List[str]] = None, pk_column: str='DBClusterParameterGroupName', pk_value: str= None ): + self.group_name = group_name + self.group_family = group_family + self.description = description + self.parameters = self.load_default_parameters() + self.tags = tags + self.pk_column = pk_column + self.pk_value = pk_value + + def load_default_parameters(self): + """ + Loads default parameters for the DB parameter group. + + Returns: + list: Default parameters for the DB parameter group + """ + # Loading default parameters - can be replaced with actual parameters + parameters = [] + parameters.append(Parameter('backup_retention_period', 7).to_dict()) + parameters.append(Parameter('preferred_backup_window', '03:00-03:30').to_dict()) + parameters.append(Parameter('preferred_maintenance_window', 'Mon:00:00-Mon:00:30').to_dict()) + return parameters + + + def to_dict(self) -> Dict: + return ObjectManager.convert_object_attributes_to_dictionary( + group_name= self.group_name, + group_family= self.group_family, + description= self.description, + parameters= self.parameters, + # tags= self.tags, + # pk_column=self.pk_column, + # pk_value=self.pk_value + ) + @staticmethod + def get_object_name(): + return DBClusterParameterGroup.__name__ + +from typing import Optional, List, Dict + +class Parameter: + def __init__(self, parameter_name: str, parameter_value: str, description: str = '', source: str = 'engine-default', apply_method: str = '', is_modifiable: bool = True):#, apply_type: str = '', data_type: str = '', allowed_values: str = '', is_modifiable: bool = True, minimum_engine_version: str = '', apply_method: str = '', supported_engine_modes: Optional[List[str]] = None): + self.parameter_name = parameter_name + self.parameter_value = parameter_value + self.description = description + # self.source = source + # self.apply_type = apply_type + # self.data_type = data_type + # self.allowed_values = allowed_values + self.is_modifiable = is_modifiable + # self.minimum_engine_version = minimum_engine_version + self.apply_method = apply_method + # self.supported_engine_modes = supported_engine_modes + + def to_dict(self) -> Dict: + return ObjectManager.convert_object_attributes_to_dictionary( + parameter_name= self.parameter_name, + parameter_value= self.parameter_value, + description= self.description, + # source= self.source, + # apply_type=self.apply_type, + # data_type= self.data_type, + # allowed_values=self.allowed_values, + is_modifiable= self.is_modifiable, + # minimum_engine_version= self.minimum_engine_version, + apply_method= self.apply_method + # supported_engine_modes= self.supported_engine_modes + ) diff --git a/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py new file mode 100644 index 00000000..65af796b --- /dev/null +++ b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py @@ -0,0 +1,201 @@ +from abc import abstractmethod +import json +import os +import sys +from typing import Optional, Dict +# from DataAccess import ObjectManager +from NEW_KT_DB.Service.Abc.DBO import DBO +# from DB.KT_DB.Models.ParameterGroupModel import ParameterGroupModel +from NEW_KT_DB.Validation.GeneralValidations import is_valid_user_group_name, is_valid +from NEW_KT_DB.Models.DBClusterParameterGroupModel import DBClusterParameterGroup +from NEW_KT_DB.DataAccess import DBClusterManager#, DBClusterParameterGroupManager +from NEW_KT_DB.DataAccess import DBClusterParameterGroupManager +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../..'))) +from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager + +class DBClusterParameterGroupService(DBO): + """ + Service class for managing generic parameter groups. + """ + column_index_mapping = { + 'group_name': 0, + 'group_family': 1, + 'description': 2, + 'parameters': 3 + } + + def __init__(self, dal:DBClusterParameterGroupManager, dal_cluster: DBClusterManager, storage_manager: StorageManager): + """ + Initialize the service with a ObjectManager instance. + + :param dal: instance to interact with the database. + :param dal_cluster: ClusterManager instance to handle cluster-related operations. + """ + self.dal = dal + self.dal_cluster = dal_cluster + self.storage_manager=storage_manager + + def create(self, group_name: str, group_family: str, description: Optional[str] = None): + """ + Create a new parameter group. + + :param group_name: The name of the parameter group. + :param group_family: The family to which the parameter group belongs. + :param description: An optional description for the parameter group. + :param is_cluster: Indicates if the group is a DBCluster parameter group. Defaults to True. + :return: A dictionary containing details about the created parameter group. + """ + if not is_valid_user_group_name(group_name): + raise ValueError(f"group_name {group_name} is not valid") + if self.dal.is_identifier_exist(group_name): + raise ValueError(f"ParameterGroup with NAME '{group_name}' already exists.") + group = DBClusterParameterGroup(group_name, group_family, description) + parameter_group_dict=group.to_dict() + data_tuple = ( + parameter_group_dict['group_name'], + parameter_group_dict['group_family'], + parameter_group_dict.get('description', None), + json.dumps(parameter_group_dict['parameters']) + ) + self.dal.createInMemoryDBCluster(data_tuple) + file_name=f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + self.storage_manager.create_directory('db_cluster_parameter_groups') + self.storage_manager.create_file(file_name, json.dumps(parameter_group_dict)) + print(f"Creating parameter group '{group_name}' in family '{group_family}' with description '{description}'") + group_tuple=self.get(group_name) + return self.describe(group_tuple) + + def delete(self, group_name: str): + """ + Delete an existing parameter group. + + :param group_name: The name of the parameter group to delete. + :param class_name: The class name of the parameter group. + """ + if group_name == "default": + raise ValueError("You can't delete a default parameter group") + if not self.dal.is_identifier_exist(group_name): + raise ValueError(f"Parameter Group '{group_name}' does not exist.") + clusters = self.dal_cluster.get_all_clusters() + for c in clusters: + if c[6] == group_name: + raise ValueError("Can't delete parameter group associated with any DB clusters") + self.dal.deleteInMemoryDBCluster(group_name) + file_name = f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + self.storage_manager.delete_file(file_name) + print(f"Deleting parameter group '{group_name}'") + + def describe_group(self, title: str, parameter_group_name: str = None, max_records: int = 100, marker: str = None) -> Dict: + """ + Describe a specific parameter group. + + :param title: The title for the output data. + :param parameter_group_name: The name of the parameter group to describe. Optional. + :param max_records: The maximum number of records to return. + :param marker: The marker to start listing from. Used for pagination. + :return: A dictionary containing details about the parameter group(s). + """ + parameter_groups_local = [] + if parameter_group_name is not None: + data = self.get(parameter_group_name) + parameter_groups_local.append(self.describe(data)) + else: + parameter_groups = self.dal.get_all_groups() + count = 0 + for p in parameter_groups: + if p[DBClusterParameterGroupService.column_index_mapping['group_name']] == marker or marker is None: + marker = None + count += 1 + if count <= max_records: + parameter_groups_local.append(self.describe(p)) + else: + marker = p[DBClusterParameterGroupService.column_index_mapping['group_name']] + if marker is None: + return {title: parameter_groups_local} + return {'Marker': marker, title: parameter_groups_local} + + def camel_to_snake_case(self, name: str) -> str: + """ + Convert a CamelCase string to snake_case. + + :param name: The CamelCase string to convert. + :return: The snake_case version of the input string. + """ + return ''.join(['_' + c.lower() if c.isupper() else c for c in name]).lstrip('_') + + def convert_dict_keys_to_snake_case(self, input_dict: Dict) -> Dict: + """ + Convert all keys in a dictionary from CamelCase to snake_case. + + :param input_dict: The input dictionary with CamelCase keys. + :return: A new dictionary with snake_case keys. + """ + return {self.camel_to_snake_case(key): value for key, value in input_dict.items()} + + def modify(self, title: str, group_name: str, parameters: Optional[list[Dict[str, any]]] = None): + """ + Modify an existing parameter group. + + :param title: The title for the output data. + :param group_name: The name of the parameter group to modify. + :param parameters: A list of dictionaries with updates to apply to the parameter group. + :return: A dictionary containing details about the modified parameter group. + """ + parameter_group = self.get(group_name) + parameters_in_parameter_group=parameter_group[DBClusterParameterGroupService.column_index_mapping['parameters']] + parameters_in_parameter_group=json.loads(parameters_in_parameter_group) + # print(f"parameter_group{parameter_group}") + + for new_parameter in parameters: + is_valid(new_parameter['IsModifiable'], [True, False], 'IsModifiable') + is_valid(new_parameter['ApplyMethod'], ['immediate', 'pending-reboot'], 'ApplyMethod') + for idx, old_parameter in enumerate(parameters_in_parameter_group): + if new_parameter['ParameterName'] == old_parameter['parameter_name']: + if old_parameter['is_modifiable'] == False: + raise ValueError(f"You can't modify the parameter {old_parameter['parameter_name']}") + new_parameter_updates = self.convert_dict_keys_to_snake_case(new_parameter) + updated_parameter = {**old_parameter, **new_parameter_updates} + parameters_in_parameter_group[idx] = updated_parameter + self.dal.modifyDBCluster(group_name, f"parameters='{json.dumps(parameters_in_parameter_group)}'") + file_name=f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + group_family=parameter_group[DBClusterParameterGroupService.column_index_mapping['group_family']] + description=parameter_group[DBClusterParameterGroupService.column_index_mapping['description']] + group = DBClusterParameterGroup(group_name, group_family, description) + parameter_group_dict=group.to_dict() + parameter_group_dict['parameters']=parameters_in_parameter_group + self.storage_manager.write_to_file(file_name, json.dumps(parameter_group_dict)) + return {title: group_name} + + def describe(self, data: tuple) -> Dict: + """ + Abstract method to describe a parameter group. + + :param name: The name of the parameter group. + :param arn: The Amazon Resource Name (ARN) for the parameter group. + :param data: The data for the parameter group. + :return: A dictionary containing the description of the parameter group. + """ + describe = { + 'DBClusterParameterGroupName': data[DBClusterParameterGroupService.column_index_mapping['group_name']], + 'DBParameterGroupFamily': data[DBClusterParameterGroupService.column_index_mapping['group_family']], + 'Description': data[DBClusterParameterGroupService.column_index_mapping['description']], + 'DBClusterParameterGroupArn': f'arn:aws:rds:region:account:dbcluster-parameter_group/{data[DBClusterParameterGroupService.column_index_mapping["group_name"]]}' + } + return describe + + def get(self, group_name: str) -> Dict: + """ + Retrieve a parameter group by its name. + + :param group_name: The name of the parameter group to retrieve. + :return: A dictionary representing the parameter group. + :raises ValueError: If the parameter group does not exist. + + This method queries the data access layer (DAL) to retrieve the parameter group with the specified name. + If no parameter group is found, it raises a ValueError indicating that the parameter group does not exist. + Otherwise, it returns the first result as a dictionary. + """ + result = self.dal.get(group_name) + if result == []: + raise ValueError(f"Parameter Group '{group_name}' does not exist.") + return result[0] \ No newline at end of file diff --git a/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py new file mode 100644 index 00000000..88bb6825 --- /dev/null +++ b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py @@ -0,0 +1,374 @@ +import json +import os +import sys +import pytest +from unittest.mock import Mock, patch +from unittest.mock import MagicMock +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) +from NEW_KT_DB.DataAccess.DBClusterManager import DBClusterManager +from NEW_KT_DB.Controller.DBClusterParameterGroupController import DBClusterParameterGroupController +from NEW_KT_DB.Service.Classes.DBClusterParameterGroupService import DBClusterParameterGroupService +from NEW_KT_DB.DataAccess.DBClusterParameterGroupManager import DBClusterParameterGroupManager +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../..'))) +from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager +from GeneralTests import * + +@pytest.fixture +def parameter_group_manager(): + return DBClusterParameterGroupManager(':memory:') + +# @pytest.fixture +# def cluster_manager(): +# return DBClusterManager(':memory:') + +@pytest.fixture +def cluster_manager(): + # Create a mock for DBClusterManager and its method get_all_clusters + mock_cluster_manager = Mock(spec=DBClusterManager) + # Set the return value of get_all_clusters + mock_cluster_manager.get_all_clusters.return_value = {} + return mock_cluster_manager + +@pytest.fixture +def parameter_group_service(parameter_group_manager, cluster_manager, storage_manager): + return DBClusterParameterGroupService(parameter_group_manager, cluster_manager, storage_manager) + +@pytest.fixture +def parameter_group_controller(parameter_group_service): + return DBClusterParameterGroupController(parameter_group_service) + +# Generic function to create a parameter group +def create_parameter_group(controller, group_name, group_family, description): + return controller.create_db_cluster_parameter_group(group_name, group_family, description) + +# Generic function for file name +def generate_file_name_for_group (group_name): + return f'db_cluster_parameter_groups/db_cluster_parameter_group_{group_name}.json' + +# Generic function to assert the parameter group's details +def assert_parameter_group_details(result, index, expected_group_name, expected_family, expected_description): + """ + Assert the details of a specific DBClusterParameterGroup in the result. + + :param result: The result dictionary returned from the describe_db_cluste_parameter_group function. + :param index: The index of the parameter group in the result list to check. + :param expected_group_name: The expected DBClusterParameterGroupName value. + :param expected_family: The expected DBParameterGroupFamily value. + :param expected_description: The expected Description value. + """ + parameter_group = result['DBClusterParameterGroup'][index] + + assert parameter_group['DBClusterParameterGroupName'] == expected_group_name, \ + f"Expected DBClusterParameterGroupName to be '{expected_group_name}' but got '{parameter_group['DBClusterParameterGroupName']}'" + assert parameter_group['DBParameterGroupFamily'] == expected_family, \ + f"Expected DBParameterGroupFamily to be '{expected_family}' but got '{parameter_group['DBParameterGroupFamily']}'" + assert parameter_group['Description'] == expected_description, \ + f"Expected Description to be '{expected_description}' but got '{parameter_group['Description']}'" + +def test_create_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + group_family = "TestFamily" + description = "Test Description" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + result = create_parameter_group(parameter_group_controller, group_name, group_family, description) + assert result['DBClusterParameterGroupName'] == group_name + assert result['DBParameterGroupFamily'] == group_family + assert result['Description'] == description + full_path = os.path.abspath(file_name) + print(f"Full path of the file: {full_path}") + # Check if the correct file was created + assert_file_exists(storage_manager, file_name) + + # Check if the file content matches the expected result + expected_data = {'group_name': group_name, 'group_family': group_family, 'description': description} + assert_json_content(storage_manager, file_name, expected_data) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_create_existing_parameter_group(parameter_group_controller): + group_name = "TestGroup" + group_family = "TestFamily" + + # Ensure the group exists + create_parameter_group(parameter_group_controller, group_name, group_family, "Test Description") + + # Test if exception is raised when trying to create an existing group + with pytest.raises(ValueError, match=f"ParameterGroup with NAME '{group_name}' already exists."): + create_parameter_group(parameter_group_controller, group_name, group_family, "Another Description") + +def test_create_parameter_group_with_invalid_name(parameter_group_controller): + invalid_group_name = "InvalidGroupName!" + + # Test if exception is raised when trying to create a group with invalid_name + with pytest.raises(ValueError, match=f"group_name {invalid_group_name} is not valid"): + create_parameter_group(parameter_group_controller, invalid_group_name, "ValidFamily", "Valid Description") + +# def test_delete_parameter_group(parameter_group_controller): +# group_name = "TestGroup" +# file_name = generate_file_name_for_group(group_name) + +# # Create the parameter group +# create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + +# # Ensure the file exists before deletion +# assert_file_exists(file_name) + +# # Delete the parameter group +# parameter_group_controller.delete_db_cluste_parameter_group(group_name) + +# # Check if the file was deleted +# assert not os.path.exists(file_name), f"Expected file {file_name} was not deleted." + +def test_delete_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Ensure the file exists before deletion + assert_file_exists(storage_manager, file_name) + + # Delete the parameter group + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + + # Check if the file was deleted + assert not os.path.exists(file_name), f"Expected file {file_name} was not deleted." + +def test_delete_parameter_group_with_associated_cluster(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Mock get_all_clusters to return a cluster associated with the parameter group + parameter_group_controller.service.dal_cluster.get_all_clusters.return_value =[("","","","","","",group_name)] #{"TestCluster": {"group_name": group_name}} + + # Attempt to delete the parameter group, expect an exception due to association with cluster + with pytest.raises(ValueError, match="Can't delete parameter group associated with any DB clusters"): + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_delete_nonexistent_parameter_group(parameter_group_controller): + group_name = "NonExistentGroup" + + # Test if exception is raised when trying to delete a non-existent group + with pytest.raises(ValueError, match=f"Parameter Group '{group_name}' does not exist."): + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + +# def test_delete_parameter_group_with_associated_cluster(parameter_group_controller, cluster_manager): +# group_name = "TestGroup" +# file_name = generate_file_name_for_group(group_name) + +# # Create the parameter group +# create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + +# # Associate the parameter group with a cluster +# cluster_manager.create_cluster({"cluster_id": "TestCluster", "group_name": group_name}) + +# # Attempt to delete the parameter group, expect an exception due to association with cluster +# with pytest.raises(ValueError, match="Can't delete parameter group associated with any DB clusters"): +# parameter_group_controller.delete_db_cluste_parameter_group(group_name) + +# # Cleanup +# delete_file_if_exists(file_name) + +def test_delete_default_parameter_group(parameter_group_controller, storage_manager): + group_name = "default" + file_name = generate_file_name_for_group(group_name) + + # Create the default parameter group + create_parameter_group(parameter_group_controller, group_name, "DefaultFamily", "Default group description") + + # Test if exception is raised when trying to delete the default group + with pytest.raises(ValueError, match="You can't delete a default parameter group"): + parameter_group_controller.delete_db_cluste_parameter_group(group_name) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create a parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Modify the parameter group with new parameters + parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': 14, 'IsModifiable': True, 'ApplyMethod': 'immediate'} + ] + parameter_group_controller.modify_db_cluste_parameter_group(group_name, parameters) + + # Check if the modifications were applied + expected_parameters = {'parameters': [{'parameter_name': 'backup_retention_period', 'parameter_value': 14, 'description': '', + 'is_modifiable': True, 'apply_method': 'immediate'}, {'parameter_name': 'preferred_backup_window', 'parameter_value': '03:00-03:30', + 'description': '', 'is_modifiable': True, 'apply_method': ''}, {'parameter_name': 'preferred_maintenance_window', + 'parameter_value': 'Mon:00:00-Mon:00:30', 'description': '', 'is_modifiable': True, 'apply_method': ''}]} + + assert_json_content(storage_manager, file_name, expected_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_nonexistent_parameter_group(parameter_group_controller): + group_name = "NonExistentGroup" + parameters = [{'ParameterName': 'backup_retention_period', 'ParameterValue': 14, 'IsModifiable': True, 'ApplyMethod': 'immediate'}] + + # Test if exception is raised when trying to modify a non-existent group + with pytest.raises(ValueError, match=f"Parameter Group '{group_name}' does not exist."): + parameter_group_controller.modify_db_cluste_parameter_group(group_name, parameters) + +def test_modify_non_modifiable_parameter(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + # Define a non-modifiable parameter + parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': 5, 'IsModifiable': False, 'ApplyMethod': 'immediate'} + ] + parameter_group_controller.modify_db_cluste_parameter_group( group_name, parameters) + + # Attempt to change a non-modifiable parameter, expect an exception + with pytest.raises(ValueError, match="You can't modify the parameter backup_retention_period"): + new_parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': 14, 'IsModifiable': False, 'ApplyMethod': 'immediate'} + ] + parameter_group_controller.modify_db_cluste_parameter_group(group_name, new_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_with_invalid_is_modifiable(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + + + invalid_parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': '14', 'IsModifiable': 'invalid_value', 'ApplyMethod': 'immediate'} + ] + + with pytest.raises(ValueError, match="value invalid_value is invalid for IsModifiable"): + parameter_group_controller.modify_db_cluste_parameter_group(group_name, invalid_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_with_invalid_apply_method(parameter_group_controller, storage_manager): + group_name = "TestGroup" + file_name = generate_file_name_for_group(group_name) + + create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") + + + invalid_parameters = [ + {'ParameterName': 'backup_retention_period', 'ParameterValue': '14', 'IsModifiable': True, 'ApplyMethod': 'invalid_value'} + ] + + with pytest.raises(ValueError, match="value invalid_value is invalid for ApplyMethod"): + parameter_group_controller.modify_db_cluste_parameter_group(group_name, invalid_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_describe_parameter_group(parameter_group_controller, storage_manager): + group_name = "TestGroup" + family="TestFamily" + description="Test Description" + file_name = generate_file_name_for_group(group_name) + + # Create a parameter group + create_parameter_group(parameter_group_controller, group_name, family, description) + + # Describe the parameter group + result = parameter_group_controller.describe_db_cluste_parameter_group(group_name) + # Check the result contains the correct description + assert_parameter_group_details(result, 0, group_name, family, description) + result = parameter_group_controller.describe_db_cluste_parameter_group() + # Check the result contains the correct description + assert_parameter_group_details(result, 0, group_name, family, description) + + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_describe_nonexistent_parameter_group(parameter_group_controller): + group_name = "NonExistentGroup" + + # Test if exception is raised when trying to describe a non-existent group + with pytest.raises(ValueError, match=f"Parameter Group '{group_name}' does not exist."): + parameter_group_controller.describe_db_cluste_parameter_group(group_name) + +def test_describe_group_without_parameter_group_name(parameter_group_controller, storage_manager): + max_records = 2 + marker = None + + + # Mock the return of get_all_groups method to simulate multiple parameter groups + mock_parameter_groups = { + "Group1": {"group_name": "Group1", "family": "TestFamily1", "description": "Description 1"}, + "Group2": {"group_name": "Group2", "family": "TestFamily2", "description": "Description 2"}, + "Group3": {"group_name": "Group3", "family": "TestFamily3", "description": "Description 3"}, + } + for p in mock_parameter_groups.values(): + create_parameter_group(parameter_group_controller, p['group_name'], p['family'], p['description']) + # with patch.object(parameter_group_controller.service.dal, 'get_all_groups', return_value=mock_parameter_groups): + # result = parameter_group_controller.describe_db_cluste_parameter_group() + # parameter_group_controller.dal.get_all_groups = lambda: mock_parameter_groups + + # Call the describe_group without a parameter_group_name + result = parameter_group_controller.describe_db_cluste_parameter_group(max_records=max_records, marker=marker) + + # Check that the correct number of parameter groups are returned based on max_records + assert len(result["DBClusterParameterGroup"]) == max_records + for idx, p in enumerate(mock_parameter_groups.values()): + if idx >= max_records: + break + assert_parameter_group_details(result, idx, p['group_name'], p['family'], p['description']) + + + + # Check if pagination marker is returned + assert 'Marker' in result + assert result['Marker'] == "Group3" + for p in mock_parameter_groups.values(): + file_name=generate_file_name_for_group(p['group_name']) + delete_file_if_exists(storage_manager, file_name) + + +# def test_describe_group_without_any_parameters(parameter_group_controller): +# # title = "Default Title" + +# # Mock the return of get_all_groups method to simulate multiple parameter groups +# mock_parameter_groups = { +# "Group1": {"group_name": "Group1", "family": "TestFamily1", "description": "Description 1"}, +# "Group2": {"group_name": "Group2", "family": "TestFamily2", "description": "Description 2"}, +# } +# # for p in mock_parameter_groups.values(): +# # create_parameter_group(parameter_group_controller, p['group_name'], p['family'], p['description']) + +# parameter_group_controller.get_all_groups = lambda: mock_parameter_groups + +# # Call describe_group without any parameters (using default values) +# result = parameter_group_controller.describe_db_cluste_parameter_group() + +# # Verify that all parameter groups are returned (up to max_records default which is 100) +# assert len(result["DBClusterParameterGroup"]) == len(mock_parameter_groups) +# assert result["DBClusterParameterGroup"][0]['DBClusterParameterGroupName'] == "Group1" +# assert result["DBClusterParameterGroup"][1]['DBClusterParameterGroupName'] == "Group2" + +# # Check that marker is not returned since there are less than 100 records +# assert 'Marker' not in result diff --git a/DB/NEW_KT_DB/Test/GeneralTests.py b/DB/NEW_KT_DB/Test/GeneralTests.py index a97eadb4..641d447b 100644 --- a/DB/NEW_KT_DB/Test/GeneralTests.py +++ b/DB/NEW_KT_DB/Test/GeneralTests.py @@ -1,10 +1,27 @@ +import json +import os +import sys import pytest -from KT_STORAGE import StorageManager - +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../..'))) +from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager +@pytest.fixture def storage_manager(): - """Fixture to create an instance of OptionGroup.""" - return StorageManager() + """Fixture to create an instance of StorageManager.""" + return StorageManager('test') + +def assert_file_exists(storage_manager, file_name): + assert storage_manager.is_file_exist(file_name), f"Expected file {file_name} was not created." + +# Generic function to delete a file +def delete_file_if_exists(storage_manager, file_name): + storage_manager.delete_file(file_name) -def test_file_exists(file_name): - assert storage_manager.is_file_exist(file_name) \ No newline at end of file +# Generic function to load JSON file and assert its content +def assert_json_content(storage_manager, file_name, expected_data): + full_path = os.path.join(storage_manager.base_directory, file_name) + with open(full_path, 'r') as json_file: + data = json.load(json_file) + for key, value in expected_data.items(): + print(value) + assert data[key] == value, f"Expected {key} to be {value}, but got {data[key]}" \ No newline at end of file From 216f9f79bfdb8a345bea7d41033bbc968649c12c Mon Sep 17 00:00:00 2001 From: tamar koledetzky Date: Thu, 19 Sep 2024 16:10:35 +0300 Subject: [PATCH 3/4] parameter_proup_finish --- DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py | 1 + DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py | 1 + DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py | 1 - DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py | 1 - DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py | 1 + DB/NEW_KT_DB/Test/GeneralTests.py | 1 + 6 files changed, 4 insertions(+), 2 deletions(-) diff --git a/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py b/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py index cac47782..038208a4 100644 --- a/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py +++ b/DB/NEW_KT_DB/Controller/DBClusterParameterGroupController.py @@ -1,6 +1,7 @@ from NEW_KT_DB.Service.Classes.DBClusterParameterGroupService import DBClusterParameterGroupService from typing import Optional, Dict + class DBClusterParameterGroupController: def __init__(self, service: DBClusterParameterGroupService): self.service = service diff --git a/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py index a0ce1b93..31e2df4a 100644 --- a/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py +++ b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py @@ -4,6 +4,7 @@ from NEW_KT_DB.DataAccess.ObjectManager import ObjectManager from NEW_KT_DB.Models.DBClusterParameterGroupModel import DBClusterParameterGroup + class DBClusterParameterGroupManager: def __init__(self, db_file: str): '''Initialize ObjectManager with the database connection.''' diff --git a/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py index c8389164..9cb393a5 100644 --- a/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py +++ b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py @@ -2,7 +2,6 @@ from typing import Dict, Optional, List from NEW_KT_DB.DataAccess.ObjectManager import ObjectManager - class DBClusterParameterGroup: pk_column = 'group_name' diff --git a/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py index 65af796b..699985ca 100644 --- a/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py +++ b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py @@ -3,7 +3,6 @@ import os import sys from typing import Optional, Dict -# from DataAccess import ObjectManager from NEW_KT_DB.Service.Abc.DBO import DBO # from DB.KT_DB.Models.ParameterGroupModel import ParameterGroupModel from NEW_KT_DB.Validation.GeneralValidations import is_valid_user_group_name, is_valid diff --git a/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py index 88bb6825..09bd058d 100644 --- a/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py +++ b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py @@ -13,6 +13,7 @@ from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager from GeneralTests import * + @pytest.fixture def parameter_group_manager(): return DBClusterParameterGroupManager(':memory:') diff --git a/DB/NEW_KT_DB/Test/GeneralTests.py b/DB/NEW_KT_DB/Test/GeneralTests.py index 641d447b..f3a9d0c7 100644 --- a/DB/NEW_KT_DB/Test/GeneralTests.py +++ b/DB/NEW_KT_DB/Test/GeneralTests.py @@ -5,6 +5,7 @@ sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../..'))) from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager + @pytest.fixture def storage_manager(): """Fixture to create an instance of StorageManager.""" From 94c9bc431f55b50574f268b6a3ed42f5f4af0e9f Mon Sep 17 00:00:00 2001 From: tamar koledetzky Date: Thu, 19 Sep 2024 18:40:33 +0300 Subject: [PATCH 4/4] after_CR --- .../DBClusterParameterGroupManager.py | 8 +-- .../Models/DBClusterParameterGroupModel.py | 2 +- .../Classes/DBClusterParameterGroupService.py | 8 +-- .../Test/DBClusterParameterGroupTests.py | 62 ------------------- 4 files changed, 7 insertions(+), 73 deletions(-) diff --git a/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py index 31e2df4a..f1275610 100644 --- a/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py +++ b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py @@ -10,7 +10,7 @@ def __init__(self, db_file: str): '''Initialize ObjectManager with the database connection.''' self.object_manager = ObjectManager(db_file) self.object_manager.create_management_table( - DBClusterParameterGroup.get_object_name(), DBClusterParameterGroup.table_schema, 'TEXT') + DBClusterParameterGroup.get_object_name(), DBClusterParameterGroup.table_structure, 'TEXT') def createInMemoryDBCluster(self, data): @@ -33,8 +33,4 @@ def is_identifier_exist(self, group_name): result= self.object_manager.get_from_memory(self.__class__.__name__[:-len("Manager")], columns='*', criteria=f'{DBClusterParameterGroup.pk_column} = "{group_name}"') if result !=[]: return True - return False - - - - + return False \ No newline at end of file diff --git a/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py index 9cb393a5..0a6753b2 100644 --- a/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py +++ b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py @@ -5,7 +5,7 @@ class DBClusterParameterGroup: pk_column = 'group_name' - table_schema = """ + table_structure = """ group_name TEXT PRIMARY KEY, group_family TEXT, description TEXT, diff --git a/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py index 699985ca..7cdf50cf 100644 --- a/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py +++ b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py @@ -113,7 +113,7 @@ def describe_group(self, title: str, parameter_group_name: str = None, max_recor return {title: parameter_groups_local} return {'Marker': marker, title: parameter_groups_local} - def camel_to_snake_case(self, name: str) -> str: + def convert_camel_case_string_to_snake(self, name: str) -> str: """ Convert a CamelCase string to snake_case. @@ -122,14 +122,14 @@ def camel_to_snake_case(self, name: str) -> str: """ return ''.join(['_' + c.lower() if c.isupper() else c for c in name]).lstrip('_') - def convert_dict_keys_to_snake_case(self, input_dict: Dict) -> Dict: + def convert_dict_keys_from_camel_case_to_snake(self, input_dict: Dict) -> Dict: """ Convert all keys in a dictionary from CamelCase to snake_case. :param input_dict: The input dictionary with CamelCase keys. :return: A new dictionary with snake_case keys. """ - return {self.camel_to_snake_case(key): value for key, value in input_dict.items()} + return {self.convert_camel_case_string_to_snake(key): value for key, value in input_dict.items()} def modify(self, title: str, group_name: str, parameters: Optional[list[Dict[str, any]]] = None): """ @@ -152,7 +152,7 @@ def modify(self, title: str, group_name: str, parameters: Optional[list[Dict[str if new_parameter['ParameterName'] == old_parameter['parameter_name']: if old_parameter['is_modifiable'] == False: raise ValueError(f"You can't modify the parameter {old_parameter['parameter_name']}") - new_parameter_updates = self.convert_dict_keys_to_snake_case(new_parameter) + new_parameter_updates = self.convert_dict_keys_from_camel_case_to_snake(new_parameter) updated_parameter = {**old_parameter, **new_parameter_updates} parameters_in_parameter_group[idx] = updated_parameter self.dal.modifyDBCluster(group_name, f"parameters='{json.dumps(parameters_in_parameter_group)}'") diff --git a/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py index 09bd058d..2579909c 100644 --- a/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py +++ b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py @@ -18,10 +18,6 @@ def parameter_group_manager(): return DBClusterParameterGroupManager(':memory:') -# @pytest.fixture -# def cluster_manager(): -# return DBClusterManager(':memory:') - @pytest.fixture def cluster_manager(): # Create a mock for DBClusterManager and its method get_all_clusters @@ -107,22 +103,6 @@ def test_create_parameter_group_with_invalid_name(parameter_group_controller): with pytest.raises(ValueError, match=f"group_name {invalid_group_name} is not valid"): create_parameter_group(parameter_group_controller, invalid_group_name, "ValidFamily", "Valid Description") -# def test_delete_parameter_group(parameter_group_controller): -# group_name = "TestGroup" -# file_name = generate_file_name_for_group(group_name) - -# # Create the parameter group -# create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") - -# # Ensure the file exists before deletion -# assert_file_exists(file_name) - -# # Delete the parameter group -# parameter_group_controller.delete_db_cluste_parameter_group(group_name) - -# # Check if the file was deleted -# assert not os.path.exists(file_name), f"Expected file {file_name} was not deleted." - def test_delete_parameter_group(parameter_group_controller, storage_manager): group_name = "TestGroup" file_name = generate_file_name_for_group(group_name) @@ -163,23 +143,6 @@ def test_delete_nonexistent_parameter_group(parameter_group_controller): with pytest.raises(ValueError, match=f"Parameter Group '{group_name}' does not exist."): parameter_group_controller.delete_db_cluste_parameter_group(group_name) -# def test_delete_parameter_group_with_associated_cluster(parameter_group_controller, cluster_manager): -# group_name = "TestGroup" -# file_name = generate_file_name_for_group(group_name) - -# # Create the parameter group -# create_parameter_group(parameter_group_controller, group_name, "TestFamily", "Test Description") - -# # Associate the parameter group with a cluster -# cluster_manager.create_cluster({"cluster_id": "TestCluster", "group_name": group_name}) - -# # Attempt to delete the parameter group, expect an exception due to association with cluster -# with pytest.raises(ValueError, match="Can't delete parameter group associated with any DB clusters"): -# parameter_group_controller.delete_db_cluste_parameter_group(group_name) - -# # Cleanup -# delete_file_if_exists(file_name) - def test_delete_default_parameter_group(parameter_group_controller, storage_manager): group_name = "default" file_name = generate_file_name_for_group(group_name) @@ -348,28 +311,3 @@ def test_describe_group_without_parameter_group_name(parameter_group_controller, for p in mock_parameter_groups.values(): file_name=generate_file_name_for_group(p['group_name']) delete_file_if_exists(storage_manager, file_name) - - -# def test_describe_group_without_any_parameters(parameter_group_controller): -# # title = "Default Title" - -# # Mock the return of get_all_groups method to simulate multiple parameter groups -# mock_parameter_groups = { -# "Group1": {"group_name": "Group1", "family": "TestFamily1", "description": "Description 1"}, -# "Group2": {"group_name": "Group2", "family": "TestFamily2", "description": "Description 2"}, -# } -# # for p in mock_parameter_groups.values(): -# # create_parameter_group(parameter_group_controller, p['group_name'], p['family'], p['description']) - -# parameter_group_controller.get_all_groups = lambda: mock_parameter_groups - -# # Call describe_group without any parameters (using default values) -# result = parameter_group_controller.describe_db_cluste_parameter_group() - -# # Verify that all parameter groups are returned (up to max_records default which is 100) -# assert len(result["DBClusterParameterGroup"]) == len(mock_parameter_groups) -# assert result["DBClusterParameterGroup"][0]['DBClusterParameterGroupName'] == "Group1" -# assert result["DBClusterParameterGroup"][1]['DBClusterParameterGroupName'] == "Group2" - -# # Check that marker is not returned since there are less than 100 records -# assert 'Marker' not in result