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/Controller/DBInstanceNaiveController.py b/DB/NEW_KT_DB/Controller/DBInstanceNaiveController.py deleted file mode 100644 index e2dd70ec..00000000 --- a/DB/NEW_KT_DB/Controller/DBInstanceNaiveController.py +++ /dev/null @@ -1,46 +0,0 @@ -import datetime -from typing import Optional, Dict -from Service.Classes.DBInstanceNaiveService import DBInstanceService - -class DBInstanceController: - - def __init__(self, service: DBInstanceService): - self.service = service - - def create_db_instance(self, **kwargs): - """ - Create a new DBInstance by passing the necessary attributes to the service. - - Params: kwargs: The required and optional attributes for creating a DBInstance. - Return: A dictionary containing the newly created DBInstance. - """ - return self.service.create(**kwargs) - - def delete_db_instance(self,db_instance_identifier,skip_final_snapshot=False,final_db_snapshot_identifier=None,delete_automated_backups=False): - """ - Delete a DBInstance by its identifier with options to handle final snapshots and automated backups. - - Params: db_instance_identifier: The primary key (ID) of the DBInstance to delete. - skip_final_snapshot: If True, skip creating a final snapshot before deletion. Defaults to False. - final_db_snapshot_identifier: If skip_final_snapshot is False, specify an identifier for the final DB snapshot. - delete_automated_backups: If True, delete automated backups along with the DBInstance. Defaults to False. - """ - self.service.delete(db_instance_identifier,skip_final_snapshot,final_db_snapshot_identifier,delete_automated_backups) - - def modify_db_instance(self, **kwargs): - """ - Modify an existing DBInstance by passing updated attributes. - - Params: kwargs: The attributes to modify for the DBInstance. - Return: A dictionary containing the modified DBInstance. - """ - return self.service.modify(**kwargs) - - def describe_db_instance(self, db_instance_identifier: str): - """ - Retrieve details of a DBInstance by its identifier. - - Params: db_instance_identifier: The primary key (ID) of the DBInstance to describe. - Return: A dictionary containing the details of the DBInstance. - """ - return self.service.describe(db_instance_identifier) diff --git a/DB/NEW_KT_DB/Controller/DBSnapshotController.py b/DB/NEW_KT_DB/Controller/DBSnapshotController.py deleted file mode 100644 index 83d1d6dc..00000000 --- a/DB/NEW_KT_DB/Controller/DBSnapshotController.py +++ /dev/null @@ -1,143 +0,0 @@ -""" -DBSnapshotController Module ---------------------------- - -This module defines the `DBSnapshotController` class, which provides a high-level interface for managing -database snapshots associated with a specific database instance. - -The controller works by interacting with a `DBInstanceService` object, allowing users to perform -operations such as: - -- Creating snapshots -- Deleting snapshots -- Restoring a snapshot to a DB instance -- Listing all available snapshots for a DB instance -- Describing details of a specific snapshot -- Modifying snapshot attributes - -### Classes: - - DBSnapshotController: A controller class for managing DB snapshots. - -### Example Usage: - # Initialize the service and controller - db_instance_service = DBInstanceService() - snapshot_controller = DBSnapshotController(db_instance_service) - - # Create a new snapshot - snapshot_controller.create_snapshot('my-db-instance', 'my-snapshot') - - # Delete a snapshot - snapshot_controller.delete_snapshot('my-db-instance', 'my-snapshot') - - # Restore a snapshot - snapshot_controller.restore_snapshot('my-db-instance', 'my-snapshot') - - # List all snapshots - snapshots = snapshot_controller.list_snapshots('my-db-instance') - - # Describe a specific snapshot - snapshot_details = snapshot_controller.describe_snapshot('my-db-instance', 'my-snapshot') - - # Modify a snapshot - snapshot_controller.modify_snapshot('my-db-instance', 'my-snapshot', new_name='updated-snapshot') - -### Dependencies: - - DBInstanceService: A service class that provides lower-level operations for database instances and snapshots. - -""" - -from DB.NEW_KT_DB.Service.Classes.DBInstanceService import DBInstanceService - -class DBSnapshotController: - """ - This class provides control over database snapshots for a specific DB instance. - It uses the DBInstanceService to perform operations like create, delete, restore, and manage snapshots. - """ - - def __init__(self, db_instance_service: DBInstanceService): - """ - Initialize the DBSnapshotController with a DBInstanceService object. - - Args: - db_instance_service (DBInstanceService): The service responsible for managing DB instances and snapshots. - """ - self.db_instance_service = db_instance_service - - def create_snapshot(self, db_instance_identifier: str, db_snapshot_identifier: str): - """ - Create a snapshot for a given DB instance. - - Args: - db_instance_identifier (str): The identifier of the DB instance for which the snapshot will be created. - db_snapshot_identifier (str): The unique identifier for the snapshot to be created. - - Returns: - The result of the snapshot creation from the DBInstanceService. - """ - return self.db_instance_service.create_snapshot(db_instance_identifier, db_snapshot_identifier) - - def delete_snapshot(self, db_instance_identifier: str, db_snapshot_identifier: str): - """ - Delete a specific snapshot for a given DB instance. - - Args: - db_instance_identifier (str): The identifier of the DB instance that has the snapshot. - db_snapshot_identifier (str): The unique identifier of the snapshot to be deleted. - - Returns: - The result of the snapshot deletion from the DBInstanceService. - """ - return self.db_instance_service.delete_snapshot(db_instance_identifier, db_snapshot_identifier) - - def restore_snapshot(self, db_instance_identifier: str, db_snapshot_identifier: str): - """ - Restore a specific snapshot for a given DB instance. - - Args: - db_instance_identifier (str): The identifier of the DB instance to which the snapshot will be restored. - db_snapshot_identifier (str): The unique identifier of the snapshot to restore. - - Returns: - The result of the snapshot restoration from the DBInstanceService. - """ - return self.db_instance_service.restore_version(db_instance_identifier, db_snapshot_identifier) - - def list_snapshots(self, db_instance_identifier: str): - """ - List all snapshots associated with a specific DB instance. - - Args: - db_instance_identifier (str): The identifier of the DB instance whose snapshots will be listed. - - Returns: - List of snapshot identifiers for the given DB instance. - """ - db_instance = self.db_instance_service.get(db_instance_identifier) - return list(db_instance._node_subSnapshot_name_to_id.keys()) - - def describe_snapshot(self, db_instance_identifier: str, db_snapshot_identifier: str): - """ - Retrieve details for a specific snapshot of a DB instance. - - Args: - db_instance_identifier (str): The identifier of the DB instance containing the snapshot. - db_snapshot_identifier (str): The unique identifier of the snapshot to describe. - - Returns: - The details of the specified snapshot from the DBInstanceService. - """ - return self.db_instance_service.describe_snapshot(db_instance_identifier, db_snapshot_identifier) - - def modify_snapshot(self, db_instance_identifier: str, db_snapshot_identifier: str, **kwargs): - """ - Modify a specific snapshot for a given DB instance with provided attributes. - - Args: - db_instance_identifier (str): The identifier of the DB instance containing the snapshot. - db_snapshot_identifier (str): The unique identifier of the snapshot to be modified. - **kwargs: Additional attributes to modify for the snapshot. - - Returns: - The result of the snapshot modification from the DBInstanceService. - """ - return self.db_instance_service.modify_snapshot(db_instance_identifier, db_snapshot_identifier, **kwargs) diff --git a/DB/NEW_KT_DB/Controller/EventSubscriptionController.py b/DB/NEW_KT_DB/Controller/EventSubscriptionController.py deleted file mode 100644 index cdc8051d..00000000 --- a/DB/NEW_KT_DB/Controller/EventSubscriptionController.py +++ /dev/null @@ -1,75 +0,0 @@ -from typing import List, Tuple -from DB.NEW_KT_DB.Models.EventSubscriptionModel import EventCategory, EventSubscription, SourceType -from Service.Classes.EventSubscriptionService import EventSubscriptionService - - -class EventSubscriptionController: - """ - Controller class for managing event subscriptions. - """ - - def __init__(self, service: EventSubscriptionService) -> None: - """ - Initialize the EventSubscriptionController. - - Args: - service (EventSubscriptionService): The service to handle event subscription operations. - """ - self.service = service - - def create_event_subscription(self, subscription_name: str, sources: List[Tuple[SourceType, str]], - event_categories: List[EventCategory], sns_topic_arn: str, source_type: SourceType = SourceType.All) -> None: - """ - Create a new event subscription. - - Args: - subscription_name (str): The name of the subscription. - sources (List[Tuple[SourceType, str]]): List of source types and their identifiers. - event_categories (List[EventCategory]): List of event categories to subscribe to. - sns_topic_arn (str): The ARN of the SNS topic for notifications. - source_type (SourceType, optional): The type of source. Defaults to SourceType.All. - """ - self.service.create(subscription_name=subscription_name, sources=sources, - event_categories=event_categories, sns_topic_arn=sns_topic_arn, source_type=source_type) - - def delete_event_subscription(self, subscription_name: str): - """ - Delete an event subscription. - - Args: - subscription_name (str): The name of the subscription to delete. - """ - self.service.delete(subscription_name=subscription_name) - - def describe_event_subscriptions(self, columns = None, criteria = None) -> None: - """ - Describe event subscriptions. - - Args: - marker (str): The pagination token for the next set of results. - max_records (int, optional): The maximum number of records to return. Defaults to 100. - subscription_name (str, optional): The name of a specific subscription to describe. Defaults to ''. - """ - self.service.describe(columns, criteria) - - def modify_event_subscription(self, subscription_name: str, event_categories: List[EventCategory], sns_topic_arn: str, source_type: SourceType = SourceType.ALL) -> None: - """ - Modify an existing event subscription. - - Args: - subscription_name (str): The name of the subscription to modify. - event_categories (List[EventCategory]): Updated list of event categories to subscribe to. - sns_topic_arn (str): Updated ARN of the SNS topic for notifications. - source_type (SourceType, optional): Updated type of source. Defaults to SourceType.ALL. - """ - self.service.modify(subscription_name=subscription_name, - event_categories=event_categories, sns_topic_arn=sns_topic_arn, source_type=source_type) - - def get(self) -> List[EventSubscription]: - """ - Retrieve a list of all event subscriptions. - - Returns: - List[EventSubscription]: A list of EventSubscription objects representing all current event subscriptions. - """ - return self.service.get() diff --git a/DB/NEW_KT_DB/DataAccess/DBClusterManager.py b/DB/NEW_KT_DB/DataAccess/DBClusterManager.py index ef1863bc..4e375552 100644 --- a/DB/NEW_KT_DB/DataAccess/DBClusterManager.py +++ b/DB/NEW_KT_DB/DataAccess/DBClusterManager.py @@ -1,8 +1,8 @@ from typing import Dict, Any import json import sqlite3 -from DataAccess import ObjectManager -from Models.DBClusterModel import Cluster +from NEW_KT_DB.DataAccess import ObjectManager +from NEW_KT_DB.Models.DBClusterModel import Cluster from typing import Optional class DBClusterManager: diff --git a/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py b/DB/NEW_KT_DB/DataAccess/DBClusterParameterGroupManager.py new file mode 100644 index 00000000..5cb163c3 --- /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, pk_column_data_type='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/DBInstanceManager.py b/DB/NEW_KT_DB/DataAccess/DBInstanceManager.py deleted file mode 100644 index ddadccb1..00000000 --- a/DB/NEW_KT_DB/DataAccess/DBInstanceManager.py +++ /dev/null @@ -1,107 +0,0 @@ -""" -DBInstanceManager Module ------------------------- - -This module provides the `DBInstanceManager` class, which manages `DBInstanceModel` objects in memory using an `ObjectManager`. -The manager handles operations such as creating, modifying, describing, and deleting DBInstance records stored in memory. - -### Classes: - - DBInstanceManager: A class for managing in-memory database instances (`DBInstanceModel`), using JSON serialization for object data storage. - -### Methods: - - `__init__(db_file: str)`: Initializes the `DBInstanceManager` with a database file and creates the management table. - - `close_connections()`: Closes any open database connections. - - `createInMemoryDBInstance(db_instance)`: Stores a `DBInstanceModel` object in memory by serializing it as JSON. - - `modifyDBInstance(db_instance)`: Modifies an existing `DBInstanceModel` in memory by updating its metadata. - - `describeDBInstance(db_instance_identifier) -> Dict[str, Any]`: Retrieves a `DBInstanceModel` object by its identifier and returns its metadata. - - `deleteInMemoryDBInstance(db_instance_identifier)`: Deletes a `DBInstanceModel` from memory using its identifier. - - `getDBInstance(db_instance_identifier)`: Fetches a `DBInstanceModel` object by its identifier. - - `is_db_instance_exists(db_instance_identifier) -> bool`: Checks if a `DBInstanceModel` exists in memory by its identifier. - -### Example Usage: - db_instance_manager = DBInstanceManager('db_file.db') - - # Create a DBInstance in memory - db_instance = DBInstanceModel(db_instance_identifier="db1", status="available") - db_instance_manager.createInMemoryDBInstance(db_instance) - - # Modify a DBInstance - db_instance.status = "stopped" - db_instance_manager.modifyDBInstance(db_instance) - - # Describe a DBInstance - metadata = db_instance_manager.describeDBInstance("db1") - - # Check if a DBInstance exists - exists = db_instance_manager.is_db_instance_exists("db1") - - # Delete a DBInstance - db_instance_manager.deleteInMemoryDBInstance("db1") - -### Dependencies: - - ObjectManager: A class responsible for low-level management of data in memory. - - DBInstanceModel: A model class representing a database instance, serialized as JSON for storage. - -""" - - -from typing import Dict, Any -import json -from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager -from DB.NEW_KT_DB.Models.DBInstanceModel import DBInstanceModel - - -class DBInstanceManager: - object_name = __name__.split('.')[-1].replace('Manager', '').lower() - - def __init__(self, db_file: str): - self.object_manager = ObjectManager(db_file) - self.object_manager.create_management_table( - self.object_name, DBInstanceModel.table_structure, pk_column_data_type='TEXT') - - - def createInMemoryDBInstance(self, db_instance): - metadata = json.dumps(db_instance.to_dict()) - data = (db_instance.db_instance_identifier, metadata) - self.object_manager.save_in_memory(self.object_name, data) - - def modifyDBInstance(self, db_instance): - metadata = json.dumps(db_instance.to_dict()) - updates = f"metadata = '{metadata}'" - criteria = f"db_instance_identifier = '{ - db_instance.db_instance_identifier}'" - self.object_manager.update_in_memory( - self.object_name, updates, criteria) - - def describeDBInstance(self, db_instance_identifier) -> Dict[str, Any]: - criteria = f"db_instance_identifier = '{db_instance_identifier}'" - result = self.object_manager.get_from_memory( - self.object_name, "*", criteria) - - if result: - metadata = json.loads(result[0][1]) - return metadata - else: - raise ValueError(f"DB Instance with identifier { - db_instance_identifier} not found.") - - def deleteInMemoryDBInstance(self, db_instance_identifier): - criteria = f"db_instance_identifier = '{db_instance_identifier}'" - self.object_manager.delete_from_memory_by_criteria( - self.object_name, criteria) - - def getDBInstance(self, db_instance_identifier): - criteria = f"db_instance_identifier = '{db_instance_identifier}'" - # result = self.object_manager.get_from_memory(self.object_name, ["*"], criteria) - result = self.object_manager.get_from_memory( - self.object_name, "*", criteria) - - return result - - def isDbInstanceExists(self, db_instance_identifier): - try: - self.object_manager.get_from_memory( - self.object_name, db_instance_identifier) - return True - except ValueError: - return False diff --git a/DB/NEW_KT_DB/DataAccess/DBInstanceNaiveManager.py b/DB/NEW_KT_DB/DataAccess/DBInstanceNaiveManager.py deleted file mode 100644 index 3aa8efbb..00000000 --- a/DB/NEW_KT_DB/DataAccess/DBInstanceNaiveManager.py +++ /dev/null @@ -1,76 +0,0 @@ -import os -import sys -from typing import Dict, Any, List -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 DataAccess.ObjectManager import ObjectManager -from Models.DBInstanceNaiveModel import DBInstance - -class DBInstanceManager: - def __init__(self, object_manager: ObjectManager): - self.object_manager = object_manager - # Create the management table for DBInstance using its object name and table structure - self.object_manager.create_management_table(DBInstance.object_name, DBInstance.table_structure) - - def createInMemoryDBInstance(self, db_instance: DBInstance): - """ - Create a new DBInstance in memory. - - Params db_instance: DBInstance object to be saved in memory. - """ - self.object_manager.save_in_memory(DBInstance.object_name, db_instance.to_sql()) - - def deleteInMemoryDBInstance(self, db_instance_identifier: str): - """ - Delete a DBInstance from memory by its identifier. - - Params: db_instance_identifier: The primary key (ID) of the DBInstance to delete. - """ - self.object_manager.delete_from_memory_by_pk( - pk_column=DBInstance.pk_column, - pk_value=db_instance_identifier, - object_name=DBInstance.object_name - ) - - def describeDBInstance(self, db_instance_identifier: str): - """ - Retrieve the details of a DBInstance based on its identifier. - - Params: db_instance_identifier: The primary key (ID) of the DBInstance to describe. - - Return: List of DBInstance attributes matching the criteria. - """ - # Fetch DBInstance from memory using its primary key - return self.object_manager.get_from_memory( - criteria=f"{DBInstance.pk_column} = '{db_instance_identifier}'", - object_name=DBInstance.object_name, - columns='*' - ) - - def modifyDBInstance(self, db_instance_identifier: str, updates: str): - """ - Modify the attributes of an existing DBInstance in memory. - - Params: db_instance_identifier: The primary key (ID) of the DBInstance to modify. - updates: SQL-like string containing the updates to be applied (e.g., "port = '3306'"). - """ - # Update the instance's attributes based on the provided update string - self.object_manager.update_in_memory( - criteria=f"{DBInstance.pk_column} = '{db_instance_identifier}'", - object_name=DBInstance.object_name, - updates=updates - ) - - def is_db_instance_exist(self, db_instance_identifier: int) -> bool: - """ - Check if a DBInstance with the given identifier exists in memory. - - Params: db_instance_identifier: The primary key (ID) of the DBInstance to check. - - Return: True if the DBInstance exists, otherwise False. - """ - # Check if the object exists by its primary key in the management table - return bool(self.object_manager.db_manager.is_object_exist( - self.object_manager._convert_object_name_to_management_table_name(DBInstance.object_name), - criteria=f"{DBInstance.pk_column} = '{db_instance_identifier}'" - )) diff --git a/DB/NEW_KT_DB/DataAccess/DBSubnetGroupManager.py b/DB/NEW_KT_DB/DataAccess/DBSubnetGroupManager.py deleted file mode 100644 index b065e821..00000000 --- a/DB/NEW_KT_DB/DataAccess/DBSubnetGroupManager.py +++ /dev/null @@ -1,63 +0,0 @@ -from typing import Dict, Any, List - -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__), ".."))) - -from DataAccess import ObjectManager -from Models.DBSubnetGroupModel import DBSubnetGroup -import Exceptions.DBSubnetGroupExceptions as DBSubnetGroupExceptions - -class DBSubnetGroupManager: - def __init__(self, object_manager: ObjectManager): - self.object_manager = object_manager - self.object_manager.create_management_table( - DBSubnetGroup.object_name, DBSubnetGroup.table_structure - ) - - def create(self, subnet_group: DBSubnetGroup): - self.object_manager.save_in_memory( - DBSubnetGroup.object_name, subnet_group.to_sql_insert() - ) - - def get(self, name: str): - data = self.object_manager.get_from_memory( - DBSubnetGroup.object_name, criteria=f"{DBSubnetGroup.pk_column} = '{name}'" - ) - if data: - return DBSubnetGroup(*data[0]) - else: - raise DBSubnetGroupExceptions.DBSubnetGroupNotFound(f"subnet group with name '{name}' not found") - - def delete(self, name: str): - exists = self.object_manager.get_from_memory( - DBSubnetGroup.object_name, criteria=f"{DBSubnetGroup.pk_column} = '{name}'" - ) - # when no result were found, exists is an empty list and therefore if exists will result in false - if exists: - self.object_manager.delete_from_memory_by_pk( - DBSubnetGroup.object_name, DBSubnetGroup.pk_column, name - ) - else: - raise DBSubnetGroupExceptions.DBSubnetGroupNotFound(f"subnet group with name '{name}' not found") - - - def describe(self, name: str): - # get the object from memory and return it in dictionary form - return self.get(name).to_dict() - - def modify(self, subnet_group: DBSubnetGroup): - # get the updates in a format that is good for an sql update query - updates = subnet_group.to_sql_update() - # use the object manager to update the object in memory, with the criteria the the primary key equals the object id - self.object_manager.update_in_memory( - DBSubnetGroup.object_name, - updates, - criteria=f"{DBSubnetGroup.pk_column} = '{subnet_group.db_subnet_group_name}'", - ) - - def list_db_subnet_groups(self): - results = self.object_manager.get_from_memory(DBSubnetGroup.object_name) - return [DBSubnetGroup(*result) for result in results] diff --git a/DB/NEW_KT_DB/DataAccess/EventSubscriptionManager.py b/DB/NEW_KT_DB/DataAccess/EventSubscriptionManager.py deleted file mode 100644 index fad5a6e3..00000000 --- a/DB/NEW_KT_DB/DataAccess/EventSubscriptionManager.py +++ /dev/null @@ -1,169 +0,0 @@ -from typing import Dict, Any, List, Optional, Tuple -import json -from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager -from DB.NEW_KT_DB.Models.EventSubscriptionModel import EventCategory, EventSubscription, SourceType - - -class EventSubscriptionManager: - """ - Manages event subscriptions in the database. - """ - - def __init__(self, db_file: str): - """ - Initialize EventSubscriptionManager with the database connection. - - Args: - db_file (str): Path to the SQLite database file. - """ - self.object_manager = ObjectManager(db_file) - self.object_manager.create_management_table( - EventSubscription.get_object_name(), EventSubscription.table_structure, pk_column_data_type='TEXT') - - def createInMemoryEventSubscription(self, event_subscription: EventSubscription) -> None: - """ - Create a new event subscription in memory. - - Args: - event_subscription (EventSubscription): The event subscription to create. - """ - self.object_manager.save_in_memory( - event_subscription.get_object_name(), event_subscription.to_sql()) - - def deleteInMemoryEventSubscription(self, subscription_name: str) -> None: - """ - Delete an event subscription from memory. - - Args: - subscription_name (str): The name of the subscription to delete. - """ - self.object_manager.delete_from_memory_by_pk( - EventSubscription.get_object_name(), EventSubscription.pk_column, subscription_name) - - def describeEventSubscriptionById(self, subscription_name: str) -> Dict: - """ - Retrieve an event subscription by its ID (name). - - Args: - subscription_name (str): The name of the subscription to retrieve. - - Returns: - Dict: A dictionary representation of the event subscription, or None if not found. - """ - event_subscription = self.object_manager.get_from_memory( - EventSubscription.get_object_name(), - criteria=f'{EventSubscription.pk_column} = "{subscription_name}"' - ) - if not event_subscription: - return None - return EventSubscription.values_to_dict(*event_subscription[0]) - - def modifyEventSubscription(self, event_subscription: EventSubscription) -> None: - """ - Modify an existing event subscription in memory. - - Args: - event_subscription (EventSubscription): The updated event subscription. - """ - subscription_dict = event_subscription.to_dict() - updates = [] - for key, value in subscription_dict.items(): - if isinstance(value, (dict, list)): - updates.append(f"{key} = '{json.dumps(value)}'") - elif isinstance(value, str): - updates.append(f"{key} = '{value}'") - else: - updates.append(f"{key} = {repr(value)}") - - updates = ", ".join(updates) - self.object_manager.update_in_memory( - EventSubscription.get_object_name(), - updates, - f'{EventSubscription.pk_column} = "{event_subscription.pk_value}"' - ) - - def describeEventSubscriptionByCriteria(self, columns: Optional[List[str]] = '*', criteria: Dict[str, Any] = None) -> List[Dict]: - """ - Retrieve event subscriptions based on specified criteria. - - Args: - columns (Optional[List[str]]): List of columns to retrieve. Defaults to all columns. - criteria (Dict[str, Any]): Criteria for filtering event subscriptions. - - Returns: - List[Dict]: A list of dictionaries representing the matching event subscriptions. - """ - if criteria: - key, value = next(iter(criteria.items())) - criteria = f'{key} = "{value}"' - event_subscription_data = self.object_manager.get_from_memory( - EventSubscription.get_object_name(), columns, criteria - ) - return [EventSubscription.values_to_dict(*event_subscription) for event_subscription in event_subscription_data] - - def get(self, criteria: Dict[str, Any] = None) -> List[EventSubscription]: - """ - Retrieve event subscriptions based on specified criteria. - - Args: - criteria (Dict[str, Any]): Criteria for filtering event subscriptions. - - Returns: - List[EventSubscription]: A list of EventSubscription objects matching the criteria. - """ - if criteria: - key, value = criteria.popitem() - criteria = f'{key} = "{value}"' - event_subscriptions_data = self.object_manager.get_from_memory( - EventSubscription.get_object_name(), criteria=criteria) - - return [EventSubscriptionManager.sql_to_object(event_subscription) for event_subscription in event_subscriptions_data] - - def get_by_id(self, subscription_name: str) -> EventSubscription: - """ - Retrieve an event subscription by its ID (name). - - Args: - subscription_name (str): The name of the subscription to retrieve. - - Returns: - EventSubscription: The EventSubscription object, or None if not found. - """ - event_subscriptions_data = self.object_manager.get_from_memory( - EventSubscription.get_object_name(), criteria=f'{EventSubscription.pk_column} = "{subscription_name}"') - - if not event_subscriptions_data: - return None - - return EventSubscriptionManager.sql_to_object(event_subscriptions_data[0]) - - @staticmethod - def sql_to_object(sql_subscription: Tuple[str]) -> EventSubscription: - """ - Convert SQL data to an EventSubscription object. - - Args: - sql_subscription (Tuple[str]): SQL data representing an event subscription. - - Returns: - EventSubscription: The converted EventSubscription object. - """ - subscription_name, sources, source_type, event_categories, sns_topic_arn = sql_subscription - sources = json.loads(sources) - event_categories = json.loads(event_categories) - - sources_list = [(SourceType(source_type), source_id) for source_type, - source_ids in sources.items() for source_id in source_ids] - - event_categories_list = [EventCategory( - category) for category in event_categories] - - event_subscription = EventSubscription( - subscription_name=subscription_name, - sources=sources_list, - event_categories=event_categories_list, - sns_topic_arn=sns_topic_arn, - source_type=SourceType(source_type) - ) - - return event_subscription diff --git a/DB/NEW_KT_DB/DataAccess/ObjectManager.py b/DB/NEW_KT_DB/DataAccess/ObjectManager.py index 58dcc4cb..2d54891e 100644 --- a/DB/NEW_KT_DB/DataAccess/ObjectManager.py +++ b/DB/NEW_KT_DB/DataAccess/ObjectManager.py @@ -2,7 +2,7 @@ import json import sqlite3 -from DB.NEW_KT_DB.DataAccess.DBManager import DBManager +from NEW_KT_DB.DataAccess.DBManager import DBManager class ObjectManager: def __init__(self, db_file: str): @@ -11,7 +11,6 @@ def __init__(self, db_file: str): def create_management_table(self, object_name, table_structure='default', pk_column_data_type='INTEGER'): - table_name = self._convert_object_name_to_management_table_name(object_name) pk_constraint = ' AUTOINCREMENT' if pk_column_data_type == 'INTEGER' else '' @@ -19,7 +18,6 @@ def create_management_table(self, object_name, table_structure='default', pk_col table_structure = f'object_id {pk_column_data_type} PRIMARY KEY {pk_constraint},type_object TEXT NOT NULL,metadata TEXT NOT NULL' self.db_manager.create_table(table_name, table_structure) - def _insert_object_to_management_table(self, table_name, object_info, columns_to_populate=None): if columns_to_populate is None: @@ -108,3 +106,14 @@ def convert_object_attributes_to_dictionary(**kwargs): for key, value in kwargs.items(): dict[key] = value return dict + + def is_exists(self, object): + table_name = self._convert_object_name_to_management_table_name(object.object_name) + try: + query=f'select * from {table_name} where {object.pk_column} = {object.pk_value}' + result=self.db_manager.execute_query_with_single_result(query) + if result is None: + return False + return True + except sqlite3.OperationalError as e: + 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..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/Models/DBInstanceModel.py b/DB/NEW_KT_DB/Models/DBInstanceModel.py deleted file mode 100644 index c181caf9..00000000 --- a/DB/NEW_KT_DB/Models/DBInstanceModel.py +++ /dev/null @@ -1,221 +0,0 @@ -""" -DBInstanceModel - -This class represents the model for a database instance. It encapsulates the attributes -and behavior of a database instance, including its configuration, state, and associated snapshots. - -The model includes validation for various attributes and manages the versioning of the database -through a system of nodes and snapshots. - -Attributes: - db_instance_identifier: Unique identifier for the database instance. - allocated_storage: Amount of storage allocated to the instance. - master_username: Username for the master user of the database. - master_user_password: Password for the master user. - port: Port number on which the database instance accepts connections. - status: Current status of the database instance. - created_time: Timestamp of when the instance was created. - endpoint: File system path where the instance data is stored. - _node_subSnapshot_dic: Dictionary of snapshot nodes. - _node_subSnapshot_name_to_id: Mapping of snapshot names to their IDs. - _current_version_ids_queue: Queue of snapshot IDs representing the current version chain. - -Methods: - to_dict: Convert the instance attributes to a dictionary. - -The class also includes nested Node_SubSnapshot class for managing individual snapshots. -""" - -from datetime import datetime -from collections import deque -import os -from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager -import uuid -from DB.NEW_KT_DB.Validation.DBInstanceValiditions import validate_allocated_storage, validate_master_user_name, validate_master_user_password, validate_port, validate_status -from DB.NEW_KT_DB.Validation.GeneralValidations import is_valid_db_instance_identifier - -class DBInstanceModel: - BASE_PATH = "db_instances" - table_structure = f''' - db_instance_identifier TEXT PRIMARY KEY, - metadata TEXT NOT NULL - ''' - def __init__(self, **kwargs): - - # Validate and set db_instance_identifier - if is_valid_db_instance_identifier(kwargs.get('db_instance_identifier'), 30): - self.db_instance_identifier = kwargs['db_instance_identifier'] - else: - raise ValueError("Invalid DB Instance Identifier") - - # Validate and set allocated_storage - allocated_storage = kwargs.get('allocated_storage', 20) # Default value is 20 - validate_allocated_storage(allocated_storage) - self.allocated_storage = allocated_storage - - # Validate and set master_user_name - master_user_name = kwargs.get('master_user_name', 'admin') # Default value - validate_master_user_name(master_user_name) - self.master_username = master_user_name - - # Validate and set master_user_password - master_user_password = kwargs.get('master_user_password', 'default_password') # Default value - validate_master_user_password(master_user_password) - self.master_user_password = master_user_password - - # Validate and set port - port = kwargs.get('port', 3306) # Default value is 3306 - validate_port(port) - self.port = port - - # Validate and set status - status = kwargs.get('status', 'available') # Default value is 'available' - validate_status(status) - self.status = status - - # Set created_time (no validation required) - self.created_time = kwargs.get('created_time', datetime.now()) - self.endpoint = os.path.join( - DBInstanceModel.BASE_PATH, self.db_instance_identifier) - - self._node_subSnapshot_dic = kwargs.get('_node_subSnapshot_dic', {}) - self._node_subSnapshot_name_to_id = kwargs.get('_node_subSnapshot_name_to_id', {}) - - if '_current_version_ids_queue' in kwargs: - self._current_version_ids_queue = deque(kwargs['_current_version_ids_queue']) - else: - first_node = Node_SubSnapshot(parent_id=None, endpoint=self.endpoint) - self._node_subSnapshot_dic[first_node.id_snapshot] = first_node - self._current_version_ids_queue = deque([first_node.id_snapshot]) - - self._last_node_of_current_version = self._node_subSnapshot_dic.get(self._current_version_ids_queue[-1]) - - - def to_dict(self): - return ObjectManager.convert_object_attributes_to_dictionary( - db_instance_identifier=self.db_instance_identifier, - allocated_storage=self.allocated_storage, - master_username=self.master_username, - master_user_password=self.master_user_password, - port=self.port, - status=self.status, - created_time=self.created_time.isoformat() if self.created_time is not None else None, - endpoint=self.endpoint, - - node_subSnapshot_dic={str(k): v.to_dict() for k, v in self._node_subSnapshot_dic.items()}, - node_subSnapshot_name_to_id=self._node_subSnapshot_name_to_id, - current_version_ids_queue=[str(id_snapshot) for id_snapshot in self._current_version_ids_queue] - ) - - -class Node_SubSnapshot: - """ - A class representing a snapshot in a versioning system, used to store database schemas and deleted records. - Each snapshot can have a parent, and a new snapshot can be created by cloning the parent's database schema. - - Attributes: - id_snapshot (uuid): A unique identifier for the snapshot. Defaults to a new UUID if not provided. - parent_id (uuid): The identifier of the parent snapshot, if applicable. - dbs_paths_dic (dict): A dictionary mapping database names to their file paths. - deleted_records_db_path (str): The path where deleted records are stored for the snapshot. - snapshot_type (str): The type of snapshot (e.g., "manual"). - created_time (datetime): The time when the snapshot was created. - """ - - def __init__(self, parent=None, endpoint=None, **kwargs): - """ - Initialize a new Node_SubSnapshot instance. If a parent snapshot is provided, - the databases are cloned from the parent, and paths are set for the current snapshot. - - Args: - parent (Node_SubSnapshot): The parent snapshot to clone from (optional). - endpoint (str): The base directory where the snapshot data will be stored. - **kwargs: Optional parameters including: - - id_snapshot (uuid): An optional unique identifier for the snapshot. - - dbs_paths_dic (dict): An optional dictionary of database paths. - - deleted_records_db_path (str): An optional path for storing deleted records. - """ - self.id_snapshot = kwargs.get('id_snapshot', uuid.uuid4()) # Assign a new UUID if not provided. - self.snapshot_type = "manual" # Snapshot type defaults to "manual". - self.parent_id = parent.id_snapshot if parent else None # Assign parent ID if a parent exists. - - self.created_time = None # Creation time is not set on initialization. - - # If there is a parent and no new dbs_paths_dic is provided, clone the parent's databases. - if parent and not kwargs.get('dbs_paths_dic'): - self.dbs_paths_dic = self.clone_databases_schema(parent.dbs_paths_dic, endpoint) - else: - self.dbs_paths_dic = kwargs.get('dbs_paths_dic', {}) - - # Create or retrieve the path for deleted records in this snapshot. - self.deleted_records_db_path = kwargs.get('deleted_records_db_path', self._create_deleted_records_db_path(endpoint)) - - def to_dict(self): - """ - Convert the Node_SubSnapshot instance to a dictionary, which can be used for storage or serialization. - - Returns: - dict: A dictionary containing the snapshot's attributes. - """ - return ObjectManager.convert_object_attributes_to_dictionary( - id_snapshot=str(self.id_snapshot), - parent_id=str(self.parent_id) if self.parent_id else None, - dbs_paths_dic=self.dbs_paths_dic, - deleted_records_db_path=self.deleted_records_db_path, - snapshot_type=self.snapshot_type, - created_time=self.created_time.isoformat() if self.created_time is not None else None, - ) - - def _create_deleted_records_db_path(self, endpoint): - """ - Create a path for storing deleted records in the snapshot. A directory is created - with the snapshot's ID, and a SQLite database is initialized within that directory. - - Args: - endpoint (str): The base directory where the snapshot data is stored. - - Returns: - str: The full path to the deleted records database. - """ - deleted_records_db_path = os.path.join(endpoint, str(self.id_snapshot)) # Create a folder for the snapshot. - os.makedirs(deleted_records_db_path, exist_ok=True) # Ensure the directory exists. - deleted_records_db_path = os.path.join(deleted_records_db_path, "deleted_db.db") # Define the path to the deleted records database. - return deleted_records_db_path - - def clone_databases_schema(self, dbs_paths_dic, endpoint): - """ - Clone the database schemas from the parent snapshot to create a new snapshot. This is done by copying - each database file to the new snapshot directory and cloning its schema. - - Args: - dbs_paths_dic (dict): A dictionary of database names and paths from the parent snapshot. - endpoint (str): The base directory where the cloned databases will be stored. - - Returns: - dict: A dictionary mapping the new database names to their cloned paths. - """ - from DB.NEW_KT_DB.Service.Classes.DBInstanceService import SQLCommandHelper - - dbs_paths_new_dic = {} # Dictionary to store the new database paths. - for db, db_path in dbs_paths_dic.items(): - db_filename = os.path.basename(db_path) # Get the database filename. - new_path = os.path.join(endpoint, str(self.id_snapshot), db_filename) # Create the new path for the cloned database. - directory = os.path.dirname(new_path) - os.makedirs(directory, exist_ok=True) # Ensure the directory for the new path exists. - SQLCommandHelper.clone_database_schema(db_path, new_path) # Clone the database schema. - dbs_paths_new_dic[db] = new_path # Store the new path in the dictionary. - return dbs_paths_new_dic - - def create_child(self, endpoint): - """ - Create a child snapshot from the current snapshot. The child will inherit the database schemas - from the parent and store its own changes separately. - - Args: - endpoint (str): The base directory where the child snapshot data will be stored. - - Returns: - Node_SubSnapshot: A new child snapshot object. - """ - child = Node_SubSnapshot(parent=self, endpoint=endpoint) # Create a child snapshot. - return child diff --git a/DB/NEW_KT_DB/Models/DBInstanceNaiveModel.py b/DB/NEW_KT_DB/Models/DBInstanceNaiveModel.py deleted file mode 100644 index 4f7d8eca..00000000 --- a/DB/NEW_KT_DB/Models/DBInstanceNaiveModel.py +++ /dev/null @@ -1,59 +0,0 @@ -import json -from datetime import datetime -import os -import sys -from DataAccess.ObjectManager import ObjectManager -sys.path.insert(0, os.path.abspath( - os.path.join(os.path.dirname(__file__), '../../../'))) -from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager - - -class DBInstance: - BASE_PATH = "db_instances" - object_name = 'db_instance_naive' - pk_column = 'db_instance_id' - pk_column_data_type = 'TEXT' - table_structure = 'db_instance_id VARCHAR(255) PRIMARY KEY NOT NULL, allocated_storage INT NOT NULL, master_username VARCHAR(255) NOT NULL, master_user_password VARCHAR(255) NOT NULL, db_name VARCHAR(255) NOT NULL, port INT NOT NULL, status VARCHAR(50) NOT NULL, created_time DATETIME NOT NULL, endpoint VARCHAR(255) NOT NULL, databases TEXT NOT NULL, pk_value VARCHAR(255) NOT NULL' - - def __init__(self, **kwargs): - """ - Initialize a new DBInstance with the given parameters. - """ - self.db_instance_identifier = kwargs['db_instance_identifier'] - self.allocated_storage = kwargs.get('allocated_storage', 20) - self.master_username = kwargs['master_username'] - self.master_user_password = kwargs['master_user_password'] - self.db_name = kwargs.get('db_name', None) - self.port = kwargs.get('port', 3306) - self.status = 'available' - self.created_time = datetime.now() - self.endpoint = self.db_instance_identifier - storageManager = StorageManager(DBInstance.BASE_PATH) - storageManager.create_directory(self.endpoint) - self.databases = kwargs.get('databases', {}) - self.pk_value = kwargs.get('pk_value', self.db_instance_identifier) - - def to_dict(self): - """Retrieve the metadata of the DB instance as a dictionary.""" - return ObjectManager.convert_object_attributes_to_dictionary( - db_instance_identifier=self.db_instance_identifier, - allocated_storage=self.allocated_storage, - master_username=self.master_username, - master_user_password=self.master_user_password, - db_name=self.db_name, - port=self.port, - status=self.status, - created_time=str(self.created_time), - endpoint=self.endpoint, - databases=self.databases, - pk_value=self.pk_value - - ) - - def to_sql(self): - # Convert the model instance to a dictionary - data_dict = self.to_dict() - values = '(' + ", ".join(f'\'{json.dumps(v)}\'' if isinstance(v, dict) or isinstance(v, list) else f'\'{v}\'' if isinstance(v, str) else f'\'{str(v)}\'' - for v in data_dict.values()) + ')' - # values='(\''+self.db_instance_identifier+'\',\''+json.dumps(self.to_dict())+'\')' - return values diff --git a/DB/NEW_KT_DB/Models/EventSubscriptionModel.py b/DB/NEW_KT_DB/Models/EventSubscriptionModel.py deleted file mode 100644 index 5ca2fa33..00000000 --- a/DB/NEW_KT_DB/Models/EventSubscriptionModel.py +++ /dev/null @@ -1,154 +0,0 @@ -from enum import Enum -import json -from typing import Dict, List, Tuple - -from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager - - -class SourceType(Enum): - """ - Enumeration of possible source types for event subscriptions. - """ - DB_INSTANCE = 'db-instance' - DB_CLUSTER = 'db-cluster' - DB_PARAMETER_GROUP = 'db-parameter-group' - DB_SECURITY_GROUP = 'db-security-group' - DB_SNAPSHOT = 'db-snapshot' - DB_CLUSTER_SNAPSHOT = 'db-cluster-snapshot' - DB_PROXY = 'db-proxy' - ZERO_ETL = 'zero-etl' - CUSTOM_ENGINE_VERSION = 'custom-engine-version' - BLUE_GREEN_DEPLOYMENT = 'blue-green-deployment' - ALL = 'all' - - -class EventCategory(Enum): - """ - Enumeration of possible event categories for event subscriptions. - """ - RECOVERY = 'recovery' - READ_REPLICA = 'read replica' - FAILURE = 'failure' - FAILOVER = 'failover' - DELETION = 'deletion' - CREATION = 'creation' - CONFIGURATION_CHANGE = 'configuration change' - BACKUP = 'backup' - - -class EventSubscription: - """ - Represents an event subscription in the database. - """ - - pk_column = 'subscription_name' - table_structure = """ - subscription_name TEXT PRIMARY KEY, - sources TEXT, - source_type TEXT, - event_categories TEXT, - sns_topic_arn TEXT""" - - def __init__( - self, - subscription_name: str, - sources: List[Tuple[SourceType, str]], - event_categories: List[EventCategory], - sns_topic_arn: str, - source_type: SourceType - ) -> None: - """ - Initialize an EventSubscription object. - - Args: - subscription_name (str): The name of the subscription. - sources (List[Tuple[SourceType, str]]): List of source types and their IDs. - event_categories (List[EventCategory]): List of event categories. - sns_topic_arn (str): The SNS topic to which notifications will be sent. - source_type (SourceType): The type of source for which notifications will be received. - """ - self.subscription_name = subscription_name - self.source_type = source_type - self.sources = {source_type.value: set() for source_type in SourceType} - - for source_type, source_id in sources: - self.sources[source_type.value].add(source_id) - - self.event_categories = event_categories - self.sns_topic_arn = sns_topic_arn - - self.pk_value = self.subscription_name - - def __eq__(self, value: object) -> bool: - """ - Compare two EventSubscription objects for equality. - """ - if not isinstance(value, EventSubscription): - return False - return all([self.__getattribute__(attr) == value.__getattribute__(attr) for attr, _ in self.__dict__.items()]) and len(self.__dict__) == len(value.__dict__) - - def to_dict(self) -> Dict: - """ - Convert the EventSubscription object to a dictionary. - - Returns: - Dict: A dictionary representation of the EventSubscription. - """ - return ObjectManager.convert_object_attributes_to_dictionary( - subscription_name=self.subscription_name, - sources={ - k: list(v) for k, v in self.sources.items()}, - source_type=self.source_type.value, - event_categories=[ - ec.value for ec in self.event_categories], - sns_topic_arn=self.sns_topic_arn) - - def to_sql(self) -> str: - """ - Convert the EventSubscription object to an SQL insert statement. - - Returns: - str: A string representation of the SQL insert statement. - """ - data = self.to_dict() - values = [ - f"'{data['subscription_name']}'", - f"'{json.dumps(data['sources'])}'", - f"'{data['source_type']}'", - f"'{json.dumps(data['event_categories'])}'", - f"'{data['sns_topic_arn']}'" - ] - return f"({', '.join(values)})" - - @staticmethod - def get_object_name() -> str: - """ - Get the name of the object. - - Returns: - str: The name of the object without the 'Model' suffix. - """ - return __class__.__name__.removesuffix('Model') - - @staticmethod - def values_to_dict(subscription_name, sources, source_type, event_categories, sns_topic_arn) -> Dict: - """ - Convert database values to a dictionary. - - Args: - subscription_name (str): The name of the subscription. - sources (str): JSON string of sources. - source_type (str): The type of the source. - event_categories (str): JSON string of event categories. - sns_topic_arn (str): The ARN of the SNS topic. - - Returns: - Dict: A dictionary representation of the EventSubscription. - """ - return { - 'subscription_name': subscription_name, - 'sources': json.loads(sources), - 'source_type': source_type, - 'event_categories': json.loads(event_categories), - 'sns_topic_arn': sns_topic_arn - } 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..3fc2810e --- /dev/null +++ b/DB/NEW_KT_DB/Service/Classes/DBClusterParameterGroupService.py @@ -0,0 +1,197 @@ +from abc import abstractmethod +import json +import os +import sys +from typing import Optional, Dict +from NEW_KT_DB.Service.Abc.DBO import DBO +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) + 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/Service/Classes/DBClusterService.py b/DB/NEW_KT_DB/Service/Classes/DBClusterService.py deleted file mode 100644 index cf2fe18d..00000000 --- a/DB/NEW_KT_DB/Service/Classes/DBClusterService.py +++ /dev/null @@ -1,187 +0,0 @@ -import json -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 typing import Dict, Optional -from DataAccess import DBClusterManager -from Models import DBClusterModel -from Abc import DBO -from Validation import DBClusterValiditions -from DataAccess import DBClusterManager -from Validation.DBClusterValiditions import ( - validate_db_cluster_identifier, - validate_engine, - validate_database_name, - validate_db_cluster_parameter_group_name, - validate_db_subnet_group_name, - validate_port, - check_required_params, - validate_master_user_password, - validate_master_username -) -import Exceptions.DBClusterExceptions as DBClusterExceptions -from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager - -class DBClusterService: - def __init__(self, dal: DBClusterManager, storage_manager: StorageManager, directory:str): - self.dal = dal - self.directory = directory - self.storage_manager = storage_manager - if not self.storage_manager.is_directory_exist(directory): - self.storage_manager.create_directory(directory) - - def get_file_path(self, cluster_name: str): - return str(self.directory)+'\\'+str(cluster_name)+'.json' - - def is_cluster_exist(self, cluster_identifier: str): - cluster_path = self.get_file_path(cluster_identifier) - cluster_configurations_path = self.get_file_path(cluster_identifier+"_configurations") - if not self.storage_manager.is_file_exist(cluster_configurations_path) or not self.storage_manager.is_directory_exist(cluster_path): - return False - - if not self.dal.is_exists(cluster_identifier): - return False - - return True - - def _validate_parameters(self, **kwargs): - # Perform validations - if 'db_cluster_identifier' in kwargs and self.dal.is_db_instance_exist(kwargs.get('db_cluster_identifier')): - raise DBClusterExceptions.DBClusterAlreadyExists(kwargs.get('db_cluster_identifier')) - - if 'db_cluster_identifier' in kwargs : - validate_db_cluster_identifier(kwargs.get('db_cluster_identifier')) - validate_engine(kwargs.get('engine', '')) - validate_db_subnet_group_name(kwargs.get('db_subnet_group_name')) - - if 'database_name' in kwargs: - validate_database_name(kwargs['database_name']) - if 'db_cluster_parameter_group_name' in kwargs : - validate_db_cluster_parameter_group_name(kwargs['db_cluster_parameter_group_name']) - if 'port' in kwargs: - validate_port(kwargs['port']) - if 'master_username' in kwargs : - validate_master_username(kwargs['master_username']) - if 'master_user_password' in kwargs : - validate_master_user_password(kwargs['master_user_password'], kwargs.get('manage_master_user_password', False)) - - - def create(self, instance_controller, **kwargs): - - '''Create a new DBCluster.''' - - # Validate required parameters - required_params = ['db_cluster_identifier', 'engine', 'db_subnet_group_name', 'allocated_storage'] - check_required_params(required_params, **kwargs) - self._validate_parameters(**kwargs) - - - # Create the cluster object - cluster = DBClusterModel.Cluster(**kwargs) - - # Create physical folder structure - cluster_directory = str(self.directory)+'\\'+str(cluster.db_cluster_identifier) - self.storage_manager.create_directory(cluster_directory) - - # Set cluster endpoint - cluster.cluster_endpoint = cluster_directory - - primary_instance_name = f'{cluster.db_cluster_identifier}-primary' - primary_instance = instance_controller.create_db_instance( - db_instance_identifier=primary_instance_name, - # cluster_identifier=cluster.db_cluster_identifier, - allocated_storage=cluster.allocated_storage, - master_username=cluster.master_username, - master_user_password=cluster.master_user_password - ) - - # Retrieve primary instance details - primary_instance_json_string = primary_instance.get("DBInstance") - cluster.instances_endpoints["primary_instance"] = primary_instance_json_string.get("endpoint") - cluster.primary_writer_instance = primary_instance_json_string.get('db_instance_identifier') - - # # Create configuration file - configuration_file_path = cluster_directory+'\\'+cluster.db_cluster_identifier + "_configurations.json" - json_object = json.dumps(cluster.to_dict()) - self.storage_manager.create_file( - file_path=configuration_file_path, content=json_object) - - cluster_to_sql = cluster.to_sql() - return self.dal.createInMemoryDBCluster(cluster_to_sql) - - - def delete(self,instance_controller, cluster_identifier:str): - '''Delete an existing DBCluster.''' - - if not self.dal.is_db_instance_exist(cluster_identifier): - raise DBClusterExceptions.DBClusterNotFoundException(cluster_identifier) - - file_path = self.get_file_path(cluster_identifier+"_configurations") - self.storage_manager.delete_file(file_path=file_path) - - directory_path = str(self.directory)+'\\'+str(cluster_identifier) - self.storage_manager.delete_directory(directory_path) - - columns = ['db_cluster_identifier', 'engine', 'allocated_storage', 'copy_tags_to_snapshot', - 'db_cluster_instance_class', 'database_name', 'db_cluster_parameter_group_name', - 'db_subnet_group_name', 'deletion_protection', 'engine_version', 'master_username', - 'master_user_password', 'manage_master_user_password', 'option_group_name', 'port', - 'replication_source_identifier', 'storage_encrypted', 'storage_type', 'tags', - 'created_at', 'status', 'primary_writer_instance', 'reader_instances', 'cluster_endpoint', - 'instances_endpoints', 'pk_column', 'pk_value'] - - #update configurations - current_cluster = self.describe(cluster_identifier) - cluster_dict = dict(zip(columns, current_cluster[0])) - instance_controller.delete_db_instance(db_instance_identifier = cluster_dict['primary_writer_instance'],skip_final_snapshot = True) - if cluster_dict['reader_instances'] != '[]' : - for id in cluster_dict['reader_instances']: - instance_controller.delete_db_instance(db_instance_identifier = id, skip_final_snapshot = True) - - - self.dal.deleteInMemoryDBCluster(cluster_identifier) - - - def describe(self, cluster_id): - '''Describe the details of DBCluster.''' - if not self.dal.is_db_instance_exist(cluster_id): - raise DBClusterExceptions.DBClusterNotFoundException(cluster_id) - - return self.dal.describeDBCluster(cluster_id) - - - def modify(self, cluster_id: str, **kwargs): - '''Modify an existing DBCluster.''' - - if not self.dal.is_db_instance_exist(cluster_id): - raise DBClusterExceptions.DBClusterNotFoundException(cluster_id) - - self._validate_parameters(**kwargs) - - str_parts = ', '.join(f"{key} = '{value}'" for key, value in kwargs.items()) - - #update in memory - self.dal.modifyDBCluster(cluster_id,str_parts) - - columns = ['db_cluster_identifier', 'engine', 'allocated_storage', 'copy_tags_to_snapshot', - 'db_cluster_instance_class', 'database_name', 'db_cluster_parameter_group_name', - 'db_subnet_group_name', 'deletion_protection', 'engine_version', 'master_username', - 'master_user_password', 'manage_master_user_password', 'option_group_name', 'port', - 'replication_source_identifier', 'storage_encrypted', 'storage_type', 'tags', - 'created_at', 'status', 'primary_writer_instance', 'reader_instances', 'cluster_endpoint', - 'instances_endpoints', 'pk_column', 'pk_value'] - - #update configurations - current_cluster = self.describe(cluster_id) - cluster_dict = dict(zip(columns, current_cluster[0])) - cluster_string = json.dumps(cluster_dict, indent=4) - - file_path = self.get_file_path(cluster_id+'_configurations') - self.storage_manager.delete_file(file_path) - self.storage_manager.create_file(file_path, cluster_string) - - - def get_all_cluster(self): - return self.dal.get_all_clusters() \ No newline at end of file diff --git a/DB/NEW_KT_DB/Service/Classes/DBInstanceNaiveService.py b/DB/NEW_KT_DB/Service/Classes/DBInstanceNaiveService.py deleted file mode 100644 index 9829241b..00000000 --- a/DB/NEW_KT_DB/Service/Classes/DBInstanceNaiveService.py +++ /dev/null @@ -1,165 +0,0 @@ -import os -import shutil -import sys -from typing import Dict, Optional -from Exceptions.DBInstanceNaiveException import DBInstanceNotFoundError, ParamValidationError,AlreadyExistsError -from Validation.DBInstanceNaiveValidition import check_extra_params, check_required_params, is_valid_db_instance_identifier -from Models.DBInstanceNaiveModel import DBInstance -from Service.Abc.DBO import DBO -from DataAccess.DBInstanceNaiveManager import DBInstanceManager -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../../'))) -from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager - - -class DBInstanceService(DBO): - - def __init__(self, dal: DBInstanceManager): - self.dal = dal - - def create(self, **attributes): - """ - Create a new DBInstance. - - Params: attributes: dict containing attributes for the new DBInstance. - Required fields: 'db_instance_identifier', 'master_username', 'master_user_password' - Optional fields: 'db_name', 'port', 'allocated_storage' - - Raises: ValueError: if the db_instance_identifier is invalid. - AlreadyExistsError: if a DB instance with the given identifier already exists. - - Return: dict with a 'DBInstance' key containing the created instance's details. - """ - required_params = ['db_instance_identifier', 'master_username', 'master_user_password'] - all_params = ['db_name', 'port', 'allocated_storage'] - all_params.extend(required_params) - - # Validate required and extra parameters - check_required_params(required_params, attributes) - check_extra_params(all_params, attributes) - - db_instance_identifier = attributes['db_instance_identifier'] - - # Validate DB instance identifier format - if not is_valid_db_instance_identifier(db_instance_identifier, 63): - raise ValueError('db_instance_identifier is invalid') - - if self.dal.is_db_instance_exist(db_instance_identifier): - raise AlreadyExistsError(f"The ID {db_instance_identifier} already exists") - - db_instance = DBInstance(**attributes) - self.dal.createInMemoryDBInstance(db_instance) - - return {'DBInstance': db_instance.to_dict()} - - def delete(self, db_instance_identifier,skip_final_snapshot=False,final_db_snapshot_identifier=None,delete_automated_backups=False): - """ - Delete a DBInstance by its identifier with options to handle final snapshots and automated backups. - - Params: db_instance_identifier: The primary key (ID) of the DBInstance to delete. - skip_final_snapshot: If True, skip creating a final snapshot before deletion. Defaults to False. - final_db_snapshot_identifier: If skip_final_snapshot is False, specify an identifier for the final DB snapshot. - delete_automated_backups: If True, delete automated backups along with the DBInstance. Defaults to False. - """ - if not self.dal.is_db_instance_exist(db_instance_identifier): - raise DBInstanceNotFoundError('This DB instance identifier does not exist') - - # Get the DBInstance to delete - object - db_instance = self.get(db_instance_identifier) - - # Handle final snapshot before deletion if required - if skip_final_snapshot == False: - if final_db_snapshot_identifier is None: - raise ParamValidationError('If skip_final_snapshot is False, final_db_snapshot_identifier must be specified') - create_db_snapshot( - db_instance_identifier=db_instance_identifier, - db_snapshot_identifier=final_db_snapshot_identifier - ) - - self.dal.deleteInMemoryDBInstance(db_instance_identifier) - - # Clean up storage (delete associated directory) - endpoint = db_instance.endpoint - storageManager = StorageManager(DBInstance.BASE_PATH) - storageManager.delete_directory(endpoint) - - # Delete the DBInstance object - del db_instance - - def describe(self, db_instance_identifier): - """ - Describe the details of a DBInstance. - - Params: db_instance_identifier: str, identifier of the DBInstance to describe. - - Raises: DBInstanceNotFoundError: if the DB instance does not exist. - - Return: dict with a 'DBInstance' key containing the details of the instance. - """ - if not self.dal.is_db_instance_exist(db_instance_identifier): - raise DBInstanceNotFoundError('This DB instance identifier does not exist') - - # Retrieve and structure the DBInstance description - describe_db_instance = self.dal.describeDBInstance(db_instance_identifier)[0] - describe_db_instance_dict = { - 'db_instance_identifier': describe_db_instance[0], - 'allocated_storage': describe_db_instance[1], - 'master_username': describe_db_instance[2], - 'master_user_password': describe_db_instance[3], - 'db_name': describe_db_instance[4], - 'port': describe_db_instance[5], - 'status': describe_db_instance[6], - 'created_time': str(describe_db_instance[7]), - 'endpoint': describe_db_instance[8], - 'databases': describe_db_instance[9], - 'pk_value': describe_db_instance[10] - } - - return {'DBInstance': describe_db_instance_dict} - - def modify(self, **updates): - """ - Modify an existing DBInstance. - - Params: updates: dict containing the attributes to modify in the DBInstance. - Required field: 'db_instance_identifier' - Optional fields: 'port', 'allocated_storage', 'master_user_password' - - Raises: DBInstanceNotFoundError: if the DB instance does not exist. - - Return: dict with a 'DBInstance' key containing the updated instance's details. - """ - required_params = ['db_instance_identifier'] - all_params = ['port', 'allocated_storage', 'master_user_password'] - all_params.extend(required_params) - - # Validate required and extra parameters - check_required_params(required_params, updates) - check_extra_params(all_params, updates) - - db_instance_identifier = updates['db_instance_identifier'] - - # Prepare the SQL-like set clause for updates - filtered_updates = {key: value for key, value in updates.items() if key != 'db_instance_identifier'} - set_clause = ', '.join([f"{key} = '{value}'" for key, value in filtered_updates.items()]) - - self.dal.modifyDBInstance(db_instance_identifier, set_clause) - - # Return the updated DB instance details - update_db_instance = self.get(db_instance_identifier) - return {'DBInstance': update_db_instance} - - def get(self, db_instance_identifier): - """ - Retrieve a DBInstance object from the database. - - Params: db_instance_identifier: str, identifier of the DBInstance to retrieve. - - Return: DBInstance object or None if not found. - """ - describe_result = self.describe(db_instance_identifier) - - if describe_result: - describe_result = describe_result['DBInstance'] - return DBInstance(**describe_result) - - return None diff --git a/DB/NEW_KT_DB/Service/Classes/EventSubscriptionService.py b/DB/NEW_KT_DB/Service/Classes/EventSubscriptionService.py deleted file mode 100644 index 06dc3cb3..00000000 --- a/DB/NEW_KT_DB/Service/Classes/EventSubscriptionService.py +++ /dev/null @@ -1,152 +0,0 @@ -import json -from typing import Any, Dict, List, Tuple -from DB.NEW_KT_DB.DataAccess.EventSubscriptionManager import EventSubscriptionManager -from DB.NEW_KT_DB.Models.EventSubscriptionModel import EventCategory, EventSubscription, SourceType -from DB.NEW_KT_DB.Service.Abc.DBO import DBO -from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager - - -class EventSubscriptionService(DBO): - """ - A service class for managing event subscriptions. - - This class provides methods to create, delete, modify, and retrieve event subscriptions. - It interacts with both an in-memory data access layer and a file storage system. - - Attributes: - dal (EventSubscriptionManager): The data access layer for event subscriptions. - storage_manager (StorageManager): The storage manager for file operations. - directory (str): The directory path for storing event subscription files. - """ - - def __init__(self, dal: EventSubscriptionManager, storage_manager: StorageManager, directory: str): - """ - Initialize the EventSubscriptionService. - - Args: - dal (EventSubscriptionManager): The data access layer for event subscriptions. - storage_manager (StorageManager): The storage manager for file operations. - directory (str): The directory path for storing event subscription files. - """ - self.dal = dal - self.storage_manager = storage_manager - self.directory = directory - if not self.storage_manager.is_directory_exist(directory): - self.storage_manager.create_directory(directory) - - def create(self, subscription_name: str, sources: List[Tuple[SourceType, str]], - event_categories: List[EventCategory], sns_topic_arn: str, source_type: SourceType): - """ - Create a new event subscription. - - Args: - subscription_name (str): The name of the subscription. - sources (List[Tuple[SourceType, str]]): The list of sources for the subscription. - event_categories (List[EventCategory]): The list of event categories. - sns_topic_arn (str): The ARN of the SNS topic. - source_type (SourceType): The type of the source. - """ - event_subscription = EventSubscription( - subscription_name, sources, event_categories, sns_topic_arn, source_type) - - self.dal.createInMemoryEventSubscription(event_subscription) - - self.storage_manager.create_file(self.get_file_path( - subscription_name), json.dumps(event_subscription.to_dict())) - - def delete(self, subscription_name: str): - """ - Delete an event subscription. - - Args: - subscription_name (str): The name of the subscription to delete. - """ - self.dal.deleteInMemoryEventSubscription(subscription_name) - self.storage_manager.delete_file(self.get_file_path(subscription_name)) - - def modify(self, subscription_name: str, event_categories: List[EventCategory] = None, sns_topic_arn: str = None, source_type: SourceType = None): - """ - Modify an existing event subscription. - - Args: - subscription_name (str): The name of the subscription to modify. - event_categories (List[EventCategory], optional): The new list of event categories. - sns_topic_arn (str, optional): The new ARN of the SNS topic. - source_type (SourceType, optional): The new type of the source. - """ - event_subscription = self.dal.get_by_id( - subscription_name) - - if event_categories is not None: - event_subscription.event_categories = event_categories - if sns_topic_arn is not None: - event_subscription.sns_topic_arn = sns_topic_arn - if source_type is not None: - event_subscription.source_type = source_type - - self.dal.modifyEventSubscription( - event_subscription) - - self.storage_manager.write_to_file(self.get_file_path( - subscription_name), json.dumps(event_subscription.to_dict())) - - def get_by_id(self, subscription_name: str): - """ - Get an event subscription by its name. - - Args: - subscription_name (str): The name of the subscription to retrieve. - - Returns: - EventSubscription: The event subscription with the given name. - """ - return self.dal.get_by_id(subscription_name) - - def describe(self, columns: List[str] = None, criteria: Dict[str, Any] = None) -> List[Dict]: - """ - Describe event subscriptions based on given criteria. - - Args: - columns (List[str], optional): The columns to include in the description. - criteria (Dict[str, Any], optional): The criteria to filter the subscriptions. - - Returns: - List[Dict]: A list of dictionaries describing the matching event subscriptions. - """ - return self.dal.describeEventSubscriptionByCriteria(columns, criteria) - - def describe_by_id(self, subscription_name: str) -> Dict: - """ - Describe an event subscription by its name. - - Args: - subscription_name (str): The name of the subscription to describe. - - Returns: - Dict: A dictionary describing the event subscription. - """ - return self.dal.describeEventSubscriptionById(subscription_name) - - def get(self, criteria: Dict[str, Any] = None): - """ - Get event subscriptions based on given criteria. - - Args: - criteria (Dict[str, Any], optional): The criteria to filter the subscriptions. - - Returns: - List[EventSubscription]: A list of event subscriptions matching the criteria. - """ - return self.dal.get(criteria) - - def get_file_path(self, subscription_name: str): - """ - Get the file path for a given subscription name. - - Args: - subscription_name (str): The name of the subscription. - - Returns: - str: The file path for the subscription. - """ - return f'{self.directory}/{subscription_name}.json' diff --git a/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py new file mode 100644 index 00000000..746e2bf8 --- /dev/null +++ b/DB/NEW_KT_DB/Test/DBClusterParameterGroupTests.py @@ -0,0 +1,309 @@ +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 * + +# 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' + +group_name = "TestGroup" +group_family = "TestFamily" +description = "Test Description" +file_name = generate_file_name_for_group(group_name) + +@pytest.fixture +def parameter_group_manager(): + return DBClusterParameterGroupManager('t') +# :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 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_name1=group_name+'1' + # 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_name0=group_name+'0' + # Ensure the group exists + create_parameter_group(parameter_group_controller, group_name0, 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_name1=group_name+'1' + file_name = generate_file_name_for_group(group_name1) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name1, "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_name1) + + # 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_name2=group_name+'2' + file_name = generate_file_name_for_group(group_name2) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name2, "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_name2)] #{"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_name2) + + # 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_name3=group_name+'3' + file_name = generate_file_name_for_group(group_name3) + + # Create a parameter group + create_parameter_group(parameter_group_controller, group_name3, group_family, 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_name3, 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_name4=group_name+'4' + file_name = generate_file_name_for_group(group_name4) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name4, group_family, 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_name4, 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_name4, new_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_with_invalid_is_modifiable(parameter_group_controller, storage_manager): + group_name5=group_name+'5' + file_name = generate_file_name_for_group(group_name5) + + # Create the parameter group + create_parameter_group(parameter_group_controller, group_name5, group_family, 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_name5, invalid_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_modify_with_invalid_apply_method(parameter_group_controller, storage_manager): + group_name6=group_name+'6' + file_name = generate_file_name_for_group(group_name6) + + + create_parameter_group(parameter_group_controller, group_name6, group_family, 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_name6, invalid_parameters) + + # Cleanup + delete_file_if_exists(storage_manager, file_name) + +def test_describe_parameter_group(parameter_group_controller, storage_manager): + group_name7=group_name+'7' + file_name = generate_file_name_for_group(group_name7) + + # Create a parameter group + create_parameter_group(parameter_group_controller, group_name7, group_family, description) + + # Describe the parameter group + result = parameter_group_controller.describe_db_cluste_parameter_group(group_name7) + # Check the result contains the correct description + assert_parameter_group_details(result, 0, group_name7, group_family, description) + result = parameter_group_controller.describe_db_cluste_parameter_group() + # Check the result contains the correct description + assert_parameter_group_details(result, 8, group_name7, group_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) diff --git a/DB/NEW_KT_DB/Test/DBClusterTests.py b/DB/NEW_KT_DB/Test/DBClusterTests.py deleted file mode 100644 index 93be77ce..00000000 --- a/DB/NEW_KT_DB/Test/DBClusterTests.py +++ /dev/null @@ -1,159 +0,0 @@ -import os -import sys -import pytest -import sqlite3 -sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..'))) - -from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager -from DB.NEW_KT_DB.Service.Classes.DBClusterService import DBClusterService -from DB.NEW_KT_DB.DataAccess.DBClusterManager import DBClusterManager -from Controller.DBClusterController import DBClusterController -from DataAccess.ObjectManager import ObjectManager -from Service.Classes.DBInstanceService import DBInstanceManager,DBInstanceService,AlreadyExistsError,ParamValidationError,DBInstanceNotFoundError -from Exceptions import DBClusterExceptions -from Controller.DBInstanceController import DBInstanceController - - -CLUSTER_DATA = { - 'db_cluster_identifier': 'ClusterTest', - 'engine': 'mysql', - 'allocated_storage':5, - 'db_subnet_group_name': 'my-subnet-group' - } - -@pytest.fixture -def object_manager(): - return ObjectManager('Clusters/instances.db') - -@pytest.fixture -def db_instance_manager(object_manager): - return DBInstanceManager(object_manager) - -@pytest.fixture -def db_instance_service(db_instance_manager): - return DBInstanceService(db_instance_manager) - -@pytest.fixture -def db_instance_controller(db_instance_service): - return DBInstanceController(db_instance_service) - -@pytest.fixture -def db_cluster_manager(): - return DBClusterManager('Clusters/clusters.db') - -@pytest.fixture -def storage_manager(): - return StorageManager('Clusters') - -@pytest.fixture -def db_cluster_service(db_cluster_manager,storage_manager): - return DBClusterService(db_cluster_manager,storage_manager, 'Clusters') - -@pytest.fixture -def db_cluster_controller(db_cluster_service, db_instance_controller): - return DBClusterController(db_cluster_service, db_instance_controller) - -@pytest.fixture -def db_cluster_controller_with_cleanup(db_cluster_controller): - - # Return the controller, but ensure cleanup happens after the test - yield db_cluster_controller - - # Finalizer to clean up after the test - db_cluster_controller.delete_db_cluster('ClusterTest') - -def test_create_cluster_works(db_cluster_controller_with_cleanup): - - res = db_cluster_controller_with_cleanup.create_db_cluster(**CLUSTER_DATA) - assert res == None - -def test_create_cluster_missing_required_fields(db_cluster_controller): - cluster_data = { - 'db_cluster_identifier': 'ClusterTest', - 'allocated_storage':5, - 'db_subnet_group_name': 'my-subnet-group' - } - with pytest.raises(DBClusterExceptions.MissingRequiredArgument): - db_cluster_controller.create_db_cluster(**cluster_data) - -def test_create_cluster_invalide_identifier(db_cluster_controller): - cluster_data = { - 'db_cluster_identifier': '1Cluste--Test', - 'engine': 'mysql', - 'allocated_storage':5, - 'db_subnet_group_name': 'my-subnet-group' - } - with pytest.raises(DBClusterExceptions.InvalidDBClusterArgument): - db_cluster_controller.create_db_cluster(**cluster_data) - -def test_create_cluster_already_exist(db_cluster_controller): - - db_cluster_controller.create_db_cluster(**CLUSTER_DATA) - with pytest.raises(DBClusterExceptions.DBClusterAlreadyExists): - db_cluster_controller.create_db_cluster(**CLUSTER_DATA) - - db_cluster_controller.delete_db_cluster('ClusterTest') - -def test_delete_cluster_works(db_cluster_controller): - - db_cluster_controller.create_db_cluster(**CLUSTER_DATA) - db_cluster_controller.delete_db_cluster('ClusterTest') - with pytest.raises(DBClusterExceptions.DBClusterNotFoundException): - db_cluster_controller.delete_db_cluster('ClusterTest') - -def test_delete_cluster_does_not_exist(db_cluster_controller): - with pytest.raises(DBClusterExceptions.DBClusterNotFoundException): - db_cluster_controller.delete_db_cluster('ClusterTest') - -def test_describe_cluster_works(db_cluster_controller_with_cleanup): - - db_cluster_controller_with_cleanup.create_db_cluster(**CLUSTER_DATA) - res = db_cluster_controller_with_cleanup.describe_db_cluster('ClusterTest') - assert isinstance(res, list) - -def test_describe_cluster_does_not_exist(db_cluster_controller): - with pytest.raises(DBClusterExceptions.DBClusterNotFoundException): - db_cluster_controller.describe_db_cluster('ClusterTest') - - -def test_modify_cluster_works(db_cluster_controller_with_cleanup): - - db_cluster_controller_with_cleanup.create_db_cluster(**CLUSTER_DATA) - update_data = { - 'engine': 'postgres', - } - db_cluster_controller_with_cleanup.modify_db_cluster('ClusterTest', **update_data) - res = db_cluster_controller_with_cleanup.describe_db_cluster('ClusterTest') - columns = ['db_cluster_identifier', 'engine', 'allocated_storage', 'copy_tags_to_snapshot', - 'db_cluster_instance_class', 'database_name', 'db_cluster_parameter_group_name', - 'db_subnet_group_name', 'deletion_protection', 'engine_version', 'master_username', - 'master_user_password', 'manage_master_user_password', 'option_group_name', 'port', - 'replication_source_identifier', 'storage_encrypted', 'storage_type', 'tags', - 'created_at', 'status', 'primary_writer_instance', 'reader_instances', 'cluster_endpoint', - 'instances_endpoints', 'pk_column', 'pk_value'] - - cluster_dict = dict(zip(columns, res[0])) - assert cluster_dict['engine'] == 'postgres' - -def test_modify_cluster_does_not_exist(db_cluster_controller): - update_data = { - 'engine': 'postgres', - } - with pytest.raises(DBClusterExceptions.DBClusterNotFoundException): - db_cluster_controller.modify_db_cluster('ClusterTest',**update_data) - - -def test_modify_cluster_invalide_engine(db_cluster_controller_with_cleanup): - - db_cluster_controller_with_cleanup.create_db_cluster(**CLUSTER_DATA) - update_data = { - 'engine': 'invalid', - } - with pytest.raises(DBClusterExceptions.InvalidDBClusterArgument): - db_cluster_controller_with_cleanup.modify_db_cluster('ClusterTest',**update_data) - -def test_get_all_clusters(db_cluster_controller_with_cleanup): - - db_cluster_controller_with_cleanup.create_db_cluster(**CLUSTER_DATA) - res = db_cluster_controller_with_cleanup.get_all_db_clusters() - assert isinstance(res, list) \ No newline at end of file diff --git a/DB/NEW_KT_DB/Test/DBSubnetGroupTests.py b/DB/NEW_KT_DB/Test/DBSubnetGroupTests.py deleted file mode 100644 index 6697641d..00000000 --- a/DB/NEW_KT_DB/Test/DBSubnetGroupTests.py +++ /dev/null @@ -1,477 +0,0 @@ -import pytest -import random -import string - -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 Service.Classes.DBSubnetGroupService import DBSubnetGroupService -from DataAccess.DBSubnetGroupManager import DBSubnetGroupManager -from Controller.DBSubnetGroupController import DBSubnetGroupController -from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager -from DataAccess.ObjectManager import ObjectManager -from Models.DBSubnetGroupModel import DBSubnetGroup -import Exceptions.DBSubnetGroupExceptions as DBSubnetGroupExceptions -import sqlite3 - -object_manager = ObjectManager("../object_management_db.db") - -manager = DBSubnetGroupManager(object_manager=object_manager) -storage_manager = StorageManager("DB/s3") - -service = DBSubnetGroupService(manager, storage_manager=storage_manager) -controller = DBSubnetGroupController(service) - -@pytest.fixture -def clear_table(): - # Connect to the SQLite database - conn = conn = sqlite3.connect("../object_management_db.db") - cursor = conn.cursor() - - # Clear the table if it exists - table_name = "mng_DBSubnetGroups" - cursor.execute(f"DELETE FROM {table_name};") - - # Commit changes and close the connection - conn.commit() - conn.close() - - # Yield to allow tests to run - yield - - -def test_create(clear_table): - - # remove existing subnet group from previous tests - controller.create_db_subnet_group( - db_subnet_group_name="subnet_group_1", - subnets=[{"subnet_id": "subnet-12345678"}, {"subnet_id": "subnet-87654321"}], - db_subnet_group_description="Test subnet group", - vpc_id="vpc-12345678", - db_subnet_group_arn="arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_1", - ) - - # check that file was created (no error raised on get) - storage_manager.is_file_exist("db_subnet_groups/subnet_group_1") - # check that object was saved to management table (no error raised on get) - controller.get_db_subnet_group("subnet_group_1") - - # check that file content is correct - # storage manager doesn't have a get or read function - # from_storage = DBSubnetGroup( - # **DBSubnetGroup.from_bytes_to_dict( - # storage_manager.get("db_subnet_groups", "subnet_group_1", "0")["content"] - # ) - # ) - - file = open("C:/Users/temim/בוטקפמ vast data/KT_Cloud/DB/s3/db_subnet_groups/subnet_group_1", "r") - str_data = file.read() - file.close() - from_storage = DBSubnetGroup.from_str(str_data) - from_db = controller.get_db_subnet_group("subnet_group_1") - # check values were stored correctly in management table as well as storage - - # group name - assert from_storage.db_subnet_group_name == "subnet_group_1" - assert from_db.db_subnet_group_name == "subnet_group_1" - - # subnets - for subnet in from_storage.subnets: - assert subnet in [ - {"subnet_id": "subnet-12345678"}, - {"subnet_id": "subnet-87654321"}, - ] - for subnet in from_db.subnets: - assert subnet in [ - {"subnet_id": "subnet-12345678"}, - {"subnet_id": "subnet-87654321"}, - ] - - # description - assert from_storage.db_subnet_group_description == "Test subnet group" - assert from_db.db_subnet_group_description == "Test subnet group" - - # vpc_id - assert from_storage.vpc_id == "vpc-12345678" - assert from_db.vpc_id == "vpc-12345678" - - # arn - assert ( - from_storage.db_subnet_group_arn - == "arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_1" - ) - assert ( - from_db.db_subnet_group_arn - == "arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_1" - ) - - -def test_unique_constraint(): - with pytest.raises(DBSubnetGroupExceptions.DBSubnetGroupAlreadyExists): - controller.create_db_subnet_group( - db_subnet_group_name="subnet_group_1", - subnets=[ - {"subnet_id": "subnet-87654321"}, - {"subnet_id": "subnet-12345678"}, - ], - db_subnet_group_description="Another subnet group with same name", - vpc_id="vpc-87654321", - db_subnet_group_arn="arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_1", - ) - - -def test_get(): - subnet_group_1 = controller.get_db_subnet_group("subnet_group_1") - assert subnet_group_1 != None - assert subnet_group_1.db_subnet_group_name == "subnet_group_1" - assert subnet_group_1.db_subnet_group_description == "Test subnet group" - assert subnet_group_1.vpc_id == "vpc-12345678" - assert ( - subnet_group_1.db_subnet_group_arn - == "arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_1" - ) - assert subnet_group_1.status == "pending" - for subnet in subnet_group_1.subnets: - assert subnet in [ - {"subnet_id": "subnet-12345678"}, - {"subnet_id": "subnet-87654321"}, - ] - - -def test_modify(): - controller.modify_db_subnet_group( - "subnet_group_1", - subnets=[{"subnet_id": "subnet-12345988"}, {"subnet_id": "subnet-876543881"}], - ) - - # from_storage = DBSubnetGroup( - # **DBSubnetGroup.from_bytes_to_dict( - # storage_manager.get("db_subnet_groups", "subnet_group_1", "0")["content"] - # ) - # ) - file = open("C:/Users/temim/בוטקפמ vast data/KT_Cloud/DB/s3/db_subnet_groups/subnet_group_1", "r") - from_storage = DBSubnetGroup.from_str(file.read()) - file.close() - - from_db = controller.get_db_subnet_group("subnet_group_1") - assert from_storage.db_subnet_group_name == from_db.db_subnet_group_name - for subnet in from_storage.subnets: - assert subnet in from_db.subnets - assert ( - from_storage.db_subnet_group_description == from_db.db_subnet_group_description - ) - assert from_storage.vpc_id == from_db.vpc_id - assert from_storage.db_subnet_group_arn == from_db.db_subnet_group_arn - assert from_storage.status == from_db.status - - -def test_describe(): - subnet_group_1 = controller.describe_db_subnet_group("subnet_group_1") - assert type(subnet_group_1) == dict - assert type(subnet_group_1["subnets"]) == list - assert type(subnet_group_1["subnets"][0]) == dict - subnet_group_1 = DBSubnetGroup(**subnet_group_1) - assert subnet_group_1 != None - assert subnet_group_1.db_subnet_group_name == "subnet_group_1" - assert subnet_group_1.db_subnet_group_description == "Test subnet group" - assert subnet_group_1.vpc_id == "vpc-12345678" - assert ( - subnet_group_1.db_subnet_group_arn - == "arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_1" - ) - assert subnet_group_1.status == "pending" - for subnet in subnet_group_1.subnets: - assert subnet in [ - {"subnet_id": "subnet-12345988"}, - {"subnet_id": "subnet-876543881"}, - ] - - -def test_delete(): - controller.delete_db_subnet_group("subnet_group_1") - with pytest.raises(FileNotFoundError): - file = open("C:/Users/temim/בוטקפמ vast data/KT_Cloud/DB/s3/db_subnet_groups/subnet_group_1", "r") - from_storage = DBSubnetGroup.from_str(file.read()) - file.close() - with pytest.raises(DBSubnetGroupExceptions.DBSubnetGroupNotFound): - controller.get_db_subnet_group("subnet_group_1") - - -@pytest.mark.parametrize("index", range(20)) -def test_insert_many(index): - controller.create_db_subnet_group( - db_subnet_group_name=f"subnet_group_{index}", - subnets=[ - {"subnet_id": f"subnet-1234567{index}"}, - {"subnet_id": f"subnet-8765432{index}"}, - ], - db_subnet_group_description=f"Test subnet group {index}", - vpc_id=f"vpc-1234567{index}", - db_subnet_group_arn=f"arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_{index}", - ) - - # check that file was created (no error raised on get) - file = open(f"C:/Users/temim/בוטקפמ vast data/KT_Cloud/DB/s3/db_subnet_groups/subnet_group_{index}", "r") - from_storage = DBSubnetGroup.from_str(file.read()) - file.close() - # check that object was saved to management table (no error raised on get) - controller.get_db_subnet_group(f"subnet_group_{index}") - - # check that file content is correct - # from_storage = DBSubnetGroup( - # **DBSubnetGroup.from_bytes_to_dict( - # storage_manager.get("db_subnet_groups", f"subnet_group_{index}", "0")[ - # "content" - # ] - # ) - # ) - file = open(f"C:/Users/temim/בוטקפמ vast data/KT_Cloud/DB/s3/db_subnet_groups/subnet_group_{index}", "r") - from_storage = DBSubnetGroup.from_str(file.read()) - file.close() - from_db = controller.get_db_subnet_group(f"subnet_group_{index}") - # check values were stored correctly in management table as well as storage - - # group name - assert from_storage.db_subnet_group_name == f"subnet_group_{index}" - assert from_db.db_subnet_group_name == f"subnet_group_{index}" - - # subnets - for subnet in from_storage.subnets: - assert subnet in [ - {"subnet_id": f"subnet-1234567{index}"}, - {"subnet_id": f"subnet-8765432{index}"}, - ] - for subnet in from_db.subnets: - assert subnet in [ - {"subnet_id": f"subnet-1234567{index}"}, - {"subnet_id": f"subnet-8765432{index}"}, - ] - - # description - assert from_storage.db_subnet_group_description == f"Test subnet group {index}" - assert from_db.db_subnet_group_description == f"Test subnet group {index}" - - # vpc_id - assert from_storage.vpc_id == f"vpc-1234567{index}" - assert from_db.vpc_id == f"vpc-1234567{index}" - - # arn - assert ( - from_storage.db_subnet_group_arn - == f"arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_{index}" - ) - assert ( - from_db.db_subnet_group_arn - == f"arn:aws:rds:us-west-2:123456789012:subgrp:subnet_group_{index}" - ) - - -@pytest.mark.parametrize("index", range(20)) -def test_delete_many_from_prev_test(index): - db_subnet_group_name = f"subnet_group_{index}" - controller.delete_db_subnet_group(db_subnet_group_name) - with pytest.raises(FileNotFoundError): - file = open(f"C:/Users/temim/בוטקפמ vast data/KT_Cloud/DB/s3/db_subnet_groups/subnet_group_{index}", "r") - from_storage = DBSubnetGroup.from_str(file.read()) - file.close() - with pytest.raises(DBSubnetGroupExceptions.DBSubnetGroupNotFound): - controller.get_db_subnet_group(db_subnet_group_name) - - -# tests below this comment were generated by Cody -def generate_random_string(length): - return "".join(random.choices(string.ascii_lowercase + string.digits, k=length)) - - -@pytest.mark.parametrize("num_subnets", [1, 5, 20]) -def test_create_db_subnet_group_with_varying_subnets(num_subnets): - vpc_id = f"vpc-{generate_random_string(8)}" - subnets = [ - {"subnet_id": f"subnet-{generate_random_string(8)}"} for _ in range(num_subnets) - ] - db_subnet_group_name = f"test-group-{generate_random_string(8)}" - description = f"Test subnet group with {num_subnets} subnets" - - controller.create_db_subnet_group( - db_subnet_group_name=db_subnet_group_name, - db_subnet_group_description=description, - subnets=subnets, - vpc_id=vpc_id, - ) - - # Verify the created group - from_db = controller.get_db_subnet_group(db_subnet_group_name) - print(from_db.subnets) - assert from_db.db_subnet_group_name == db_subnet_group_name - assert from_db.db_subnet_group_description == description - assert from_db.vpc_id == vpc_id - assert len(from_db.subnets) == num_subnets - for subnet in subnets: - assert subnet in from_db.subnets - - # Clean up - controller.delete_db_subnet_group(db_subnet_group_name) - - -@pytest.mark.parametrize("num_groups", [5, 10, 20]) -def test_create_multiple_db_subnet_groups(num_groups): - vpc_id = f"vpc-{generate_random_string(8)}" - groups = [] - - for _ in range(num_groups): - db_subnet_group_name = f"test-group-{generate_random_string(8)}" - description = f"Test subnet group {_}" - subnets = [ - {"subnet_id": f"subnet-{generate_random_string(8)}"} for _ in range(2) - ] - - controller.create_db_subnet_group( - db_subnet_group_name=db_subnet_group_name, - db_subnet_group_description=description, - subnets=subnets, - vpc_id=vpc_id, - ) - groups.append(db_subnet_group_name) - - # Verify all groups were created - for group_name in groups: - from_db = controller.get_db_subnet_group(group_name) - assert from_db.db_subnet_group_name == group_name - assert from_db.vpc_id == vpc_id - - # Clean up - for group_name in groups: - controller.delete_db_subnet_group(group_name) - - -def test_update_db_subnet_group(): - vpc_id = f"vpc-{generate_random_string(8)}" - initial_subnets = [ - {"subnet_id": f"subnet-{generate_random_string(8)}"} for _ in range(2) - ] - db_subnet_group_name = f"test-group-{generate_random_string(8)}" - initial_description = "Initial description" - - controller.create_db_subnet_group( - db_subnet_group_name=db_subnet_group_name, - db_subnet_group_description=initial_description, - subnets=initial_subnets, - vpc_id=vpc_id, - ) - - # Update the group - new_description = "Updated description" - new_subnet = {"subnet_id": f"subnet-{generate_random_string(8)}"} - updated_subnets = initial_subnets + [new_subnet] - - controller.modify_db_subnet_group( - db_subnet_group_name, - db_subnet_group_description=new_description, - subnets=updated_subnets, - ) - - # Verify the update - from_db = controller.get_db_subnet_group(db_subnet_group_name) - assert from_db.db_subnet_group_name == db_subnet_group_name - assert from_db.db_subnet_group_description == new_description - assert from_db.vpc_id == vpc_id - assert len(from_db.subnets) == len(updated_subnets) - for subnet in from_db.subnets: - assert subnet in updated_subnets - - # Clean up - controller.delete_db_subnet_group(db_subnet_group_name) - - -def test_delete_nonexistent_db_subnet_group(): - non_existent_group_name = f"non-existent-group-{generate_random_string(8)}" - - with pytest.raises(Exception): - controller.delete_db_subnet_group(non_existent_group_name) - - -@pytest.mark.parametrize("num_operations", [50, 100, 200]) -def test_concurrent_operations(num_operations): - vpc_id = f"vpc-{generate_random_string(8)}" - group_names = [ - f"test-group-{generate_random_string(8)}" for _ in range(num_operations) - ] - - # Create groups - for group_name in group_names: - subnets = [ - {"subnet_id": f"subnet-{generate_random_string(8)}"} for _ in range(2) - ] - controller.create_db_subnet_group( - db_subnet_group_name=group_name, - db_subnet_group_description=f"Test group {group_name}", - subnets=subnets, - vpc_id=vpc_id, - ) - - # Perform random operations - for _ in range(num_operations): - operation = random.choice(["get", "update", "delete"]) - group_name = random.choice(group_names) - - if operation == "get": - try: - controller.get_db_subnet_group(group_name) - except Exception: - pass - elif operation == "update": - try: - new_description = f"Updated description for {group_name}" - new_subnet = {"subnet_id": f"subnet-{generate_random_string(8)}"} - controller.modify_db_subnet_group( - db_subnet_group_name=group_name, - db_subnet_group_description=new_description, - subnet_ids=[new_subnet["subnet_id"]], - ) - except Exception: - pass - elif operation == "delete": - try: - controller.delete_db_subnet_group(group_name) - group_names.remove(group_name) - except Exception: - pass - - # Clean up any remaining groups - for group_name in group_names: - controller.delete_db_subnet_group(group_name) - - -def test_db_subnet_group_listing(): - vpc_id = f"vpc-{generate_random_string(8)}" - num_groups = 5 - group_names = [] - - # Create groups - for i in range(num_groups): - group_name = f"test-group-{generate_random_string(8)}" - group_names.append(group_name) - subnets = [ - {"subnet_id": f"subnet-{generate_random_string(8)}"} for _ in range(2) - ] - controller.create_db_subnet_group( - db_subnet_group_name=group_name, - db_subnet_group_description=f"Test group {i}", - subnet=subnets, - vpc_id=vpc_id, - ) - - # List all groups - all_groups = controller.list_db_subnet_groups() - - # Verify all created groups are in the list - for group_name in group_names: - assert any(group.db_subnet_group_name == group_name for group in all_groups) - - # Clean up - for group_name in group_names: - controller.delete_db_subnet_group(group_name) diff --git a/DB/NEW_KT_DB/Test/EventSubscriptionTests.py b/DB/NEW_KT_DB/Test/EventSubscriptionTests.py deleted file mode 100644 index 9e681435..00000000 --- a/DB/NEW_KT_DB/Test/EventSubscriptionTests.py +++ /dev/null @@ -1,119 +0,0 @@ -import pytest - -from DB.NEW_KT_DB.Service.Classes.EventSubscriptionService import EventSubscriptionService -from DB.NEW_KT_DB.DataAccess.EventSubscriptionManager import EventSubscriptionManager -from DB.NEW_KT_DB.Models.EventSubscriptionModel import EventCategory, EventSubscription, SourceType -from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager - - -@pytest.fixture -def dal(): - return EventSubscriptionManager('test_db_file.db') - - -@pytest.fixture -def storage_manager(): - dir = 'test_storage_directory' - manager = StorageManager(dir) - yield manager - manager.delete_directory(dir) - - -@pytest.fixture -def event_subscription_service(dal: EventSubscriptionManager, storage_manager: StorageManager): - dir = 'test_directory' - service = EventSubscriptionService(dal, storage_manager, dir) - yield service - storage_manager.delete_directory(dir) - - -@pytest.fixture -def event_subscription(event_subscription_service: EventSubscriptionService): - sources = [ - (SourceType.DB_INSTANCE, 'test_instance'), - (SourceType.DB_CLUSTER, 'db_cluster') - ] - event_subscription = EventSubscription( - 'test_subscription', - sources, - [EventCategory.CREATION, EventCategory.DELETION], - 'test_sns_topic_arn', - SourceType.DB_INSTANCE - ) - event_subscription_service.create( - event_subscription.subscription_name, - sources, # Pass sources directly, not event_subscription.sources - event_subscription.event_categories, - event_subscription.sns_topic_arn, - event_subscription.source_type - ) - yield event_subscription - event_subscription_service.delete(event_subscription.subscription_name) - - -def test_create(event_subscription_service: EventSubscriptionService, event_subscription: EventSubscription): - # Assuming that __eq__ is implemented for EventSubscription - assert event_subscription == event_subscription_service.get_by_id( - event_subscription.subscription_name) - - assert event_subscription_service.storage_manager.is_file_exist( - event_subscription_service.get_file_path(event_subscription.subscription_name)) - - -def test_delete(event_subscription_service: EventSubscriptionService, event_subscription: EventSubscription): - event_subscription_service.delete(event_subscription.subscription_name) - - assert not event_subscription_service.get_by_id( - event_subscription.subscription_name) - - assert not event_subscription_service.storage_manager.is_file_exist( - event_subscription_service.get_file_path(event_subscription.subscription_name)) - - -def test_modify(event_subscription_service: EventSubscriptionService, event_subscription: EventSubscription): - - updated_event_subscription = EventSubscription('test_subscription', [(SourceType.DB_INSTANCE, 'test_instance'), ( - SourceType.DB_CLUSTER, 'db_cluster')], [EventCategory.BACKUP, EventCategory.DELETION, EventCategory.CREATION], 'test_sns_topic_arn', SourceType.DB_INSTANCE) - event_subscription_service.modify( - event_subscription.subscription_name, event_categories=updated_event_subscription.event_categories) - - # Assuming that __eq__ is implemented for EventSubscription - assert updated_event_subscription == event_subscription_service.get_by_id( - event_subscription.subscription_name) - - updated_event_subscription.sns_topic_arn = 'new_sns_topic_arn' - event_subscription_service.modify( - event_subscription.subscription_name, sns_topic_arn=updated_event_subscription.sns_topic_arn) - # Assuming that __eq__ is implemented for EventSubscription - assert updated_event_subscription == event_subscription_service.get_by_id( - event_subscription.subscription_name) - - updated_event_subscription.source_type = SourceType.DB_CLUSTER - event_subscription_service.modify( - event_subscription.subscription_name, source_type=updated_event_subscription.source_type) - - # Assuming that __eq__ is implemented for EventSubscription - assert updated_event_subscription == event_subscription_service.get_by_id( - event_subscription.subscription_name) - - -def test_describe_by_id(event_subscription_service: EventSubscriptionService, event_subscription: EventSubscription): - assert event_subscription.to_dict() == event_subscription_service.describe_by_id( - event_subscription.subscription_name) - - -def test_describe(event_subscription_service: EventSubscriptionService, event_subscription: EventSubscription): - assert [event_subscription.to_dict()] == event_subscription_service.describe( - criteria={'subscription_name': event_subscription.subscription_name}) - - -def test_get(event_subscription_service: EventSubscriptionService, event_subscription: EventSubscription): - # Assuming that __eq__ is implemented for EventSubscription - assert event_subscription == event_subscription_service.get( - {'subscription_name': event_subscription.subscription_name})[0] - - -def test_get_by_id(event_subscription_service: EventSubscriptionService, event_subscription: EventSubscription): - # Assuming that __eq__ is implemented for EventSubscription - assert event_subscription == event_subscription_service.get_by_id( - event_subscription.subscription_name) diff --git a/DB/NEW_KT_DB/Test/GeneralTests.py b/DB/NEW_KT_DB/Test/GeneralTests.py index 65957f2d..e921220c 100644 --- a/DB/NEW_KT_DB/Test/GeneralTests.py +++ b/DB/NEW_KT_DB/Test/GeneralTests.py @@ -1,11 +1,27 @@ - +import json +import os +import sys +import pytest +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 is_file_exist(storage_manager: StorageManager, file_path: str): - return storage_manager.is_file_exist(file_path) +# 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]}" diff --git a/DB/NEW_KT_DB/Test/test_DBInstanceNaive.py b/DB/NEW_KT_DB/Test/test_DBInstanceNaive.py deleted file mode 100644 index 94af2364..00000000 --- a/DB/NEW_KT_DB/Test/test_DBInstanceNaive.py +++ /dev/null @@ -1,124 +0,0 @@ -import os -import sys -import pytest -import sqlite3 -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../'))) -from Service.Classes.DBInstanceNaiveService import DBInstanceManager,DBInstanceService,AlreadyExistsError,ParamValidationError,DBInstanceNotFoundError -from Exceptions.DBInstanceNaiveException import MissingRequireParamError -from Controller.DBInstanceNaiveController import DBInstanceController -from DataAccess.ObjectManager import ObjectManager - -@pytest.fixture -def object_manager(): - return ObjectManager(':memory:') - -@pytest.fixture -def db_instance_manager(object_manager): - return DBInstanceManager(object_manager) - -@pytest.fixture -def db_instance_service(db_instance_manager): - return DBInstanceService(db_instance_manager) - -@pytest.fixture -def snapshot_service(object_manager): - return SnapShotService(SnapShotManager(object_manager)) - -@pytest.fixture -def db_instance_controller(db_instance_service): - return DBInstanceController(db_instance_service) - -def test_create_invalid_identifier(db_instance_controller): - # Test for invalid db_instance_identifier - with pytest.raises(ValueError): - db_instance_controller.create_db_instance( - db_instance_identifier="invalid!@#", - master_username="admin", - master_user_password="password" - ) - -def test_create_missing_required_param(db_instance_controller): - # Test for missing required parameter - with pytest.raises(MissingRequireParamError): - db_instance_controller.create_db_instance( - master_username="admin", - master_user_password="password" - ) - -def test_create_valid_db_instance(db_instance_controller): - attributes = { - "db_instance_identifier": "db123", - "master_username": "admin", - "master_user_password": "password" - } - response = db_instance_controller.create_db_instance(**attributes) - assert response['DBInstance']['db_instance_identifier'] == "db123" - assert response['DBInstance']['master_username'] == "admin" - with pytest.raises(AlreadyExistsError): - db_instance_controller.create_db_instance(**attributes) - -def test_delete_db_instance(db_instance_controller): - attributes = { - "db_instance_identifier": "db123", - "master_username": "admin", - "master_user_password": "password" - } - - db_instance_controller.create_db_instance(**attributes) - - db_instance_controller.delete_db_instance( - db_instance_identifier="db123", - skip_final_snapshot= True - ) - with pytest.raises(DBInstanceNotFoundError): - db_instance_controller.describe_db_instance("db123") - -def test_delete_with_snapshot_invalide_params_db_instance(db_instance_controller,snapshot_service): - attributes = { - "db_instance_identifier": "db123", - "master_username": "admin", - "master_user_password": "password" - } - - db_instance_controller.create_db_instance(**attributes) - - with pytest.raises(ParamValidationError): - db_instance_controller.delete_db_instance( - db_instance_identifier= "db123", - skip_final_snapshot= False - ) - - with pytest.raises(DBInstanceNotFoundError): - db_instance_controller.delete_db_instance( - db_instance_identifier= "invalide_id", - skip_final_snapshot= True - ) - - db_instance_controller.delete_db_instance( - db_instance_identifier="db123", - skip_final_snapshot= False, - final_db_snapshot_identifier="final_db_snapshot_identifier_db123" - ) - - snapshot_service.describe_db_instance("final_db_snapshot_identifier_db123") - -def test_modify_db_instance(db_instance_controller): - attributes = { - "db_instance_identifier": "db123", - "master_username": "admin", - "master_user_password": "password" - } - - db_instance_controller.create_db_instance(**attributes) - - updates = { - "db_instance_identifier": "db123", - "allocated_storage": 50 - } - - response = db_instance_controller.modify_db_instance(**updates) - assert response['DBInstance'].allocated_storage == 50 - -def test_describe_db_instance_not_found(db_instance_controller): - with pytest.raises(DBInstanceNotFoundError): - db_instance_controller.describe_db_instance("non_existent_instance")