Merge pull request #19 from PSNAppz/develop

Added Basic Test Case for Sync Link API

tests/base_setup.py

@@ -4,4 +4,4 @@
 
 
 def base_setup():
-    _config.db_dbname = "sparuidb"
+    _config.db_dbname = "openg2p_spar_test_db"

tests/test_sync_mapper_controller.py

@@ -1,150 +1,147 @@
-import asyncio
-import unittest
 from datetime import datetime
-from unittest.mock import MagicMock, patch
-
-from fastapi import Request
-from openg2p_g2pconnect_common_lib.common.schemas.status_codes import StatusEnum
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from openg2p_g2pconnect_common_lib.common.schemas import (
+    RequestHeader,
+    StatusEnum,
+    SyncResponseHeader,
+    SyncResponseStatusReasonCodeEnum,
+)
 from openg2p_g2pconnect_common_lib.mapper.schemas import (
+    LinkRequest,
+    LinkRequestMessage,
+    LinkResponse,
+    LinkResponseMessage,
+    SingleLinkRequest,
     SingleLinkResponse,
-    SingleResolveResponse,
-    SingleUnlinkResponse,
-    SingleUpdateResponse,
 )
 from openg2p_spar_mapper_api.controllers.sync_mapper_controller import (
     SyncMapperController,
 )
+from openg2p_spar_mapper_api.services import (
+    RequestValidation,
+    RequestValidationException,
+)
 
 
-class TestSyncMapperController(unittest.TestCase):
-    def setUp(self) -> None:
-        self.mock_request = MagicMock(spec=Request)
-        self.mock_request_validation = patch(
-            "openg2p_spar_mapper_api.controllers.sync_mapper_controller.RequestValidation"
-        ).start()
-        self.mock_response_helper = patch(
-            "openg2p_spar_mapper_api.controllers.sync_mapper_controller.SyncResponseHelper"
-        ).start()
+@pytest.fixture(autouse=True)
+def setup_controller():
+    controller = SyncMapperController()
+    controller.mapper_service = AsyncMock()
 
-        self.mock_link_response = SingleLinkResponse(
-            reference_id="string",
-            timestamp=datetime.now(),
-            status=StatusEnum.succ,
-            additional_info=[{}],
-            locale="eng",
-            id="string",
-            fa="string",
-            name="string",
-            phone_number="string",
-        )
-        self.mock_update_response = SingleUpdateResponse(
-            reference_id="string",
-            timestamp=datetime.now(),
-            status=StatusEnum.succ,
-            additional_info=[{}],
-            locale="eng",
-            id="string",
-            fa="string",
-            name="string",
-            phone_number="string",
-        )
-        self.mock_resolve_response = SingleResolveResponse(
-            reference_id="string",
-            timestamp=datetime.now(),
-            status=StatusEnum.succ,
-            additional_info=[{}],
-            locale="eng",
-            id="string",
-            fa="string",
-            name="string",
-            phone_number="string",
-        )
-        self.mock_unlink_response = SingleUnlinkResponse(
-            reference_id="string",
-            timestamp=datetime.now(),
+    request_validation_mock = MagicMock()
+    request_validation_mock.validate_request = MagicMock(return_value=True)
+    request_validation_mock.validate_link_request_header = MagicMock(return_value=True)
+
+    mock_link_response = LinkResponse(
+        header=SyncResponseHeader(
+            message_id="test_message_id",
+            message_ts=datetime.now().isoformat(),
+            action="link",
             status=StatusEnum.succ,
-            additional_info=[{}],
-            locale="eng",
-            id="string",
-            fa="string",
-            name="string",
-            phone_number="string",
+            status_reason_code=None,
+            status_reason_message="Success",
+        ),
+        message=LinkResponseMessage(
+            transaction_id="trans_id",
+            link_response=[
+                SingleLinkResponse(
+                    reference_id="test_ref",
+                    timestamp=datetime.now(),
+                    status=StatusEnum.succ,
+                    additional_info=[{}],
+                    fa="test_fa",
+                    status_reason_code=None,
+                    status_reason_message="Test message",
+                    locale="en",
+                )
+            ],
+        ),
+    )
+
+    response_helper_mock = MagicMock()
+    response_helper_mock.construct_success_sync_link_response.return_value = (
+        mock_link_response
+    )
+
+    mock_error_response = LinkResponse(
+        header=SyncResponseHeader(
+            message_id="error_message_id",
+            message_ts=datetime.now().isoformat(),
+            action="error_action",
+            status=StatusEnum.rjct,
+            status_reason_code=SyncResponseStatusReasonCodeEnum.rjct_action_not_supported.value,
+            status_reason_message="Validation error",
+        ),
+        message=LinkResponseMessage(
+            transaction_id="error_trans_id",
+            link_response=[],
+        ),
+    )
+
+    # Mock SyncResponseHelper for error scenario
+    response_helper_mock.construct_error_sync_response.return_value = (
+        mock_error_response
+    )
+
+    with patch(
+        "openg2p_spar_mapper_api.services.RequestValidation.get_component",
+        return_value=request_validation_mock,
+    ), patch(
+        "openg2p_spar_mapper_api.services.SyncResponseHelper.get_component",
+        return_value=response_helper_mock,
+    ):
+        mock_link_request = LinkRequest(
+            header=RequestHeader(
+                message_id="test_message_id",
+                message_ts=datetime.now().isoformat(),
+                action="test_action",
+                sender_id="test_sender",
+                total_count=1,
+            ),
+            message=LinkRequestMessage(
+                transaction_id="test_transaction_id",
+                link_request=[
+                    SingleLinkRequest(
+                        reference_id="test_ref",
+                        timestamp=datetime.now(),
+                        id="test_id",
+                        fa="test_fa",
+                    )
+                ],
+            ),
         )
-
-        self.mock_mapper_service = MagicMock()
-        self.controller = SyncMapperController()
-        self.controller.mapper_service = self.mock_mapper_service
-        self.controller.mapper_service.link.return_value = [self.mock_link_response]
-
-    def tearDown(self) -> None:
-        patch.stopall()
-
-    def test_link_sync(self):
-        print("Test case test_link_sync is being executed")
-
-        async def run_test():
-            self.mock_link_response.reference_id = "12334324"
-            self.mock_link_response.timestamp = datetime(2024, 4, 4)
-            self.mock_request_validation.validate_request.side_effect = None
-            self.mock_request_validation.validate_link_request_header.side_effect = None
-
-            response = await self.controller.link_sync(self.mock_request)
-
-            self.mock_mapper_service.link.assert_called_once()
-            self.assertIsNotNone(response)
-            self.mock_response_helper.get_component().construct_error_sync_response.assert_not_called()
-            self.mock_response_helper.get_component().construct_success_sync_link_response.assert_called_once_with(
-                self.mock_request, [self.mock_link_response]
-            )
-            self.assertEqual(response.reference_id, "12334324")
-            self.assertEqual(response.timestamp, datetime(2024, 4, 4))
-
-            # self.assertEqual(response, self.mock_link_response)
-            # self.assertEqual(response.timestamp, self.mock_link_response.timestamp)
-
-        asyncio.run(run_test())
-
-    # def test_update_sync(self):
-    #     async def run_test():
-    #         self.mock_request_validation.validate_request.side_effect = None
-    #         self.mock_request_validation.validate_update_request_header.side_effect = None
-    #         response = await self.controller.update_sync(self.mock_request)
-    #         self.mock_mapper_service.update.assert_called_once()
-    #         self.assertIsNotNone(response)
-    #         self.mock_response_helper.get_component().construct_error_sync_response.assert_not_called()
-    #         self.mock_response_helper.get_component().construct_success_sync_update_response.assert_called_once_with(
-    #             self.mock_request, [self.mock_update_response]
-    #         )
-
-    #         asyncio.run(run_test())
-
-    # def test_resolve_sync(self):
-    #     async def run_test():
-    #         self.mock_request_validation.validate_request.side_effect = None
-    #         self.mock_request_validation.validate_resolve_request_header.side_effect = None
-    #         response = await self.controller.resolve_sync(self.mock_request)
-    #         self.mock_mapper_service.resolve.assert_called_once()
-    #         self.assertIsNotNone(response)
-    #         self.mock_response_helper.get_component().construct_error_sync_response.assert_not_called()
-    #         self.mock_response_helper.get_component().construct_success_sync_resolve_response.assert_called_once_with(
-    #             self.mock_request, [self.mock_resolve_response]
-    #         )
-    #         asyncio.run(run_test())
-
-    # def test_unlink_sync(self):
-    #     async def run_test():
-    #         self.mock_request_validation.validate_request.side_effect = None
-    #         self.mock_request_validation.validate_unlink_request_header.side_effect = None
-    #         response = await self.controller.unlink_sync(self.mock_request)
-    #         self.mock_mapper_service.unlink.assert_called_once()
-    #         self.assertIsNotNone(response)
-    #         self.mock_response_helper.get_component().construct_error_sync_response.assert_not_called()
-    #         self.mock_response_helper.get_component().construct_success_sync_unlink_response.assert_called_once_with(
-    #             self.mock_request, [self.mock_unlink_response]
-    #         )
-
-    #         asyncio.run(run_test())
-
-
-if __name__ == "__main__":
-    unittest.main()
+        yield controller, mock_link_request
+
+
+@pytest.mark.asyncio
+async def test_link_sync_success(setup_controller):
+    controller, mock_link_request = setup_controller
+    assert controller is not None
+    response = await controller.link_sync(mock_link_request)
+    assert response.header.status == StatusEnum.succ
+    assert response.message.transaction_id == "trans_id"
+    controller.mapper_service.link.assert_called_once_with(mock_link_request)
+
+
+@pytest.mark.asyncio
+async def test_link_sync_validation_error(setup_controller):
+    controller, mock_link_request = setup_controller
+    validation_error = RequestValidationException(
+        code=SyncResponseStatusReasonCodeEnum.rjct_action_not_supported,
+        message="Validation error",
+    )
+    with patch.object(
+        RequestValidation.get_component(),
+        "validate_request",
+        side_effect=validation_error,
+    ), patch.object(
+        RequestValidation.get_component(),
+        "validate_link_request_header",
+        side_effect=validation_error,
+    ):
+        response = await controller.link_sync(mock_link_request)
+        assert response.header.status == StatusEnum.rjct
+        assert validation_error.message in response.header.status_reason_message
+        controller.mapper_service.link.assert_not_called()

tox.ini

@@ -0,0 +1,20 @@
+[tox]
+env_list =
+  clean,
+  py310
+minversion = 4.11.3
+
+[testenv]
+description = run the tests with pytest
+commands = pytest --cov=openg2p_spar_mapper_api --cov=tests
+deps =
+  pytest-asyncio
+  pytest-cov
+  git+https://github.com/openg2p/openg2p-fastapi-common@develop\#subdirectory=openg2p-fastapi-common
+  git+https://github.com/openg2p/openg2p-fastapi-common@develop\#subdirectory=openg2p-fastapi-auth
+  git+https://github.com/openg2p/openg2p-g2pconnect-common-lib.git@develop#egg=openg2p-g2pconnect-common-lib
+  .
+[testenv:clean]
+deps = coverage
+skip_install = true
+commands = coverage erase