diff --git a/DB/NEW_KT_DB/Controller/DBProxyEndpointController.py b/DB/NEW_KT_DB/Controller/DBProxyEndpointController.py index 4acc23e9..d8fbfef5 100644 --- a/DB/NEW_KT_DB/Controller/DBProxyEndpointController.py +++ b/DB/NEW_KT_DB/Controller/DBProxyEndpointController.py @@ -7,9 +7,11 @@ def __init__(self, service: DBProxyEndpointService): self.service = service - def create_db_proxy_endpoint(self, DBProxyName:str, DBProxyEndpointName:str, TargetRole:str = 'READ_WRITE', Tags:Optional[List[Dict[str, str]]] = None): + def create_db_proxy_endpoint(self, DBProxyName:str, DBProxyEndpointName:str, VpcSubnetIds:List[str], + VpcSecurityGroupIds:Optional[List[str]] = None, TargetRole:str = 'READ_WRITE', Tags:Optional[List[Dict[str, str]]] = None): """Create a db proxy endpoint""" - return self.service.create(DBProxyName, DBProxyEndpointName, TargetRole, Tags) + return self.service.create(DBProxyName, DBProxyEndpointName,VpcSubnetIds, + VpcSecurityGroupIds, TargetRole, Tags) def delete_db_proxy_endpoint(self, DBProxyEndpointName:str): @@ -17,9 +19,10 @@ def delete_db_proxy_endpoint(self, DBProxyEndpointName:str): return self.service.delete(DBProxyEndpointName) - def modify_db_proxy_endpoint(self, DBProxyEndpointName:str, NewDBProxyEndpointName:Optional[str] = None): + def modify_db_proxy_endpoint(self, DBProxyEndpointName:str, NewDBProxyEndpointName:Optional[str] = None, + VpcSubnetIds:Optional[List[str]] = None): """Modify a db proxy endpoint""" - return self.service.modify(DBProxyEndpointName, NewDBProxyEndpointName) + return self.service.modify(DBProxyEndpointName, NewDBProxyEndpointName, VpcSubnetIds) def describe_db_proxy_endpoint(self, diff --git a/DB/NEW_KT_DB/DBs/mainDB.db b/DB/NEW_KT_DB/DBs/mainDB.db index 55373148..5677567a 100644 Binary files a/DB/NEW_KT_DB/DBs/mainDB.db and b/DB/NEW_KT_DB/DBs/mainDB.db differ diff --git a/DB/NEW_KT_DB/DataAccess/DBProxyEndpointManager.py b/DB/NEW_KT_DB/DataAccess/DBProxyEndpointManager.py index c67d5714..e3b2349f 100644 --- a/DB/NEW_KT_DB/DataAccess/DBProxyEndpointManager.py +++ b/DB/NEW_KT_DB/DataAccess/DBProxyEndpointManager.py @@ -2,8 +2,7 @@ from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager from DB.NEW_KT_DB.Models.DBProxyEndpointModel import DBProxyEndpoint import json - - +import ast class DBProxyEndpointManager: # Static functions @@ -66,7 +65,7 @@ def map_query_data_to_col_value_dict(data, cols: Optional[List[str]] = None): """ if not cols: cols = DBProxyEndpointManager.convert_table_structure_to_columns_arr(DBProxyEndpoint.table_structure) - data_mapping = {col: val for col, val in zip(cols, data)} + data_mapping = {col: ast.literal_eval(val) if (isinstance(val, str) and val[0] == '[' and val[-1] == ']') else val for col, val in zip(cols, data)} return data_mapping # cast columns arr to str for query @@ -76,7 +75,7 @@ def map_query_data_to_col_value_dict(data, cols: Optional[List[str]] = None): # Select one object by its unique name if name: error = f"db proxy endpoint with name '{name}' not found" - data = self.object_manager.get_from_memory(DBProxyEndpoint.object_name, columns, criteria=f'{DBProxyEndpoint.pk_column} = "{name}"') + data = self.object_manager.get_from_memory(DBProxyEndpoint.object_name, columns, criteria=f"{DBProxyEndpoint.pk_column} = '{name}'") # Select all objects else: error = f"there is no objects in table of {DBProxyEndpoint.object_name}" @@ -115,7 +114,9 @@ def describe(self, name: Optional[str] = None, Filters:Optional[List[Dict[str, A description = self.get_object_attributes_dict() # If there are filters return only objects that in conditions of all filters if Filters: - description = [obj for obj in description if [col for col in obj.keys() if col not in Filters or obj[col] in Filters[col]] != []] + for Filter in Filters: + description = [obj for obj in description if all(col not in Filter['Name'] or obj[col] in Filter['Values'] for col in obj.keys())] + return {DBProxyEndpoint.object_name: description} diff --git a/DB/NEW_KT_DB/Integration/DBProxyEndpointIntegration.py b/DB/NEW_KT_DB/Integration/DBProxyEndpointIntegration.py new file mode 100644 index 00000000..d74d2869 --- /dev/null +++ b/DB/NEW_KT_DB/Integration/DBProxyEndpointIntegration.py @@ -0,0 +1,125 @@ +from datetime import datetime +import os +import sys +import pytest +from unittest.mock import MagicMock +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..'))) +from DB.NEW_KT_DB.Controller.DBProxyEndpointController import DBProxyEndpointController +from DB.NEW_KT_DB.Service.Classes.DBProxyEndpointService import DBProxyEndpointService +from DB.NEW_KT_DB.DataAccess.DBProxyEndpointManager import DBProxyEndpointManager +from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager +from Storage.NEW_KT_Storage.DataAccess.StorageManager import StorageManager +from DB.NEW_KT_DB.Exceptions.DBProxyEndpointExceptions import DBProxyEndpointNotFoundException + +def db_proxy_service_mock(): + mock = MagicMock() + mock.is_exists.side_effect = lambda db_proxy_name: db_proxy_name == 'my-proxy' + return mock + +# Dependencies injection +object_manager = ObjectManager('DB/NEW_KT_DB/DBs/mainDB.db') +storage_manager = StorageManager("dbProxyEndpoints") +endpoint_manager = DBProxyEndpointManager(object_manager) +endpoint_service = DBProxyEndpointService(endpoint_manager, storage_manager, db_proxy_service_mock()) +endpoint_controller = DBProxyEndpointController(endpoint_service) + +DB_PROXY_ENDPOINT_NAME = "endpoint" +NEW_DB_PROXY_ENDPOINT_NAME = "our-endpoint" +DB_PROXY_NAME = "my-proxy" +VPC_SUBNET_IDS = ['subnet-12345678', 'subnet-87654321'] +TARGET_ROLE = "READ_WRITE" + +# Demonstrate all dbproxy endpoint functionallity + +print('''---------------------Start Of session----------------------''') +print() + + +print(f'''{datetime.now()} deonstration of object db proxy endpoint start''') +print() +print("----------------------------------------------------------------") + +# create +start_time = datetime.now() +print(f'''{start_time} going to create db proxy endpoint names "endpoint" in db proxy "my proxy"''') +res = endpoint_controller.create_db_proxy_endpoint(DB_PROXY_NAME, DB_PROXY_ENDPOINT_NAME, VPC_SUBNET_IDS, TARGET_ROLE) +end_time = datetime.now() +print(f'''{end_time} db proxy endpoint "endpoint" created successfully''') +duration = end_time - start_time +print("duration: ",duration) +print("----------------------------------------------------------------") + +# Test create +print("-----Test create---------") +print("Response: ",res) +print(endpoint_controller.describe_db_proxy_endpoint(DB_PROXY_ENDPOINT_NAME)) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_create_db_proxy_endpoint']) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_create_when_db_proxy_not_exist']) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_create_db_proxy_endpoint_with_existing_name']) +print("----------------------------------------------------------------") + +# Delete +start_time = datetime.now() +print(f'''{start_time} going to delete db db proxy endpoint "endpoint"''') +res = endpoint_controller.delete_db_proxy_endpoint(DB_PROXY_ENDPOINT_NAME) +end_time = datetime.now() +print(f'''{end_time} db proxy endpoint "endpoint" deleted successfully''') +duration = end_time - start_time +print("duration: ",duration) +print("----------------------------------------------------------------") + +# Test delete +print("-----Test delete---------") +try: + print(endpoint_controller.describe_db_proxy_endpoint(DB_PROXY_ENDPOINT_NAME)) +except DBProxyEndpointNotFoundException as e: + print(e) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_delete_db_proxy_endpoint']) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_delete_non_valid_state_db_proxy_endpoint']) +print("----------------------------------------------------------------") + + +# Modify +endpoint_before_modify = endpoint_controller.create_db_proxy_endpoint(DB_PROXY_NAME, DB_PROXY_ENDPOINT_NAME, VPC_SUBNET_IDS, TARGET_ROLE) +start_time = datetime.now() +print(f'''{start_time} going to modify db proxy endpoint "endpoint" to name "our-endpoint"''') +endpoint_after_modify = endpoint_controller.modify_db_proxy_endpoint(DB_PROXY_ENDPOINT_NAME, NEW_DB_PROXY_ENDPOINT_NAME) +end_time = datetime.now() +print(f'''{end_time} db db proxy endpoint "endpoint" modified successfully''') +duration = end_time - start_time +print("duration: ",duration) +print("----------------------------------------------------------------") + +# Test modify +print("Test modify") +print("before modify:") +print(endpoint_before_modify) +print("after modify:") +print(endpoint_after_modify) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_modify_name_to_db_proxy_endpoint']) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_modify_non_exist_db_proxy_endpoint']) + +# Describe +start_time = datetime.now() +print(f'''{start_time} going to describe db proxy endpoint "our-endpoint"''') +print(endpoint_controller.describe_db_proxy_endpoint(NEW_DB_PROXY_ENDPOINT_NAME)) +end_time = datetime.now() +duration = end_time - start_time +print(f'''{end_time} db proxy endpoint "our-endpoint" was described successfully''') +print("duration: ",duration) +print("----------------------------------------------------------------") + +# Test describe +print("Test describe") +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_describe_db_proxy_endpoint']) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_describe_with_filters']) +pytest.main(['-q', 'DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py::test_describe_with_non_correct_filters']) + +print(f'''{datetime.now()} deonstration of object db proxy endpoint ended successfully''') +print('''---------------------End Of session----------------------''') + +# delete from memory all integration leftovers +endpoint_controller.delete_db_proxy_endpoint(NEW_DB_PROXY_ENDPOINT_NAME) + + + diff --git a/DB/NEW_KT_DB/Models/DBProxyEndpointModel.py b/DB/NEW_KT_DB/Models/DBProxyEndpointModel.py index 6cacd3c8..4f353885 100644 --- a/DB/NEW_KT_DB/Models/DBProxyEndpointModel.py +++ b/DB/NEW_KT_DB/Models/DBProxyEndpointModel.py @@ -12,6 +12,8 @@ class DBProxyEndpoint: table_structure = f""" DBProxyEndpointName VARCHAR(63) PRIMARY KEY NOT NULL, DBProxyName VARCHAR(63) NOT NULL, + VpcSubnetIds TEXT NOT NULL, + VpcSecurityGroupIds TEXT NULL, TargetRole VARCHAR(10) NOT NULL, Tags JSONB DEFAULT '{{}}', Status VARCHAR(20) NOT NULL, @@ -22,17 +24,21 @@ class DBProxyEndpoint: def __init__(self, - DBProxyEndpointName:str, - DBProxyName:str, - TargetRole:Optional[str] = None, - Tags:Optional[List[Dict[str, str]]] = None, - Status:str = 'creating', - created_date = datetime.now(), - endpoint:str ='', - IsDefault:bool = False): + DBProxyEndpointName:str, + DBProxyName:str, + VpcSubnetIds:List[str], + VpcSecurityGroupIds:Optional[List[str]] = None, + TargetRole:Optional[str] = None, + Tags:Optional[List[Dict[str, str]]] = None, + Status:str = 'creating', + created_date = datetime.now(), + endpoint:str ='', + IsDefault:bool = False): self.DBProxyName=DBProxyName self.DBProxyEndpointName=DBProxyEndpointName - self.TargetRole=TargetRole + self.VpcSubnetIds = VpcSubnetIds + self.VpcSecurityGroupIds = VpcSecurityGroupIds + self.TargetRole = TargetRole self.Tags = Tags self.Status = Status self.CreatedDate = created_date @@ -47,6 +53,8 @@ def to_dict(self) -> Dict: return ObjectManager.convert_object_attributes_to_dictionary( DBProxyEndpointName = self.DBProxyEndpointName, DBProxyName = self.DBProxyName, + VpcSubnetIds = self.VpcSubnetIds, + VpcSecurityGroupIds = self.VpcSecurityGroupIds, TargetRole = self.TargetRole, Tags = self.Tags, Status = self.Status, diff --git a/DB/NEW_KT_DB/Service/Classes/DBProxyEndpointService.py b/DB/NEW_KT_DB/Service/Classes/DBProxyEndpointService.py index 1ed43e03..b581c918 100644 --- a/DB/NEW_KT_DB/Service/Classes/DBProxyEndpointService.py +++ b/DB/NEW_KT_DB/Service/Classes/DBProxyEndpointService.py @@ -35,7 +35,8 @@ def custom_serializer(obj): return object_json - def create(self, DBProxyName:str, DBProxyEndpointName:str, TargetRole:str = 'READ_WRITE', Tags:Optional[List[Dict[str, str]]] = None, IsDefault:bool = False): + def create(self, DBProxyName:str, DBProxyEndpointName:str, VpcSubnetIds:List[str], + VpcSecurityGroupIds:Optional[List[str]] = None, TargetRole:str = 'READ_WRITE', Tags:Optional[List[Dict[str, str]]] = None, IsDefault:bool = False): '''Create a new DBProxy endpoint.''' # Validations if not validate_name(DBProxyName): @@ -52,7 +53,7 @@ def create(self, DBProxyName:str, DBProxyEndpointName:str, TargetRole:str = 'REA raise DBProxyEndpointAlreadyExistsException(DBProxyEndpointName) # create object - db_proxy_endpoint:DBProxyEndpoint = DBProxyEndpoint(DBProxyEndpointName, DBProxyName, TargetRole, Tags, IsDefault=IsDefault) + db_proxy_endpoint:DBProxyEndpoint = DBProxyEndpoint(DBProxyEndpointName, DBProxyName, VpcSubnetIds, VpcSecurityGroupIds, TargetRole, Tags, IsDefault=IsDefault) # create physical object as described in task file_name = self._convert_endpoint_name_to_endpoint_file_name(DBProxyEndpointName) @@ -107,7 +108,7 @@ def describe(self, Filters) - def modify(self, DBProxyEndpointName:str, NewDBProxyEndpointName:Optional[str] = None, + def modify(self, DBProxyEndpointName:str, NewDBProxyEndpointName:Optional[str] = None, VpcSubnetIds:Optional[List[str]] = None, TargetRole:Optional[str] = None, Tags:Optional[str] = None, Status:Optional[str] = None): '''Modify an existing DBProxy endpoint.''' @@ -123,7 +124,7 @@ def modify(self, DBProxyEndpointName:str, NewDBProxyEndpointName:Optional[str] = if endpoint_state != 'available': raise InvalidDBProxyEndpointStateException(DBProxyEndpointName, endpoint_state) - updates = {key: val for key in ['TargetRole', 'Tags', 'Status'] for val in [TargetRole, Tags, Status] if val is not None} + updates = {key: val for key in ['VpcSubnetIds','TargetRole', 'Tags', 'Status'] for val in [VpcSubnetIds, TargetRole, Tags, Status] if val is not None} # If need to change the name: if NewDBProxyEndpointName: diff --git a/DB/NEW_KT_DB/Test/GeneralTests.py b/DB/NEW_KT_DB/Test/GeneralTests.py index eefc66ce..236a6127 100644 --- a/DB/NEW_KT_DB/Test/GeneralTests.py +++ b/DB/NEW_KT_DB/Test/GeneralTests.py @@ -13,6 +13,9 @@ def storage_manager(): def assert_file_exist(storage_manager, file_name): assert storage_manager.is_file_exist(file_name), f"Expected file {file_name} was not created." +def is_file_exist(storage_manager, file_name): + return storage_manager.is_file_exist(file_name) + # Generic function to load JSON file and assert its content def assert_json_content(storage_manager, file_name, expected_data): '''Validates the content of a JSON file stored in the storage manager's base directory. diff --git a/DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py b/DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py index dbbc050c..e5fed85f 100644 --- a/DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py +++ b/DB/NEW_KT_DB/Test/test_DBProxyEndpointTests.py @@ -1,6 +1,7 @@ from typing import Dict, Literal, Optional from unittest.mock import MagicMock import pytest +import os from DB.NEW_KT_DB.DataAccess.ObjectManager import ObjectManager from DB.NEW_KT_DB.DataAccess.DBManager import DBManager from DB.NEW_KT_DB.DataAccess.DBProxyEndpointManager import DBProxyEndpointManager @@ -20,9 +21,12 @@ def storage_manager() -> StorageManager: @pytest.fixture -def object_manager() -> ObjectManager: - object_manager:ObjectManager = ObjectManager(':memory:') - return object_manager +def object_manager(): + object_manager:ObjectManager = ObjectManager('test.db') + yield object_manager + if os.path.exists('test.db'): + os.remove('test.db') + @pytest.fixture @@ -55,11 +59,12 @@ def setup_db_proxy_endpoint(endpoint_controller: DBProxyEndpointController): """Create a db proxy endpoint for tests and delete it after tests""" db_proxy_name = "my-proxy" endpoint_name = "my-endpoint" + vpc_subnet_ids = ['subnet-12345678', 'subnet-87654321'] target_role = 'READ_WRITE' - endpoint_description = endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name, target_role) + endpoint_description = endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name,vpc_subnet_ids, TargetRole=target_role) - yield db_proxy_name, endpoint_name, target_role, endpoint_description + yield db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint_description # Perform cleanup after the test has completed try: @@ -92,11 +97,12 @@ def _test_exists_in_db(endpoint_manager:DBProxyEndpointManager, endpoint_name): return endpoint_manager.is_exists(endpoint_name) -def _test_description_is_correct(description, db_proxy_name, endpoint_name, target_role): +def _test_description_is_correct(description, db_proxy_name, endpoint_name, vpc_subnet_ids, target_role): """Checking if function response describe dbProxyEndpoint object correctly""" endpoint_data = description[DBProxyEndpoint.object_name][0] assert endpoint_data['DBProxyName'] == db_proxy_name assert endpoint_data['DBProxyEndpointName'] == endpoint_name + assert endpoint_data['VpcSubnetIds'] == vpc_subnet_ids assert endpoint_data['TargetRole'] == target_role @@ -104,10 +110,10 @@ def test_create_db_proxy_endpoint(endpoint_service:DBProxyEndpointService, setup endpoint_manager:DBProxyEndpointManager): # Create and get response - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint # Check if response is correct - _test_description_is_correct(endpoint, db_proxy_name, endpoint_name, target_role) + _test_description_is_correct(endpoint, db_proxy_name, endpoint_name, vpc_subnet_ids, target_role) # Check if phisical file exists endpoint_file_name = endpoint_service._convert_endpoint_name_to_endpoint_file_name(endpoint_name) @@ -118,50 +124,51 @@ def test_create_db_proxy_endpoint(endpoint_service:DBProxyEndpointService, setup def test_create_with_non_valid_parameters(endpoint_controller:DBProxyEndpointController) : - valid_params = ["my-proxy","my-endpoint",'READ_WRITE',[{ 'Key': 'string','Value': 'string'}]] - non_valid_params = ["my_proxy","my_en455", "ff",{'Key': 1,'Value': 'string'}] + valid_params = ["my-proxy","my-endpoint",['subnet-12345678', 'subnet-87654321'],'READ_WRITE',[{ 'Key': 'string','Value': 'string'}]] + non_valid_params = ["my_proxy","my_en455", ['subnet-12345678', 'subnet-87654321'], "ff",{'Key': 1,'Value': 'string'}] with pytest.raises(InvalidParamException): - endpoint_controller.create_db_proxy_endpoint(valid_params[0], non_valid_params[1], valid_params[2], valid_params[3]) + endpoint_controller.create_db_proxy_endpoint(valid_params[0], non_valid_params[1], valid_params[2],valid_params[3], valid_params[4]) with pytest.raises(InvalidParamException): - endpoint_controller.create_db_proxy_endpoint(valid_params[0], valid_params[1], non_valid_params[2], valid_params[3]) + endpoint_controller.create_db_proxy_endpoint(valid_params[0], valid_params[1], valid_params[2], non_valid_params[3], valid_params[4]) with pytest.raises(InvalidParamException): - endpoint_controller.create_db_proxy_endpoint(non_valid_params[0], valid_params[1], valid_params[2], valid_params[3]) + endpoint_controller.create_db_proxy_endpoint(non_valid_params[0], valid_params[1], valid_params[2],valid_params[3], valid_params[4]) with pytest.raises(InvalidParamException): - endpoint_controller.create_db_proxy_endpoint(valid_params[0], valid_params[1], valid_params[2], non_valid_params[3]) + endpoint_controller.create_db_proxy_endpoint(valid_params[0], valid_params[1], valid_params[2],valid_params[3], non_valid_params[4]) def test_create_when_db_proxy_not_exist(endpoint_controller:DBProxyEndpointController): db_proxy_name = "not-exist-proxy" endpoint_name = "my-endpoint" + vpc_subnet_ids = ['subnet-12345678', 'subnet-87654321'] target_role = 'READ_WRITE' # Create with pytest.raises(DBProxyNotFoundException): - endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name, target_role) + endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name, vpc_subnet_ids, target_role) def test_create_db_proxy_endpoint_with_existing_name(setup_db_proxy_endpoint: tuple[Literal['my-proxy'], Literal['my-endpoint'], Literal['READ_WRITE'], dict[str, list[dict]]],endpoint_controller:DBProxyEndpointController): # Create first - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint # Create sec with pytest.raises(DBProxyEndpointAlreadyExistsException): - endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name, target_role) + endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name, vpc_subnet_ids, target_role) def test_delete_db_proxy_endpoint(setup_db_proxy_endpoint: tuple[Literal['my-proxy'], Literal['my-endpoint'], Literal['READ_WRITE'], dict[str, list[dict]]], endpoint_controller:DBProxyEndpointController, storage_manager:StorageManager, endpoint_manager:DBProxyEndpointManager, endpoint_service:DBProxyEndpointService): # Create and get response - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint # Delete endpoint_description = endpoint_controller.delete_db_proxy_endpoint(endpoint_name) # Check if response is correct - _test_description_is_correct(endpoint_description, db_proxy_name, endpoint_name, target_role) + _test_description_is_correct(endpoint_description, db_proxy_name, endpoint_name, vpc_subnet_ids, target_role) # Check if phisical file not exists endpoint_file_name = endpoint_service._convert_endpoint_name_to_endpoint_file_name(endpoint_name) @@ -178,12 +185,9 @@ def test_delete_non_exist_db_proxy_endpoint(endpoint_controller:DBProxyEndpointC with pytest.raises(DBProxyEndpointNotFoundException): endpoint_controller.delete_db_proxy_endpoint(endpoint_name) - -def test_delete_non_valid_state_db_proxy_endpoint(endpoint_controller:DBProxyEndpointController, endpoint_service:DBProxyEndpointService): - db_proxy_name = "my-proxy" - endpoint_name = "my-endpoint" - target_role = 'READ_WRITE' - endpoint = endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name, target_role) +def test_delete_non_valid_state_db_proxy_endpoint(endpoint_controller:DBProxyEndpointController, endpoint_service:DBProxyEndpointService, + setup_db_proxy_endpoint): + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint state = 'not available' endpoint_service.modify(endpoint_name, Status= state) with pytest.raises(InvalidDBProxyEndpointStateException): @@ -197,14 +201,14 @@ def test_modify_name_to_db_proxy_endpoint(setup_db_proxy_endpoint: tuple[Literal cleanup_endpoint:Optional[str]): # Create and get response - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint # Modify new_name = "your-endpoint" endpoint = endpoint_controller.modify_db_proxy_endpoint(endpoint_name, new_name) # Check if response is correct - _test_description_is_correct(endpoint, db_proxy_name, new_name, target_role) + _test_description_is_correct(endpoint, db_proxy_name, new_name, vpc_subnet_ids, target_role) # Check if phisical file is up-to-date old_endpoint_file_name = endpoint_service._convert_endpoint_name_to_endpoint_file_name(endpoint_name) @@ -229,12 +233,12 @@ def test_modify_non_exist_db_proxy_endpoint(endpoint_controller:DBProxyEndpointC def test_describe_db_proxy_endpoint(setup_db_proxy_endpoint: tuple[Literal['my-proxy'], Literal['my-endpoint'], Literal['READ_WRITE'], dict[str, list[dict]]],endpoint_controller:DBProxyEndpointController): # Create and get response - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint # Describe description = endpoint_controller.describe_db_proxy_endpoint(endpoint_name) - _test_description_is_correct(description, db_proxy_name, endpoint_name, target_role) + _test_description_is_correct(description, db_proxy_name, endpoint_name, vpc_subnet_ids, target_role) def test_describe_non_exist_db_proxy_endpoint(endpoint_controller:DBProxyEndpointController): @@ -247,24 +251,25 @@ def test_describe_with_filters(setup_db_proxy_endpoint: tuple[Literal['my-proxy' cleanup_endpoint:Optional[str]): # Create and get response first endpoint - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint # Create sec endpoint_name2 = "my-endpoint2" + vpc_subnet_ids2 = ['subnet-12345678'] target_role2 = 'READ_ONLY' - endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name2, target_role2) + endpoint_controller.create_db_proxy_endpoint(db_proxy_name, endpoint_name2, vpc_subnet_ids2, TargetRole=target_role2) # Describe filters = [ { - 'Name': 'target_role', + 'Name': 'TargetRole', 'Values': [ 'READ_WRITE' ] } ] - description = endpoint_controller.describe_db_proxy_endpoint(DBProxyEndpointName=endpoint_name, Filters=filters) + description = endpoint_controller.describe_db_proxy_endpoint(Filters=filters) # Assert endpoint_data = description[DBProxyEndpoint.object_name] @@ -277,17 +282,17 @@ def test_describe_with_filters(setup_db_proxy_endpoint: tuple[Literal['my-proxy' def test_describe_with_filters_does_nothing(setup_db_proxy_endpoint: tuple[Literal['my-proxy'], Literal['my-endpoint'], Literal['READ_WRITE'], dict[str, list[dict]]], endpoint_controller:DBProxyEndpointController): # Create and get response - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint # Describe filters = [{'Name': 'yyy','Values':['hh','jj']}, - {'Name': 'TargetRole','Values':['hh','jj']}] + {'Name': 'TargetRol','Values':['hh','jj']}] description = endpoint_controller.describe_db_proxy_endpoint(DBProxyEndpointName=endpoint_name, Filters=filters) - _test_description_is_correct(description, db_proxy_name, endpoint_name, target_role) + _test_description_is_correct(description, db_proxy_name, endpoint_name, vpc_subnet_ids, target_role) def test_describe_with_non_correct_filters(setup_db_proxy_endpoint: tuple[Literal['my-proxy'], Literal['my-endpoint'], Literal['READ_WRITE'], dict[str, list[dict]]],endpoint_controller:DBProxyEndpointController): # Create and get response - db_proxy_name, endpoint_name, target_role, endpoint = setup_db_proxy_endpoint + db_proxy_name, endpoint_name, vpc_subnet_ids, target_role, endpoint = setup_db_proxy_endpoint def test_filters(filters): nonlocal endpoint_controller