diff --git a/DB/NEW_KT_DB/Controller/DBSnapshotControllerNaive.py b/DB/NEW_KT_DB/Controller/DBSnapshotControllerNaive.py new file mode 100644 index 00000000..713a0c6d --- /dev/null +++ b/DB/NEW_KT_DB/Controller/DBSnapshotControllerNaive.py @@ -0,0 +1,47 @@ +import os +import sys +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..",".."))) +from DB.NEW_KT_DB.Service.Classes.DBSnapshotServiceNaive import DBSnapshotServiceNaive +from DB.NEW_KT_DB.Validation.DBSnapshotValidationsNaive import ( + is_valid_db_instance_id, + is_valid_db_snapshot_description, + is_valid_progress +) +class DBSnapshotControllerNaive: + def __init__(self, service: DBSnapshotServiceNaive): + self.service = service + + def create_db_snapshot(self, db_instance_identifier: str, description: str = None, progress: str = None): + # Validate parameters + if not is_valid_db_instance_id(db_instance_identifier): + raise ValueError(f"Invalid db_name: {db_instance_identifier}") + if description and not is_valid_db_snapshot_description(description): + raise ValueError(f"Invalid description: {description}") + if progress and not is_valid_progress(progress): + raise ValueError(f"Invalid progress: {progress}") + + self.service.create(db_instance_identifier, description, progress) + + def delete_db_snapshot(self): + # No validation needed for delete operation + self.service.delete() + + def modify_db_snapshot(self, owner_alias: str = None, status: str = None, + description: str = None, progress: str = None): + # Validate parameters + if description and not is_valid_db_snapshot_description(description): + raise ValueError(f"Invalid description: {description}") + if progress and not is_valid_progress(progress): + raise ValueError(f"Invalid progress: {progress}") + + self.service.modify(owner_alias, status, description, progress) + + + def describe_db_instance(self, db_snapshot_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_snapshot_identifier) \ No newline at end of file diff --git a/DB/NEW_KT_DB/DataAccess/DBManager.py b/DB/NEW_KT_DB/DataAccess/DBManager.py index dd0dde58..3b3fa4b4 100644 --- a/DB/NEW_KT_DB/DataAccess/DBManager.py +++ b/DB/NEW_KT_DB/DataAccess/DBManager.py @@ -11,21 +11,21 @@ def __init__(self, db_file: str): # rachel-8511, ShaniStrassProg def close(self): - '''Close the database connection.''' - self.connection.close() + '''Close the database connection.''' + self.connection.close() # saraNoigershel def execute_query_with_multiple_results(self, query: str) -> Optional[List[Tuple]]: - '''Execute a given query and return the results.''' - try: - c = self.connection.cursor() - c.execute(query) - results = c.fetchall() - # self.connection.commit() ??? - return results if results else None - except OperationalError as e: - raise Exception(f'Error executing query {query}: {e}') + '''Execute a given query and return the results.''' + try: + c = self.connection.cursor() + c.execute(query) + results = c.fetchall() + # self.connection.commit() ??? + return results if results else None + except OperationalError as e: + raise Exception(f'Error executing query {query}: {e}') # ShaniStrassProg @@ -56,42 +56,42 @@ def execute_query_without_results(self, query: str): def create_table(self, table_name, table_structure): '''create a table in a given db by given table_structure''' create_statement = f'''CREATE TABLE IF NOT EXISTS {table_name} ({table_structure})''' - execute_query_without_results(create_statement) + self.execute_query_without_results(create_statement) # Riki7649255 based on rachel-8511, ShaniStrassProg def insert_data_into_table(self, table_name, data): insert_statement = f'''INSERT INTO {table_name} VALUES {data}''' - execute_query_without_results(insert_statement) + self.execute_query_without_results(insert_statement) # Riki7649255 based on rachel-8511, Shani def update_records_in_table(self, table_name: str, updates: Dict[str, Any], criteria: str) -> None: - '''Update records in the specified table based on criteria.''' - - # add documentation here - set_clause = ', '.join([f'{k} = ?' for k in updates.keys()]) - values = list(updates.values()) - - update_statement = f''' - UPDATE {table_name} - SET {set_clause} - WHERE {criteria} - ''' - - execute_query_without_results(update_statement) + '''Update records in the specified table based on criteria.''' + + # add documentation here + set_clause = ', '.join([f'{k} = ?' for k in updates.keys()]) + values = list(updates.values()) + + update_statement = f''' + UPDATE {table_name} + SET {set_clause} + WHERE {criteria} + ''' + + self.execute_query_without_results(update_statement) # Riki7649255 based on rachel-8511 def delete_data_from_table(self, table_name: str, criteria: str) -> None: - '''Delete a record from the specified table based on criteria.''' - - delete_statement = f''' - DELETE FROM {table_name} - WHERE {criteria} - ''') + '''Delete a record from the specified table based on criteria.''' - execute_query_without_results(delete_statement) + delete_statement = f''' + DELETE FROM {table_name} + WHERE {criteria} + ''' + + self.execute_query_without_results(delete_statement) # rachel-8511, Riki7649255 @@ -109,7 +109,7 @@ def select_and_return_records_from_table(self, table_name: str, columns: List[st if criteria: query += f' WHERE {criteria}' try: - results = execute_query_with_multiple_results(query) + results = self.execute_query_with_multiple_results(query) return {result[0]: dict(zip(columns, result[1:])) for result in results} except OperationalError as e: raise Exception(f'Error selecting from {table_name}: {e}') @@ -120,7 +120,7 @@ def describe_table(self, table_name: str) -> Dict[str, str]: '''Describe table structure.''' try: desc_statement = f'PRAGMA table_info({table_name})' - columns = execute_query_with_multiple_results(desc_statement) + columns = self.execute_query_with_multiple_results(desc_statement) return {col[1]: col[2] for col in columns} except OperationalError as e: raise Exception(f'Error describing table {table_name}: {e}') diff --git a/DB/NEW_KT_DB/DataAccess/DBSnapshotManagerNaive.py b/DB/NEW_KT_DB/DataAccess/DBSnapshotManagerNaive.py new file mode 100644 index 00000000..e7e3510d --- /dev/null +++ b/DB/NEW_KT_DB/DataAccess/DBSnapshotManagerNaive.py @@ -0,0 +1,67 @@ +import json +import sqlite3 +import os +import sys +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..",".."))) +from typing import Dict, Any +from DB.NEW_KT_DB.Validation.DBSnapshotValidationsNaive import is_valid_db_instance_id +from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager +from DB.NEW_KT_DB.Models.DBSnapshotModelNaive import SnapshotNaive + +class DBSnapshotManagerNaive: + + def __init__(self, object_manager: ObjectManager): + self.object_manager = object_manager + # Create the management table for DBSnapshot using its object name and table structure + self.object_manager.create_management_table(SnapshotNaive.object_name, SnapshotNaive.table_structure) + + def createInMemoryDBSnapshot(self, db_instance_identifier: SnapshotNaive): + # Validate db_instance_identifier + if not is_valid_db_instance_id(db_instance_identifier): + raise ValueError(f"Invalid db_instance_identifier: {db_instance_identifier}") + + self.object_manager.save_in_memory(SnapshotNaive.object_name, db_instance_identifier.to_sql()) + + def deleteInMemoryDBSnapshot(self, db_snapshot_identifier: str): + # Validate db_instance_identifier + if not is_valid_db_instance_id(db_snapshot_identifier, 15): + raise ValueError(f"Invalid db_instance_identifier: {db_snapshot_identifier}") + + self.object_manager.delete_from_memory( + pk_column = SnapshotNaive.pk_column, + pk_value = db_snapshot_identifier, + object_name = SnapshotNaive.object_name + ) + + def describeDBSnapshot(self, db_snapshot_identifier: str): + # Validate db_instance_identifier + if not is_valid_db_instance_id(db_snapshot_identifier, 15): + raise ValueError(f"Invalid db_instance_identifier: {db_snapshot_identifier}") + + self.object_manager.get_from_memory( + criteria=f"{SnapshotNaive.pk_column} = '{db_snapshot_identifier}'", + object_name=SnapshotNaive.object_name, + columns='*' + ) + + def modifyDBSnapshot(self, db_snapshot_identifier: str, updates: str): + # Validate db_instance_identifier + if not is_valid_db_instance_id(db_snapshot_identifier, 15): + raise ValueError(f"Invalid db_instance_identifier: {db_snapshot_identifier}") + + # Assuming new_data contains fields to update, you might want to validate these fields as well + # For example: + # if 'description' in new_data and not is_valid_db_snapshot_description(new_data['description']): + # raise ValueError(f"Invalid description: {new_data['description']}") + + self.object_manager.update_in_memory( + criteria=f"{SnapshotNaive.pk_column} = '{db_snapshot_identifier}'", + object_name=SnapshotNaive.object_name, + updates=updates + ) + + def is_db_snapshot_exist(self, db_snapshot_identifier: int) -> bool: + return bool(self.object_manager.db_manager.is_object_exist( + self.object_manager._convert_object_name_to_management_table_name(SnapshotNaive.object_name), + criteria=f"{SnapshotNaive.pk_column} = '{db_snapshot_identifier}'" + )) diff --git a/DB/NEW_KT_DB/DataAccess/ObjectManager.py b/DB/NEW_KT_DB/DataAccess/ObjectManager.py index 56c6948f..6f4e1179 100644 --- a/DB/NEW_KT_DB/DataAccess/ObjectManager.py +++ b/DB/NEW_KT_DB/DataAccess/ObjectManager.py @@ -1,7 +1,11 @@ -from typing import Dict, Any import json import sqlite3 -from DBManager import DBManager +import os +import sys +from typing import Dict, Any +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..",".."))) +from DB.NEW_KT_DB.DataAccess.DBManager import DBManager + class ObjectManager: def __init__(self, db_file: str): @@ -12,7 +16,7 @@ def __init__(self, db_file: str): # for internal use only: # Riki7649255 based on rachel-8511 - def create_management_table(self, table_name, table_structure='object_id INTEGER PRIMARY KEY AUTOINCREMENT,type_object TEXT NOT NULL,metadata TEXT NOT NULL') + def create_management_table(self, table_name, table_structure='object_id INTEGER PRIMARY KEY AUTOINCREMENT,type_object TEXT NOT NULL,metadata TEXT NOT NULL'): self.db_manager.create_table(table_name, table_structure) @@ -58,9 +62,9 @@ def convert_object_name_to_management_table_name(object_name): return f'mng_{object_name}s' - def is_management_table_exist(table_name): + def is_management_table_exist(self, table_name): # check if table exists using single result query - return db_manager.execute_query_with_single_result(f'desc table {table_name}') + return self.db_manager.execute_query_with_single_result(f'desc table {table_name}') # for outer use: @@ -68,12 +72,12 @@ def save_in_memory(self, object): # insert object info into management table mng_{object_name}s # for exmple: object db_instance will be saved in table mng_db_instances - table_name = convert_object_name_to_management_table_name(self.object_name) + table_name = self.convert_object_name_to_management_table_name(self.object_name) - if not is_management_table_exist(table_name): - create_management_table(table_name) + if not self.is_management_table_exist(table_name): + self.create_management_table(table_name) - insert_object_to_management_table(table_name, object) + self.insert_object_to_management_table(table_name, object) def delete_from_memory(self,criteria='default'): @@ -82,9 +86,9 @@ def delete_from_memory(self,criteria='default'): if criteria == 'default': criteria = f'{self.pk_column} = {self.pk_value}' - table_name = convert_object_name_to_management_table_name(self.object_name) + table_name = self.convert_object_name_to_management_table_name(self.object_name) - delete_data_from_table(table_name, criteria) + self.db_manager.delete_data_from_table(table_name, criteria) def update_in_memory(self, updates, criteria='default'): @@ -93,13 +97,13 @@ def update_in_memory(self, updates, criteria='default'): if criteria == 'default': criteria = f'{self.pk_column} = {self.pk_value}' - table_name = convert_object_name_to_management_table_name(self.object_name) + table_name = self.convert_object_name_to_management_table_name(self.object_name) - update_object_in_management_table_by_criteria(table_name, updates, criteria) + self.update_object_in_management_table_by_criteria(table_name, updates, criteria) def get_from_memory(self): - get_object_from_management_table(self.object_id) + self.get_object_from_management_table(self.object_id) def convert_object_attributes_to_dictionary(**kwargs): diff --git a/DB/NEW_KT_DB/Models/DBSnapshotModelNaive.py b/DB/NEW_KT_DB/Models/DBSnapshotModelNaive.py new file mode 100644 index 00000000..f7691dde --- /dev/null +++ b/DB/NEW_KT_DB/Models/DBSnapshotModelNaive.py @@ -0,0 +1,70 @@ +import os +import sys +import json +from datetime import datetime +from typing import Dict, Optional +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..",".."))) +from DB.NEW_KT_DB.DataAccess import ObjectManager +from DB.NEW_KT_DB.Validation.DBSnapshotValidationsNaive import is_valid_db_instance_id, is_valid_db_snapshot_description, is_valid_progress, is_valid_date, is_valid_url_parameter + +class SnapshotNaive: + BASE_PATH = "db_snapshot" + object_name = "db_snapshot_naive" + pk_column = "db_snapshot_id" + pk_column_data_type = 'TEXT' + table_structure = f''' + db_instance_identifier TEXT PRIMARY KEY, + metadata TEXT NOT NULL + ''' + def __init__(self, db_snapshot_identifier, db_instance_identifier: str, creation_date: datetime, owner_alias: str, status: str, + description: Optional[str] = None, progress: Optional[str] = None, url_snapshot: Optional[str] = None): + + # Validate parameters + if not is_valid_db_instance_id(db_instance_identifier): + raise ValueError(f"Invalid db_instance_identifier: {db_instance_identifier}") + if not is_valid_date(creation_date.strftime('%Y-%m-%d')): + raise ValueError(f"Invalid creation_date: {creation_date}") + if description: + if not is_valid_db_snapshot_description(description): + raise ValueError(f"Invalid description: {description}") + if progress: + if not is_valid_progress(progress): + raise ValueError(f"Invalid progress: {progress}") + if url_snapshot: + if not is_valid_url_parameter(url_snapshot): + raise ValueError(f"Invalid url_snapshot: {url_snapshot}") + self.db_snapshot_identifier = db_snapshot_identifier + self.db_instance_identifier = db_instance_identifier + self.creation_date = creation_date + self.owner_alias = owner_alias + self.status = status + self.description = description + self.progress = progress + self.url_snapshot = url_snapshot + self.object_name = "Snapshot" + self.table_structure = {} + + def to_dict(self) -> Dict: + '''Retrieve the data of the DB snapshot as a dictionary.''' + + return ObjectManager.convert_object_attributes_to_dictionary( + db_snapshot_identifier = self.db_snapshot_identifier, + db_instance_identifier = self.db_instance_identifier, + creation_date = self.creation_date, + owner_alias = self.owner_alias, + status = self.status, + description = self.description, + progress = self.progress, + url_snapshot = self.url_snapshot, + object_name = self.object_name, + table_structure = self.table_structure + ) + + def to_sql(self): + # Convert the model snapshot 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()) + ')' + return values + + diff --git a/DB/NEW_KT_DB/Service/Classes/DBSnapshotServiceNaive.py b/DB/NEW_KT_DB/Service/Classes/DBSnapshotServiceNaive.py new file mode 100644 index 00000000..1440ca89 --- /dev/null +++ b/DB/NEW_KT_DB/Service/Classes/DBSnapshotServiceNaive.py @@ -0,0 +1,155 @@ +import sqlite3 +import shutil +import os +import sys +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..","..",".."))) +from typing import Dict, Optional +from datetime import datetime +from DB.NEW_KT_DB.Models.DBSnapshotModelNaive import SnapshotNaive +from DB.NEW_KT_DB.Service.Abc.DBO import DBO +from DB.NEW_KT_DB.Validation.DBSnapshotValidationsNaive import is_valid_db_instance_id, is_valid_db_snapshot_description, is_valid_progress +from DB.NEW_KT_DB.DataAccess.DBSnapshotManagerNaive import DBSnapshotManagerNaive + +class DBSnapshotServiceNaive(DBO): + def __init__(self, dal: DBSnapshotManagerNaive): + self.dal = dal + + def create(self, db_instance_identifier: str, description: str, progress: str): + '''Create a new DBSnapshot.''' + # Validate parameters\ + print('Enter to create') + if not is_valid_db_instance_id(db_instance_identifier): + raise ValueError(f"Invalid db_instance_identifier: {db_instance_identifier}") + if not is_valid_db_snapshot_description(description): + raise ValueError(f"Invalid description: {description}") + if not is_valid_progress(progress): + raise ValueError(f"Invalid progress: {progress}") + # Create a timestamp with the current date and time + current_timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + # Get the current username + owner_alias = os.getlogin() + + db_object = self.dal.describeDBSnapshot(db_instance_identifier) + # Define the file paths for snapshot + # db_object = describe(db_instance_identifier) + + + + db_instance_directory = db_object.BASE_PATH + '\\' + db_object.endpoint + snapshot_db_path = f"../snapshot/{db_instance_identifier}_{current_timestamp}.db" + if not os.path.exists("../snapshot/"): + os.makedirs("../snapshot/") + + db_snapshot = SnapshotNaive(db_instance_identifier, creation_date=datetime.now(), owner_alias=owner_alias, status='inital', + description=description, progress=progress, url_snapshot=snapshot_db_path) + + assert self.db_snapshot is not None, "SnapshotNaive object was not created." + + shutil.copytree(db_instance_directory, snapshot_db_path) + + self.dal.createInMemoryDBSnapshot(db_snapshot) + return {'DBSnapshot': db_snapshot.to_dict()} + + def delete(self, db_snapshot_identifier: str): + '''Delete an existing DBCluster.''' + # Validate snapshot_name + if not is_valid_db_instance_id(db_snapshot_identifier): + raise ValueError(f"Invalid snapshot_name: {db_snapshot_identifier}") + + # Delete physical object + snapshot_path = f"../snapshot/{db_snapshot_identifier}.db" + db_snapshot = self.get(db_snapshot_identifier) + if os.path.exists(snapshot_path): + os.remove(snapshot_path) + else: + print(f"Snapshot {db_snapshot_identifier} does not exist.") + # Handle an error + + # Delete from memory + self.dal.deleteInMemoryDBSnapshot(db_snapshot_identifier) + + def describe(self, db_snapshot_identifier: str): + '''Describe the details of the DB snapshot.''' + if not self.dal.is_db_snapshot_exist(db_snapshot_identifier): + # raise DBSnapshotNotFoundError('This DB instance identifier does not exist') + pass + + describe_db_snapshot = self.dal.describeDBSnapshot(db_snapshot_identifier)[0] + describe_db_snapshot_dict = { + 'db_instance_identifier': describe_db_snapshot[0], + 'creation_date': describe_db_snapshot[1], + 'owner_alias': describe_db_snapshot[2], + 'status': describe_db_snapshot[3], + 'description': describe_db_snapshot[4], + 'progress': describe_db_snapshot[5], + 'url_snapshot': describe_db_snapshot[6], + 'object_name': describe_db_snapshot[7], + 'table_structure': describe_db_snapshot[8] + } + return {'DBSnapshot': describe_db_snapshot_dict} + + + + + + def modify(self, db_snapshot_identifier: str, owner_alias: Optional[str] = None, status: Optional[str] = None, + description: Optional[str] = None, progress: Optional[str] = None): + '''Modify an existing DBCluster.''' + updates = { + 'db_snapshot_identifier': db_snapshot_identifier, + 'owner_alias': owner_alias, + 'status': status, + 'description': description, + 'progress': progress + } + + # Validate specific parameters + if description and not is_valid_db_snapshot_description(description): + raise ValueError(f"Invalid description: {description}") + if progress and not is_valid_progress(progress): + raise ValueError(f"Invalid progress: {progress}") + + required_params = ['db_snapshot_identifier'] + all_params = ['owner_alias', 'status', 'description', 'progress'] + all_params.extend(required_params) + + filtered_updates = {key: value for key, value in updates.items() if key != 'db_snapshot_identifier'} + set_clause = ', '.join([f"{key} = '{value}'" for key, value in filtered_updates.items()]) + + + if owner_alias is not None: + self.owner_alias = owner_alias + if status is not None: + self.status = status + if description is not None: + self.description = description + if progress is not None: + self.progress = progress + + self.dal.modifyDBSnapshot(db_snapshot_identifier, set_clause) + + def get(self, db_snapshot_indentifier): + '''Get code object.''' + # Return real-time object + + describe_result = self.describe(db_snapshot_indentifier) + if describe_result: + describe_result = describe_result['DBSnapshot'] + return SnapshotNaive( + describe_result['db_snapshot_identifier'], + describe_result['db_instance_identifier'], + describe_result['creation_date'], + describe_result['owner_alias'], + describe_result['status'], + describe_result['description'], + describe_result['progress'], + describe_result['url_snapshot'] + ) + + return None + + + + + + diff --git a/DB/NEW_KT_DB/Test/DBSnapshotTestsNaive.py b/DB/NEW_KT_DB/Test/DBSnapshotTestsNaive.py new file mode 100644 index 00000000..f97c5ca0 --- /dev/null +++ b/DB/NEW_KT_DB/Test/DBSnapshotTestsNaive.py @@ -0,0 +1,90 @@ +import unittest +from unittest.mock import MagicMock, patch +from datetime import datetime +import os +import shutil +import sys +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))) + +from DB.NEW_KT_DB.DataAccess.DBSnapshotManagerNaive import DBSnapshotManagerNaive +from DB.NEW_KT_DB.Service.Classes.DBSnapshotServiceNaive import DBSnapshotServiceNaive + +class TestDBSnapshotService(unittest.TestCase): + def setUp(self): + self.dal_mock = MagicMock(spec=DBSnapshotManagerNaive) + self.service = DBSnapshotServiceNaive(self.dal_mock) + + @patch('os.getlogin', return_value='test_user') + @patch('shutil.copytree') + @patch('DB.NEW_KT_DB.Service.Classes.DBSnapshotServiceNaive.DBSnapshotServiceNaive.describe', return_value=MagicMock(BASE_PATH='path', endpoint='endpoint')) + # def test_create(self, describe_mock, copytree_mock, getlogin_mock): + def test_create(self, describe_mock, copytree_mock, getlogin_mock): + db_instance_identifier = 'test-db-id' + description = 'Test snapshot' + progress = '50%' + + with patch('DB.NEW_KT_DB.Models.DBSnapshotModelNaive.SnapshotNaive') as snapshot_mock: + snapshot_instance = snapshot_mock.return_value + self.dal_mock.createInMemoryDBSnapshot.return_value = True + + result = self.service.create(db_instance_identifier, description, progress) + + snapshot_mock.assert_called_once_with( + db_instance_identifier=db_instance_identifier, + creation_date=datetime.now(), + owner_alias='test_user', + description=description, + progress=progress, + url_snapshot=f"../snapshot/{db_instance_identifier}_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.db" + ) + self.dal_mock.createInMemoryDBSnapshot.assert_called_once() + self.assertTrue(result) + + # @patch('os.path.exists', return_value=True) + # @patch('os.remove') + # def test_delete(self, remove_mock, exists_mock): + # snapshot_name = 'test-snapshot' + + # self.dal_mock.deleteInMemoryDBSnapshot.return_value = True + + # result = self.service.delete(snapshot_name) + + # exists_mock.assert_called_once_with(f"../snapshot/{snapshot_name}.db") + # remove_mock.assert_called_once_with(f"../snapshot/{snapshot_name}.db") + # self.dal_mock.deleteInMemoryDBSnapshot.assert_called_once() + # self.assertTrue(result) + + # @patch('DB.NEW_KT_DB.DataAccess.DBSnapshotManagerNaive.DBSnapshotManagerNaive.describeDBSnapshot', return_value=MagicMock()) + + # def test_describe(self, describe_mock): + # db_instance_identifier = 'test-db-id' + # result = self.service.describe(db_instance_identifier) + + # describe_mock.assert_called_once_with(db_instance_identifier) + # self.assertEqual(result, {}) + + # @patch('DB.NEW_KT_DB.Models.DBSnapshotModelNaive', return_value=MagicMock(to_dict=MagicMock(return_value={}))) + # def test_modify(self, snapshot_mock): + # owner_alias = 'new_owner' + # status = 'active' + # description = 'Updated description' + # progress = '75%' + + # self.service.modify(owner_alias=owner_alias, status=status, description=description, progress=progress) + + # self.assertEqual(self.service.owner_alias, owner_alias) + # self.assertEqual(self.service.status, status) + # self.assertEqual(self.service.description, description) + # self.assertEqual(self.service.progress, progress) + + # def test_get(self): + # self.service.db_snapshot = None + # result = self.service.get() + # self.assertIsNone(result) + + # self.service.db_snapshot = MagicMock(to_dict=MagicMock(return_value={'key': 'value'})) + # result = self.service.get() + # self.assertEqual(result, {'key': 'value'}) + +if __name__ == '__main__': + unittest.main() diff --git a/DB/NEW_KT_DB/Validation/DBClusterValiditions.py b/DB/NEW_KT_DB/Validation/DBClusterValiditions.py index fecf15d2..e501a3ad 100644 --- a/DB/NEW_KT_DB/Validation/DBClusterValiditions.py +++ b/DB/NEW_KT_DB/Validation/DBClusterValiditions.py @@ -1,5 +1,5 @@ import re -from GeneralValidations import +from GeneralValidations import is_length_in_range def is_db_cluster_name_valid(cluster_name): diff --git a/DB/NEW_KT_DB/Validation/DBSnapshotValidationsNaive.py b/DB/NEW_KT_DB/Validation/DBSnapshotValidationsNaive.py new file mode 100644 index 00000000..5e07a465 --- /dev/null +++ b/DB/NEW_KT_DB/Validation/DBSnapshotValidationsNaive.py @@ -0,0 +1,34 @@ +import re +import os +import sys +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..",".."))) +from DB.NEW_KT_DB.Validation.GeneralValidations import is_length_in_range, is_valid_number, is_valid_db_instance_identifier +from typing import Optional,Dict + +def is_valid_db_snapshot_description(description_snapshot: str) -> bool: + return is_length_in_range(description_snapshot, 1, 40) + + +def is_valid_progress(progress: str) -> bool: + num_of_progress = progress[:-1] + num_of_progress_int = int(num_of_progress) + return is_valid_number(num_of_progress_int, 0, 100) + + +def is_valid_date(date) -> bool: + pattern = r'^\d{4}-\d{2}-\d{2}$' # Date format: YYYY-MM-DD + return bool(re.match(pattern, date)) + +# db_instance_identifier + +def is_valid_db_instance_id(db_instance_identifier: str) -> bool: + return is_valid_db_instance_identifier(db_instance_identifier, 15) + + +def is_valid_url_parameter(url_snapshot: str) -> bool: + '''Check if the url_snapshot parameter is valid.''' + # pattern = r'^[\w\-]+(?:%[0-9A-Fa-f]{2})*$' + # return bool(re.match(pattern, url_snapshot)) + return True + +