diff --git a/DB/KT_DB/Service/Classes/ParameterGroupService.py b/DB/KT_DB/Service/Classes/ParameterGroupService.py index b1131484..1b85bd1e 100644 --- a/DB/KT_DB/Service/Classes/ParameterGroupService.py +++ b/DB/KT_DB/Service/Classes/ParameterGroupService.py @@ -58,7 +58,7 @@ def delete(self, group_name: str, class_name: str): data = self.dal_cluster.get_all_objects() clusters = list(data.values()) for c in clusters: - if c['db_cluster_parameter_group_name'] == group_name: + if c['group_name'] == group_name: raise ValueError("Can't delete parameter group associated with any DB clusters") self.dal.delete(group_name) print(f"Deleting parameter group '{group_name}'") 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/DBClusterManager.py b/DB/NEW_KT_DB/DataAccess/DBClusterManager.py index 802c229f..2a00505a 100644 --- a/DB/NEW_KT_DB/DataAccess/DBClusterManager.py +++ b/DB/NEW_KT_DB/DataAccess/DBClusterManager.py @@ -1,14 +1,14 @@ from typing import Dict, Any import json import sqlite3 -from DataAccess import ObjectManager +from NEW_KT_DB.DataAccess.ObjectManager import ObjectManager class DBClusterManager: def __init__(self, db_file: str): '''Initialize ObjectManager with the database connection.''' self.object_manager = ObjectManager(db_file) - self.table_name ='cluster_managment' - self.create_table() + # self.table_name ='cluster_managment' + # self.create_table() def createInMemoryDBCluster(self): @@ -25,4 +25,7 @@ def describeDBCluster(self): def modifyDBCluster(self): self.object_manager.update_in_memory() - + + def get_all_clusters(self): + return self.object_manager.get_from_memory(self.__class__.__name__[:-len("Manager")]) + \ 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..c1c6ab0d --- /dev/null +++ b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py @@ -0,0 +1,35 @@ +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_structure, '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 \ No newline at end of file diff --git a/DB/NEW_KT_DB/DataAccess/ObjectManager.py b/DB/NEW_KT_DB/DataAccess/ObjectManager.py index b75b2373..a79eec5e 100644 --- a/DB/NEW_KT_DB/DataAccess/ObjectManager.py +++ b/DB/NEW_KT_DB/DataAccess/ObjectManager.py @@ -1,7 +1,10 @@ from typing import Dict, Any, Optional import json import sqlite3 -from DBManager import DBManager +import os +import sys +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../..'))) +from NEW_KT_DB.DataAccess.DBManager import DBManager class ObjectManager: def __init__(self, db_file: str): diff --git a/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py b/DB/NEW_KT_DB/Models/DBClusterParameterGroupModel.py new file mode 100644 index 00000000..cad089c1 --- /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_structure = """ + 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..959725f5 --- /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 convert_camel_case_string_to_snake(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_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.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): + """ + 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_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)}'") + 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..2a43beb1 --- /dev/null +++ b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py @@ -0,0 +1,312 @@ +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(): + # 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, 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_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) \ No newline at end of file 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 diff --git a/Storage/KT_Storage/DataAccess/StorageManager.py b/Storage/KT_Storage/DataAccess/StorageManager.py index eee45534..94f7e5d2 100644 --- a/Storage/KT_Storage/DataAccess/StorageManager.py +++ b/Storage/KT_Storage/DataAccess/StorageManager.py @@ -3,9 +3,9 @@ import os import aiofiles import shutil -from Crypto.Cipher import AES -from Crypto.Util.Padding import pad -import base6import +from Cryptodome.Cipher import AES +from Cryptodome.Util.Padding import pad +# import base6import URL_SERVER = 's3/KT_cloud/Storage/server' @@ -48,7 +48,7 @@ def get(self, bucket, key , version_id) -> Dict[str, Any]: file_path = os.path.join(self.server_path, bucket, versioned_file_name) if not os.path.exists(file_path): - return {'error': 'File not found'} + raise FileNotFoundError(f"File '{key}' with version '{version_id}' not found in bucket '{bucket}'.") if os.path.isdir(file_path): # If the object is a directory, return its metadata and list of contents @@ -76,7 +76,7 @@ def get(self, bucket, key , version_id) -> Dict[str, Any]: def delete_by_name(self, bucket_name, version_id, key) -> None: """Delete a specified file or directory by name in a bucket and version.""" file_name, file_extension = os.path.splitext(key) - file_name_path = f"{file_name}{file_extension}" + file_name_path = f"{file_name}.v{version_id}{file_extension}" file_path = os.path.join(self.server_path, bucket_name, file_name_path) if os.path.exists(file_path): diff --git a/Storage/KT_Storage/DataAccess/VersionManager.py b/Storage/KT_Storage/DataAccess/VersionManager.py index dfccf931..d7aca55c 100644 --- a/Storage/KT_Storage/DataAccess/VersionManager.py +++ b/Storage/KT_Storage/DataAccess/VersionManager.py @@ -1,10 +1,16 @@ -from DataAccess import StorageManager +# from DataAccess import StorageManager from typing import Dict, Any import json import aiofiles import os -from Models.VesionModel import Version +import sys +import os +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '.'))) + +from ..Models.VesionModel import Version from .StorageManager import StorageManager class VersionManager: def __init__(self, metadata_file="s3 project/KT_Cloud/Storage/server/metadata.json"): diff --git a/Storage/KT_Storage/Models/VesionModel.py b/Storage/KT_Storage/Models/VesionModel.py index 6bbf2ad5..5cdb8b00 100644 --- a/Storage/KT_Storage/Models/VesionModel.py +++ b/Storage/KT_Storage/Models/VesionModel.py @@ -1,7 +1,7 @@ from datetime import datetime import hashlib -from AclModel import Acl -from Tag import Tag +from .AclModel import Acl +from .Tag import Tag # domain/versioning.py