[feature/BLAZ-2446] to dev

slp_abacus/tests/integration/test_abacus_processor.py

@@ -0,0 +1,75 @@
+import io
+
+import pytest
+from pytest import mark, param
+from starlette.datastructures import UploadFile
+
+from sl_util.sl_util.file_utils import get_byte_data
+from sl_util.tests.util.file_utils import generate_temporary_file
+from slp_abacus import AbacusProcessor
+from slp_abacus.slp_abacus.validate.abacus_validator import MAX_SIZE, MIN_SIZE
+from slp_abacus.tests.resources import test_resource_paths
+from slp_base import LoadingDiagramFileError
+from slp_base.slp_base.errors import ErrorCode, DiagramFileNotValidError
+
+SAMPLE_ID = 'id'
+SAMPLE_NAME = 'name'
+SAMPLE_FILE_INVALID_FORMAT = test_resource_paths.abacus_bad_file_format
+SAMPLE_FILE_INVALID_EXTENSION = test_resource_paths.abacus_bad_file_extension
+SAMPLE_DEFAULT_MAPPING = test_resource_paths.abacus_default_mapping
+
+class TestAbacusProcessor:
+    def test_no_abacus_file_format(self):
+        # GIVEN a valid ABACUS file
+        abacus_file = open(SAMPLE_FILE_INVALID_FORMAT, 'rb')
+
+        # AND a valid default mapping file
+        default_mapping_file = get_byte_data(SAMPLE_DEFAULT_MAPPING)
+
+        # WHEN ABACUS is processing
+        # THEN raises LoadingDiagramFileError
+        with pytest.raises(LoadingDiagramFileError) as error:
+            AbacusProcessor(SAMPLE_ID, SAMPLE_NAME, abacus_file, [default_mapping_file]).process()
+
+        # AND the error details are correct
+        assert ErrorCode.DIAGRAM_LOADING_ERROR == error.value.error_code
+        assert "Source file cannot be loaded" == error.value.title
+        assert "JSONDecodeError" == error.value.detail
+        assert "Expecting value: line 1 column 1 (char 0)" == error.value.message
+
+    @mark.parametrize('source', [
+        param(generate_temporary_file(MIN_SIZE - 1, "test_min_size.txt"), id='file too small'),
+        param(generate_temporary_file(MAX_SIZE + 1, "test_max_size.txt"), id='file too big')])
+    def test_abacus_file_sizes(self, source):
+        # GIVEN a valid default mapping file
+        default_mapping_file = get_byte_data(SAMPLE_DEFAULT_MAPPING)
+
+        # WHEN ABACUS is processing
+        # THEN raises DiagramFileNotValidError
+        with pytest.raises(DiagramFileNotValidError) as error:
+            upload_file = UploadFile(filename="source.json", file=io.BytesIO(source))
+            AbacusProcessor(SAMPLE_ID, SAMPLE_NAME, upload_file, [default_mapping_file]).process()
+
+        # AND the error details are correct
+        assert ErrorCode.DIAGRAM_NOT_VALID == error.value.error_code
+        assert "Abacus file is not valid" == error.value.title
+        assert "Provided diag_file is not valid. Invalid size" == error.value.detail
+        assert "Provided diag_file is not valid. Invalid size" == error.value.message
+
+    def test_no_abacus_valid_extension(self):
+        # GIVEN a valid ABACUS file
+        abacus_file = open(SAMPLE_FILE_INVALID_EXTENSION, 'rb')
+
+        # AND a valid default mapping file
+        default_mapping_file = get_byte_data(SAMPLE_DEFAULT_MAPPING)
+
+        # WHEN ABACUS is processing
+        # THEN raises DiagramFileNotValidError
+        with pytest.raises(DiagramFileNotValidError) as error:
+            AbacusProcessor(SAMPLE_ID, SAMPLE_NAME, abacus_file, [default_mapping_file]).process()
+
+        # AND the error details are correct
+        assert ErrorCode.DIAGRAM_NOT_VALID == error.value.error_code
+        assert "Abacus file is not valid" == error.value.title
+        assert "Invalid content type for diag_file" == error.value.detail
+        assert "Invalid content type for diag_file" == error.value.message

slp_abacus/tests/resources/abacus/aws_minimal.drawio

@@ -0,0 +1 @@
+<mxfile host="Electron" modified="2023-10-10T12:38:05.500Z" agent="5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) draw.io/19.0.3 Chrome/102.0.5005.63 Electron/19.0.3 Safari/537.36" etag="1qYMq70-eEWiWxYW-9pH" version="19.0.3" type="device"><diagram id="pOeIjdqQ_qid1hbLa7JX" name="Page-1">7VbbcpswEP0aP8bDNTiPMdhJ22TaxnGSN48MMqgRiBHCxv36rkBgbmmSmXSmDxljrD0Su9LZ3WMmphsXVxyl0S0LMJ0YWlBMTG9iGLpl6PAjkWOFXGhWBYScBGrRCViR31iBmkJzEuCss1AwRgVJu6DPkgT7ooMhztmhu2zHaDdqikI8AFY+okP0kQQiqtCZ4Zzwa0zCqI6sn19UMzGqF6uTZBEK2KEFmYuJ6XLGRDWKCxdTSV7NS/Xc8oXZZmMcJ+ItD9jEeVjPsuKeRhuP/VqHj0/emfKyRzRXB758XAHgUpYHat/iWJORMpKIklB7DhfEc7WJDTOutKaG3QP6ttMF9KElfXSBvu10Ab3vXu/F1/sbbAEDq+Ne68XXWhuEy5yzXFCSYLcpPQ3AkKOAQEpcRhkHLGEJsDePREzB0mF4iIjAqxT5ktUDtA1gO5YIVfy6UduKeOkVylsgiMWVjzITmC/2uEpItYZSlGZk2zzFsZ/zjOzxHc4q5xKFQkzlOC5C2bNTdMisachZnpbb/wKxRmc3MNz4sjA2iArpSHD2jOuDTgwTPktZfPMdobRHwB5zQaCvLikJpX/BZDikLIp3pUdghSThTWl5pqaYGAsRoCzCgTrSsBfqwoaouGhBqjeuMIux4EdYomYbxVFCZZjKPpzaXq+xqN3yNYiU1ISN71M3wkA15HhzFuvrrw/2z6fn4/333Ea338748swYNOegI0lc6le7vBTkkTiEqJRs4e5TAunjQg5ZnOZQOhmMPSTQFmV4oxuzAr7TNAmHVP5VON7Or92h1xqyOxshd/YB3I7uf8jtHQ4JVP6n5n1qXkvzeFUVI2KnW85ifvlPxa4J0Yid/jFiZ1n/ndiZr4vdaO11/gdq3m/QFtMfLCOiTJ63ZUKw+NXE+FgWWLdyRyrHnGLf6CUesrW0Z7ZpvdwOb1RW473JNM5LFWin0xlk0zmf2sN0XpjvziaYp3fWcq715m8u/gA=</diagram></mxfile>

slp_abacus/tests/resources/test_resource_paths.py

@@ -10,3 +10,6 @@
 wrong_mxgraphmodel_abacus = f'{abacus}/wrong_mxgraphmodel.abacus'
 wrong_root_abacus = f'{abacus}/wrong_root.abacus'
 not_xml = f'{abacus}/not_xml.abacus'
+abacus_bad_file_format = f'{abacus}/aws_minimal.drawio'
+abacus_bad_file_extension = f'{abacus}/invalid-file-extension.png'
+abacus_default_mapping = f'{path}/mapping/iriusrisk-abacus-mapping.yaml'

slp_cft/tests/integration/test_cft_processor.py

@@ -36,7 +36,6 @@
 OTM_EXPECTED_RESULT = test_resource_paths.otm_expected_result
 ALTSOURCE_COMPONENTS_OTM_EXPECTED = test_resource_paths.altsource_components_otm_expected
 
-
 class TestCloudformationProcessor:
     def test_altsource_components(self):
         # GIVEN a valid CFT file with altsource resources
@@ -759,7 +758,6 @@ def test_invalid_resources_mapping_file(self):
     @pytest.mark.parametrize('cft_file_size', [FILE_MAX_SIZE + 1, FILE_MIN_SIZE - 1])
     def test_min_max_cloudformation_file_sizes(self, cft_file_size):
         # GIVEN a max file size limit and a valid CFT file
-        max_file_size_allowed_in_bytes = 1024 * 1024
         cloudformation_file = generate_temporary_file(cft_file_size, "test_max_size.txt")
 
         # AND a valid CFT mapping file
@@ -781,7 +779,7 @@ def test_min_max_mapping_file_sizes(self, mapping_file_size):
         # GIVEN a valid CFT file with altsource resources
         cloudformation_file = get_byte_data(SAMPLE_VALID_CFT_FILE)
 
-        # AND a invalid size CFT mapping file
+        # AND an invalid size CFT mapping file
         mapping_file = generate_temporary_file(mapping_file_size, "test_mapping_sizes.txt")
 
         # WHEN the CFT file is processed

tests/integration/api/controllers/diagram/abacus/test_otm_controller_diagram_abacus.py

@@ -0,0 +1,98 @@
+import json
+
+import responses
+from fastapi.testclient import TestClient
+
+from sl_util.sl_util.file_utils import get_byte_data
+from slp_base.slp_base.otm_validator import OTMValidator
+from slp_base.slp_base.provider_type import application_json
+from slp_base.slp_base.schema import Schema
+from startleft.startleft.api import fastapi_server
+from tests.integration.api.controllers.diagram.test_otm_controller_diagram import get_url
+from tests.resources import test_resource_paths
+
+webapp = fastapi_server.webapp
+client = TestClient(webapp)
+
+SAMPLE_ID = 'id'
+SAMPLE_NAME = 'name'
+SAMPLE_VALID_FILE = test_resource_paths.abacus_example
+SAMPLE_DEFAULT_MAPPING = test_resource_paths.abacus_default_mapping
+SAMPLE_CUSTOM_MAPPING = test_resource_paths.abacus_custom_mapping
+
+OTM_SCHEMA_FILENAME = OTMValidator.schema_filename
+
+
+class TestOTMControllerDiagramAbacus:
+    @responses.activate
+    def test_create_otm_ok_default_mapping(self):
+        # GIVEN a source file
+        diag_file = get_byte_data(SAMPLE_VALID_FILE)
+
+        # AND a valid default mapping file
+        default_mapping_file = get_byte_data(SAMPLE_DEFAULT_MAPPING)
+
+        # WHEN processing
+        file = {'diag_file': (SAMPLE_VALID_FILE, diag_file),
+                'default_mapping_file': (SAMPLE_DEFAULT_MAPPING, default_mapping_file)}
+        body = {'diag_type': 'ABACUS', 'id': SAMPLE_ID, 'name': SAMPLE_NAME}
+        response = client.post(get_url(), files=file, data=body)
+
+        # THEN the OTM is returned inside the response as valid JSON
+        assert response.status_code == 201
+        assert response.headers.get('content-type') == application_json
+        otm = json.loads(response.text)
+
+        # AND the OTM is valid according to the OTM schema
+        schema: Schema = Schema.from_package('otm', OTM_SCHEMA_FILENAME)
+        schema.validate(otm)
+        assert schema.valid, f"OTM Schema is not valid: {schema.errors}"
+
+        # AND the OTM contains expected data
+        assert len(otm['representations']) == 1
+        assert len(otm['trustZones']) == 1
+        assert len(otm['dataflows']) == 0
+        components = otm['components']
+        assert len(components) == 8
+        assert components[0]['type'] == 'compact-server-side-web-application'
+        assert components[1]['type'] == 'CD-MSG-BROKER'
+        assert components[2]['type'] == 'back-end-server'
+        assert components[3]['type'] == 'other-database'
+        assert components[4]['type'] == 'CD-CONTENT-DELIVERY-NETWORK'
+        assert components[5]['type'] == 'compact-server-side-web-application'
+        assert components[6]['type'] == 'compact-server-side-web-application'
+        assert components[7]['type'] == 'web-client'
+
+    def test_create_otm_ok_custom_mapping(self):
+        # GIVEN a source file
+        diag_file = get_byte_data(SAMPLE_VALID_FILE)
+
+        # AND a valid default mapping file
+        default_mapping_file = get_byte_data(SAMPLE_DEFAULT_MAPPING)
+
+        # AND a valid custom mapping file
+        custom_mapping_file = get_byte_data(SAMPLE_CUSTOM_MAPPING)
+
+        # WHEN processing
+        file = {'diag_file': (SAMPLE_VALID_FILE, diag_file),
+                'default_mapping_file': (SAMPLE_DEFAULT_MAPPING, default_mapping_file),
+                'custom_mapping_file': (SAMPLE_CUSTOM_MAPPING, custom_mapping_file)}
+        body = {'diag_type': 'ABACUS', 'id': SAMPLE_ID, 'name': SAMPLE_NAME}
+        response = client.post(get_url(), files=file, data=body)
+
+        # THEN the OTM is returned inside the response as valid JSON
+        assert response.status_code == 201
+        assert response.headers.get('content-type') == application_json
+        otm = json.loads(response.text)
+        assert len(otm['representations']) == 1
+        assert len(otm['trustZones']) == 2
+        assert len(otm['dataflows']) == 0
+        components = otm['components']
+        assert len(components) == 7
+        assert components[0]['type'] == 'CD-V2-SST-POC-WEBPAGE'
+        assert components[1]['type'] == 'CD-V2-SST-POC-INTEGRATOR'
+        assert components[2]['type'] == 'back-end-server'
+        assert components[3]['type'] == 'other-database'
+        assert components[4]['type'] == 'CD-V2-SST-POC-WEBPAGE'
+        assert components[5]['type'] == 'CD-V2-SST-POC-WEBPAGE'
+        assert components[6]['type'] == 'web-client'

tests/integration/api/controllers/diagram/drawio/test_otm_controller_diagram_drawio.py

@@ -3,6 +3,8 @@
 import pytest
 import responses
 from fastapi.testclient import TestClient
+from slp_base.slp_base.provider_type import application_json
+from tests.integration.api.controllers.diagram.test_otm_controller_diagram import get_url
 
 from tests.resources import test_resource_paths
 
@@ -13,16 +15,8 @@
     terraform_aws_simple_components, invalid_extension_mtmt_file
 
 webapp = fastapi_server.webapp
-
 client = TestClient(webapp)
 
-json_mime = 'application/json'
-
-
-def get_url():
-    return diag_create_otm_controller.PREFIX + diag_create_otm_controller.URL
-
-
 class TestOTMControllerDiagramDrawio:
 
     @responses.activate
@@ -44,7 +38,7 @@ def test_create_otm_multi_page_error(self):
 
         # Then the error is returned inside the response as JSON
         assert response.status_code == 400
-        assert response.headers.get('content-type') == json_mime
+        assert response.headers.get('content-type') == application_json
         body_response = json.loads(response.text)
         assert body_response['status'] == '400'
         assert body_response['error_type'] == 'LoadingDiagramFileError'
@@ -78,7 +72,7 @@ def test_create_otm_ok(self, diagram_file_path):
 
         # Then
         assert response.status_code == 201
-        assert response.headers.get('content-type') == json_mime
+        assert response.headers.get('content-type') == application_json
         otm = json.loads(response.text)
         assert len(otm['trustZones']) > 0
         assert len(otm['components']) > 0
@@ -107,7 +101,7 @@ def test_custom_mapping_file_override_mapping_file(self, custom_mapping_file_pat
 
         # Then the OTM is returned inside the response as JSON
         assert response.status_code == diag_create_otm_controller.RESPONSE_STATUS_CODE
-        assert response.headers.get('content-type') == json_mime
+        assert response.headers.get('content-type') == application_json
 
         otm = json.loads(response.text)
         assert otm['otmVersion'] == '0.2.0'
@@ -134,7 +128,7 @@ def test_diagram_file_invalid_extensions(self, filepath):
 
         # AND the error details are correct
         assert response.status_code == 400
-        assert response.headers.get('content-type') == json_mime
+        assert response.headers.get('content-type') == application_json
 
         body_response = json.loads(response.text)
         assert body_response['status'] == '400'