Integration

DB/NEW_KT_DB/Controller/DBProxyEndpointController.py

@@ -7,19 +7,22 @@ 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):
         """Delete a db proxy endpoint"""
         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, 

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}
 
 

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)
+
+
+

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,

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:
 

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.