From 9bbe398795e2ad3097a03d94129c5f3bcfd78793 Mon Sep 17 00:00:00 2001 From: Niranjan Kumar Date: Fri, 8 Nov 2024 10:50:17 +0530 Subject: [PATCH 1/4] tests: add unit tests for controllers and services --- test-requirements.txt | 3 + tests/test_auth_controller.py | 312 +++++++++++++++++++++++++ tests/test_discovery_controller.py | 85 +++++++ tests/test_document_file_controller.py | 132 +++++++++++ tests/test_document_file_service.py | 297 +++++++++++++++++++++++ tests/test_exception.py | 33 +++ tests/test_form_controller.py | 179 ++++++++++++++ tests/test_form_service.py | 209 +++++++++++++++++ tests/test_membership_service.py | 88 +++++++ tests/test_oauth_controller.py | 155 ++++++++++++ tests/test_partner_service.py | 261 +++++++++++++++++++++ tests/test_program_controller.py | 174 ++++++++++++++ tests/test_program_service.py | 263 ++++++++++++++++++++- 13 files changed, 2188 insertions(+), 3 deletions(-) create mode 100644 tests/test_auth_controller.py create mode 100644 tests/test_discovery_controller.py create mode 100644 tests/test_document_file_controller.py create mode 100644 tests/test_document_file_service.py create mode 100644 tests/test_exception.py create mode 100644 tests/test_form_controller.py create mode 100644 tests/test_form_service.py create mode 100644 tests/test_membership_service.py create mode 100644 tests/test_oauth_controller.py create mode 100644 tests/test_partner_service.py create mode 100644 tests/test_program_controller.py diff --git a/test-requirements.txt b/test-requirements.txt index 4f53afa..212e4cd 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,3 +1,6 @@ pytest-cov +pytest-asyncio +pytest +pytest-mock 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 diff --git a/tests/test_auth_controller.py b/tests/test_auth_controller.py new file mode 100644 index 0000000..a47dbd5 --- /dev/null +++ b/tests/test_auth_controller.py @@ -0,0 +1,312 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from openg2p_fastapi_auth.models.orm.login_provider import LoginProvider +from openg2p_portal_api.controllers.auth_controller import AuthController +from openg2p_portal_api.models.orm.auth_oauth_provider import AuthOauthProviderORM +from openg2p_portal_api.models.orm.partner_orm import ( + BankORM, + PartnerBankORM, + PartnerORM, + PartnerPhoneNoORM, +) +from openg2p_portal_api.models.orm.reg_id_orm import RegIDORM, RegIDTypeORM +from openg2p_portal_api.models.profile import UpdateProfile +from sqlalchemy.exc import IntegrityError + +TEST_CONSTANTS = { + "PARTNER_ID": "test_partner_id", + "UNAUTHORIZED_MESSAGE": "Unauthorized", + "NATIONAL_ID": "National ID", + "PASSPORT": "Passport", + "BANK": "Test Bank", + "PHONE_NUMBER": "+1234567890", +} + + +class TestAuthController: + @pytest.fixture + def auth_controller(self): + with patch.object( + AuthController, "partner_service", new_callable=MagicMock + ) as mock_partner_service: + controller = AuthController() + controller.partner_service = mock_partner_service + yield controller + + @pytest.fixture + def mock_auth(self) -> MagicMock: + mock_auth = MagicMock() + mock_auth.partner_id = TEST_CONSTANTS["PARTNER_ID"] + return mock_auth + + @pytest.fixture + def mock_partner(self) -> MagicMock: + mock_partner = MagicMock() + mock_partner.id = 1 + mock_partner.email = "test@example.com" + mock_partner.gender = "M" + mock_partner.addl_name = "Test" + mock_partner.given_name = "John" + mock_partner.family_name = "Doe" + mock_partner.birthdate = "1990-01-01" + mock_partner.birth_place = "Test City" + return mock_partner + + @pytest.mark.asyncio + async def test_get_profile_success( + self, + auth_controller: AuthController, + mock_auth: MagicMock, + mock_partner: MagicMock, + ): + mock_reg_ids = [ + MagicMock( + id_type=MagicMock(name=TEST_CONSTANTS["NATIONAL_ID"]), + value="123456789", + expiry_date="2025-01-01", + ), + MagicMock( + id_type=MagicMock(name=TEST_CONSTANTS["PASSPORT"]), + value="AB123456", + expiry_date="2026-01-01", + ), + ] + + mock_bank_accounts = [ + MagicMock( + bank=MagicMock(name=TEST_CONSTANTS["BANK"]), + bank_id=1, + acc_number="1234567890", + account_holder_name="John Doe", + ) + ] + + mock_phone_numbers = [ + MagicMock( + phone_no=TEST_CONSTANTS["PHONE_NUMBER"], + is_primary=True, + date_collected="2023-01-01", + ) + ] + + PartnerORM.get_partner_data = AsyncMock(return_value=mock_partner) + RegIDORM.get_all_partner_ids = AsyncMock(return_value=mock_reg_ids) + PartnerBankORM.get_partner_banks = AsyncMock(return_value=mock_bank_accounts) + PartnerPhoneNoORM.get_partner_phone_details = AsyncMock( + return_value=mock_phone_numbers + ) + + mock_reg_id_type1 = MagicMock(spec=RegIDTypeORM) + mock_reg_id_type1.name = TEST_CONSTANTS["NATIONAL_ID"] + mock_reg_id_type2 = MagicMock(spec=RegIDTypeORM) + mock_reg_id_type2.name = TEST_CONSTANTS["PASSPORT"] + + RegIDTypeORM.get_id_type_name = AsyncMock( + side_effect=lambda x: mock_reg_id_type1 + if x == mock_reg_ids[0].id_type + else mock_reg_id_type2 + ) + + mock_bank = MagicMock() + mock_bank.name = TEST_CONSTANTS["BANK"] + BankORM.get_by_id = AsyncMock(return_value=mock_bank) + + profile = await auth_controller.get_profile(mock_auth) + + assert profile.id == 1, "Profile ID should match the mock partner ID" + assert ( + profile.email == "test@example.com" + ), "Email should match the mock partner email" + assert profile.gender == "M", "Gender should match the mock partner gender" + assert ( + profile.addl_name == "Test" + ), "Additional name should match the mock partner additional name" + assert ( + profile.given_name == "John" + ), "Given name should match the mock partner given name" + assert ( + profile.family_name == "Doe" + ), "Family name should match the mock partner family name" + assert ( + str(profile.birthdate) == "1990-01-01" + ), "Birthdate should match the mock partner birthdate" + assert ( + profile.birth_place == "Test City" + ), "Birth place should match the mock partner birth place" + + assert len(profile.ids) == 2, "Should have two IDs from mock data" + assert ( + profile.ids[0].id_type == TEST_CONSTANTS["NATIONAL_ID"] + ), "First ID type should match the mock data" + assert ( + profile.ids[0].value == "123456789" + ), "First ID value should match the mock data" + assert ( + profile.ids[1].id_type == TEST_CONSTANTS["PASSPORT"] + ), "Second ID type should match the mock data" + assert ( + profile.ids[1].value == "AB123456" + ), "Second ID value should match the mock data" + + assert len(profile.bank_ids) == 1, "Should have one bank account from mock data" + assert ( + profile.bank_ids[0].bank_name == TEST_CONSTANTS["BANK"] + ), "Bank name should match the mock data" + assert ( + profile.bank_ids[0].acc_number == "1234567890" + ), "Account number should match the mock data" + + assert ( + len(profile.phone_numbers) == 1 + ), "Should have one phone number from mock data" + assert ( + profile.phone_numbers[0].phone_no == TEST_CONSTANTS["PHONE_NUMBER"] + ), "Phone number should match the mock data" + + @pytest.mark.asyncio + async def test_get_profile_with_empty_related_data( + self, + auth_controller: AuthController, + mock_auth: MagicMock, + mock_partner: MagicMock, + ): + PartnerORM.get_partner_data = AsyncMock(return_value=mock_partner) + RegIDORM.get_all_partner_ids = AsyncMock(return_value=[]) + PartnerBankORM.get_partner_banks = AsyncMock(return_value=[]) + PartnerPhoneNoORM.get_partner_phone_details = AsyncMock(return_value=[]) + + profile = await auth_controller.get_profile(mock_auth) + + assert profile.id == 1, "Profile ID should match the mock partner ID" + assert ( + str(profile.birthdate) == "1990-01-01" + ), "Birthdate should match the mock partner birthdate" + assert len(profile.ids) == 0, "Should have no IDs as per mock data" + assert ( + len(profile.bank_ids) == 0 + ), "Should have no bank accounts as per mock data" + assert ( + len(profile.phone_numbers) == 0 + ), "Should have no phone numbers as per mock data" + + @pytest.mark.asyncio + async def test_update_profile( + self, auth_controller: AuthController, mock_auth: MagicMock + ): + update_data = UpdateProfile( + given_name="John", family_name="Doe", email="john@example.com" + ) + + auth_controller.partner_service.update_partner_info = AsyncMock() + + result = await auth_controller.update_profile(update_data, mock_auth) + + auth_controller.partner_service.update_partner_info.assert_called_once_with( + TEST_CONSTANTS["PARTNER_ID"], update_data.model_dump(exclude={"id"}) + ) + assert ( + result == "Updated the partner info" + ), "Update result should match expected message" + + @pytest.mark.asyncio + async def test_update_profile_integrity_error( + self, auth_controller: AuthController, mock_auth: MagicMock + ): + update_data = UpdateProfile(given_name="John") + + auth_controller.partner_service.update_partner_info = AsyncMock( + side_effect=IntegrityError(None, None, None) + ) + + result = await auth_controller.update_profile(update_data, mock_auth) + assert ( + result == "Could not add to registrant to program!!" + ), "Error message should match expected message" + + @pytest.mark.asyncio + async def test_get_profile_unauthorized(self): + controller = AuthController() + mock_auth = MagicMock() + mock_auth.partner_id = None + + with pytest.raises(Exception, match=TEST_CONSTANTS["UNAUTHORIZED_MESSAGE"]): + await controller.get_profile(mock_auth) + + @pytest.mark.asyncio + async def test_get_login_providers_db(self, auth_controller: AuthController): + mock_provider = MagicMock() + mock_provider.map_auth_provider_to_login_provider.return_value = LoginProvider( + id=1, name="Test Provider", type="oauth2" + ) + + AuthOauthProviderORM.get_all = AsyncMock(return_value=[mock_provider]) + + result = await auth_controller.get_login_providers_db() + + assert len(result) == 1, "Should return one provider from mock data" + assert isinstance( + result[0], LoginProvider + ), "Result should be a LoginProvider instance" + assert result[0].id == 1, "Provider ID should match the mock data" + assert ( + result[0].name == "Test Provider" + ), "Provider name should match the mock data" + + @pytest.mark.asyncio + async def test_get_login_provider_db_by_id(self, auth_controller: AuthController): + mock_provider = MagicMock() + mock_provider.map_auth_provider_to_login_provider.return_value = LoginProvider( + id=1, name="Test Provider", type="oauth2" + ) + + AuthOauthProviderORM.get_by_id = AsyncMock(return_value=mock_provider) + + result = await auth_controller.get_login_provider_db_by_id(1) + + assert isinstance( + result, LoginProvider + ), "Result should be a LoginProvider instance" + assert result.id == 1, "Provider ID should match the mock data" + assert ( + result.name == "Test Provider" + ), "Provider name should match the mock data" + + @pytest.mark.asyncio + async def test_get_login_provider_db_by_id_not_found( + self, auth_controller: AuthController + ): + AuthOauthProviderORM.get_by_id = AsyncMock(return_value=None) + + result = await auth_controller.get_login_provider_db_by_id(999) + assert result is None, "Result should be None for non-existent provider" + + @pytest.mark.asyncio + async def test_get_login_provider_db_by_iss(self, auth_controller: AuthController): + mock_provider = MagicMock() + mock_provider.map_auth_provider_to_login_provider.return_value = LoginProvider( + id=1, name="Test Provider", type="oauth2" + ) + + AuthOauthProviderORM.get_auth_provider_from_iss = AsyncMock( + return_value=mock_provider + ) + + result = await auth_controller.get_login_provider_db_by_iss("test_issuer") + + assert isinstance( + result, LoginProvider + ), "Result should be a LoginProvider instance" + assert result.id == 1, "Provider ID should match the mock data" + assert ( + result.name == "Test Provider" + ), "Provider name should match the mock data" + + @pytest.mark.asyncio + async def test_get_login_provider_db_by_iss_not_found( + self, auth_controller: AuthController + ): + AuthOauthProviderORM.get_auth_provider_from_iss = AsyncMock(return_value=None) + + result = await auth_controller.get_login_provider_db_by_iss("invalid_issuer") + assert result is None, "Result should be None for non-existent issuer" diff --git a/tests/test_discovery_controller.py b/tests/test_discovery_controller.py new file mode 100644 index 0000000..38e53da --- /dev/null +++ b/tests/test_discovery_controller.py @@ -0,0 +1,85 @@ +from typing import List +from unittest.mock import AsyncMock, patch + +import pytest +from openg2p_portal_api.controllers.discovery_controller import DiscoveryController +from openg2p_portal_api.models.program import ProgramBase +from openg2p_portal_api.services.program_service import ProgramService + +TEST_DATA = { + "PROGRAM_NAME": "Test Program", + "PROGRAM_DESCRIPTION": "Test program description", + "ORG_NAME": "Test Org", + "KEYWORD": "test_program", + "NONEXISTENT_KEYWORD": "nonexistent", +} + + +class TestDiscoveryController: + @pytest.fixture + def mock_program_service(self): + return AsyncMock(spec=ProgramService) + + @pytest.fixture + def discovery_controller(self, mock_program_service): + with patch( + "openg2p_portal_api.services.program_service.ProgramService.get_component", + return_value=mock_program_service, + ): + controller = DiscoveryController() + return controller + + @pytest.mark.asyncio + async def test_get_program_by_keyword( + self, discovery_controller, mock_program_service + ): + expected_response: List[ProgramBase] = [ + ProgramBase( + id=1, + name=TEST_DATA["PROGRAM_NAME"], + description=TEST_DATA["PROGRAM_DESCRIPTION"], + start_date="2024-01-01", + end_date="2024-12-31", + status="active", + program_type="benefit", + organization_id=1, + organization_name=TEST_DATA["ORG_NAME"], + ) + ] + mock_program_service.get_program_by_key_service.return_value = expected_response + + response = await discovery_controller.get_program_by_keyword( + keyword=TEST_DATA["KEYWORD"] + ) + + assert ( + response == expected_response + ), f"Expected response to be {expected_response}" + mock_program_service.get_program_by_key_service.assert_awaited_once_with( + TEST_DATA["KEYWORD"] + ) + + @pytest.mark.asyncio + async def test_get_program_by_keyword_empty_result( + self, discovery_controller, mock_program_service + ): + mock_program_service.get_program_by_key_service.return_value = [] + + response = await discovery_controller.get_program_by_keyword( + keyword=TEST_DATA["NONEXISTENT_KEYWORD"] + ) + + assert response == [], "Expected response to be an empty list" + mock_program_service.get_program_by_key_service.assert_awaited_once_with( + TEST_DATA["NONEXISTENT_KEYWORD"] + ) + + @pytest.mark.asyncio + async def test_program_service_property( + self, discovery_controller, mock_program_service + ): + service = discovery_controller.program_service + + assert ( + service == mock_program_service + ), "Expected program service to match the mock" diff --git a/tests/test_document_file_controller.py b/tests/test_document_file_controller.py new file mode 100644 index 0000000..5209512 --- /dev/null +++ b/tests/test_document_file_controller.py @@ -0,0 +1,132 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import UploadFile +from openg2p_fastapi_auth.models.credentials import AuthCredentials +from openg2p_fastapi_common.errors.http_exceptions import ( + BadRequestError, + UnauthorizedError, +) +from openg2p_portal_api.controllers.document_file_controller import ( + DocumentFileController, +) +from openg2p_portal_api.models.document_file import DocumentFile + + +@pytest.fixture +def document_controller(): + controller = DocumentFileController() + controller._file_service = MagicMock() + return controller + + +@pytest.fixture +def auth_credentials(): + return AuthCredentials(partner_id=1, credentials="test_token") + + +@pytest.fixture +def unauthorized_credentials(): + return AuthCredentials(partner_id=0, credentials="test_token") + + +@pytest.fixture +def mock_file(): + return MagicMock(spec=UploadFile) + + +class TestDocumentFileController: + @pytest.mark.asyncio + async def test_upload_document_success( + self, document_controller, auth_credentials, mock_file + ): + program_id = 1 + file_tag = "test_tag" + expected_response = DocumentFile(id=1, name="test.pdf", url="test_url") + document_controller.file_service.upload_document = AsyncMock( + return_value=expected_response + ) + + result = await document_controller.upload_document( + program_id, auth_credentials, file_tag, mock_file + ) + + assert ( + result == expected_response + ), "The uploaded document response does not match the expected response." + document_controller.file_service.upload_document.assert_called_once_with( + file=mock_file, programid=program_id, file_tag=file_tag + ) + + @pytest.mark.asyncio + async def test_upload_document_unauthorized( + self, document_controller, unauthorized_credentials, mock_file + ): + with pytest.raises(UnauthorizedError) as exc_info: + await document_controller.upload_document( + 1, unauthorized_credentials, "tag", mock_file + ) + assert ( + str(exc_info.value.detail) == "Unauthorized. Partner Not Found in Registry." + ), "UnauthorizedError was not raised with the expected message." + + @pytest.mark.asyncio + async def test_upload_document_failure( + self, document_controller, auth_credentials, mock_file + ): + document_controller.file_service.upload_document = AsyncMock( + side_effect=Exception() + ) + + with pytest.raises(BadRequestError) as exc_info: + await document_controller.upload_document( + 1, auth_credentials, "tag", mock_file + ) + assert ( + str(exc_info.value.detail) == "File upload failed!" + ), "BadRequestError was not raised with the expected message." + + @pytest.mark.asyncio + async def test_get_document_by_id_success( + self, document_controller, auth_credentials + ): + document_id = 1 + expected_document = DocumentFile(id=1, name="test.pdf", url="test_url") + document_controller.file_service.get_document_by_id = AsyncMock( + return_value=expected_document + ) + + result = await document_controller.get_document_by_id( + document_id, auth_credentials + ) + + assert ( + result == expected_document + ), "The retrieved document does not match the expected document." + document_controller.file_service.get_document_by_id.assert_called_once_with( + document_id + ) + + @pytest.mark.asyncio + async def test_get_document_by_id_unauthorized(self, document_controller): + auth_credentials = AuthCredentials(partner_id=0, credentials="test_token") + + with pytest.raises(UnauthorizedError) as exc_info: + await document_controller.get_document_by_id(1, auth_credentials) + assert ( + str(exc_info.value.detail) == "Unauthorized. Partner Not Found in Registry." + ), "UnauthorizedError was not raised with the expected message." + + @pytest.mark.asyncio + async def test_get_document_by_id_failure( + self, document_controller, auth_credentials + ): + document_controller.file_service.get_document_by_id = AsyncMock( + side_effect=Exception() + ) + + with pytest.raises(BadRequestError) as exc_info: + await document_controller.get_document_by_id(1, auth_credentials) + assert ( + str(exc_info.value.detail) == "Failed to retrieve document by ID" + ), "BadRequestError was not raised with the expected message." diff --git a/tests/test_document_file_service.py b/tests/test_document_file_service.py new file mode 100644 index 0000000..5417a90 --- /dev/null +++ b/tests/test_document_file_service.py @@ -0,0 +1,297 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import UploadFile +from openg2p_fastapi_common.errors.http_exceptions import BadRequestError +from openg2p_portal_api.models.document_file import DocumentFile +from openg2p_portal_api.models.orm.document_file_orm import DocumentFileORM +from openg2p_portal_api.models.orm.document_store_orm import DocumentStoreORM +from openg2p_portal_api.models.orm.document_tag_orm import DocumentTagORM +from openg2p_portal_api.models.orm.program_orm import ProgramORM +from openg2p_portal_api.services.document_file_service import DocumentFileService +from sqlalchemy.ext.asyncio import AsyncSession + +TEST_CONSTANTS = { + "PROGRAM_ID": 1, + "FILE_TAG": "test_tag", + "DOCUMENT_NAME": "test.pdf", + "DOCUMENT_ID": 1, + "DOCUMENT_NOT_FOUND_ID": 999, + "SUCCESS_MESSAGE": "File uploaded successfully.", + "FILESYSTEM_UNSUPPORTED_MESSAGE": "Uploading files via the filesystem is currently not supported.", + "INVALID_BACKEND_MESSAGE": "Backend type should be either amazon_s3 or filesystem.", + "EMPTY_CONTENT_ERROR": "Failed to upload document: Content must not be None.", +} + + +@pytest.fixture +def document_service(): + service = DocumentFileService() + mock_session = AsyncMock(spec=AsyncSession) + context_manager = AsyncMock() + context_manager.__aenter__.return_value = mock_session + service.async_session_maker = MagicMock(return_value=context_manager) + return service + + +@pytest.fixture +def mock_session(): + return AsyncMock(spec=AsyncSession) + + +@pytest.fixture +def mock_file(): + file = MagicMock(spec=UploadFile) + file.filename = TEST_CONSTANTS["DOCUMENT_NAME"] + file.file = MagicMock() + file.file.seek = MagicMock() + return file + + +@pytest.fixture +def mock_program(): + return ProgramORM( + id=TEST_CONSTANTS["PROGRAM_ID"], company_id=1, supporting_documents_store=1 + ) + + +@pytest.fixture +def mock_backend_s3(): + return DocumentStoreORM( + id=1, server_env_defaults={"x_backend_type_env_default": "amazon_s3"} + ) + + +@pytest.fixture +def mock_backend_filesystem(): + return DocumentStoreORM( + id=1, server_env_defaults={"x_backend_type_env_default": "filesystem"} + ) + + +class TestDocumentFileService: + @pytest.mark.asyncio + async def test_get_document_by_id_success(self, document_service, mock_session): + mock_document = DocumentFileORM( + name=TEST_CONSTANTS["DOCUMENT_NAME"], + backend_id=1, + file_size=1000, + checksum="abc123", + company_id=1, + active=True, + ) + + mock_result = MagicMock() + mock_result.scalar_one_or_none.return_value = mock_document + mock_session.execute.return_value = mock_result + document_service.async_session_maker.return_value.__aenter__.return_value = ( + mock_session + ) + + result = await document_service.get_document_by_id( + TEST_CONSTANTS["DOCUMENT_ID"] + ) + + assert isinstance( + result, DocumentFile + ), "Result should be an instance of DocumentFile" + assert ( + result.name == mock_document.name + ), f"Document name should be {mock_document.name}" + + @pytest.mark.asyncio + async def test_get_document_by_id_not_found(self, document_service, mock_session): + mock_result = MagicMock() + mock_result.scalar_one_or_none.return_value = None + mock_session.execute.return_value = mock_result + + document_service.async_session_maker.return_value.__aenter__.return_value = ( + mock_session + ) + + with pytest.raises(BadRequestError) as exc_info: + await document_service.get_document_by_id( + TEST_CONSTANTS["DOCUMENT_NOT_FOUND_ID"] + ) + assert ( + str(exc_info.value.detail) == "Document not found" + ), "Exception detail should indicate document not found" + + @pytest.mark.asyncio + async def test_upload_s3_success( + self, document_service, mock_session, mock_file, mock_program, mock_backend_s3 + ): + mock_result = AsyncMock() + mock_result.scalars = MagicMock() + mock_result.scalars.return_value.first.side_effect = [ + mock_program, + mock_backend_s3, + ] + mock_session.execute.return_value = mock_result + + document_service.async_session_maker.return_value.__aenter__.return_value = ( + mock_session + ) + + with patch("boto3.client") as mock_boto3: + mock_s3_client = MagicMock() + mock_boto3.return_value = mock_s3_client + + result = await document_service.s3_storage_system( + mock_file, TEST_CONSTANTS["PROGRAM_ID"], mock_backend_s3 + ) + + assert ( + result["message"] == TEST_CONSTANTS["SUCCESS_MESSAGE"] + ), "S3 upload should return success message" + mock_s3_client.upload_fileobj.assert_called_once() + + @pytest.mark.asyncio + async def test_upload_document_filesystem( + self, + document_service, + mock_session, + mock_file, + mock_program, + mock_backend_filesystem, + ): + mock_result = AsyncMock() + mock_result.scalars = MagicMock() + mock_result.scalars.return_value.first.side_effect = [ + mock_program, + mock_backend_filesystem, + ] + mock_session.execute.return_value = mock_result + + document_service.async_session_maker.return_value.__aenter__.return_value = ( + mock_session + ) + + result = await document_service.upload_document( + mock_file, TEST_CONSTANTS["PROGRAM_ID"], TEST_CONSTANTS["FILE_TAG"] + ) + + assert ( + result["message"] == TEST_CONSTANTS["FILESYSTEM_UNSUPPORTED_MESSAGE"] + ), "Filesystem upload should return unsupported message" + + @pytest.mark.asyncio + async def test_upload_document_s3( + self, document_service, mock_session, mock_file, mock_program, mock_backend_s3 + ): + mock_tag = DocumentTagORM(name=TEST_CONSTANTS["FILE_TAG"]) + mock_document = DocumentFileORM( + name=TEST_CONSTANTS["DOCUMENT_NAME"], + backend_id=1, + file_size=1000, + checksum="abc123", + company_id=1, + active=True, + ) + + mock_session.execute.side_effect = [ + AsyncMock( + scalars=MagicMock( + return_value=MagicMock(first=MagicMock(return_value=mock_program)) + ) + ), + AsyncMock( + scalars=MagicMock( + return_value=MagicMock( + first=MagicMock(return_value=mock_backend_s3) + ) + ) + ), + AsyncMock( + scalars=MagicMock( + return_value=MagicMock(first=MagicMock(return_value=None)) + ) + ), + AsyncMock( + scalars=MagicMock( + return_value=MagicMock(first=MagicMock(return_value=mock_tag)) + ) + ), + AsyncMock( + scalars=MagicMock( + return_value=MagicMock(first=MagicMock(return_value=mock_document)) + ) + ), + AsyncMock(), + ] + + document_service.async_session_maker.return_value.__aenter__.return_value = ( + mock_session + ) + mock_file.read = AsyncMock(return_value=b"test content") + + with patch.object(document_service, "s3_storage_system") as mock_s3_storage: + mock_s3_storage.return_value = { + "message": TEST_CONSTANTS["SUCCESS_MESSAGE"] + } + result = await document_service.upload_document( + mock_file, TEST_CONSTANTS["PROGRAM_ID"], TEST_CONSTANTS["FILE_TAG"] + ) + + assert ( + result["message"] == TEST_CONSTANTS["SUCCESS_MESSAGE"] + ), "S3 upload should return success message" + mock_s3_storage.assert_called_once_with( + mock_file, f"test-pdf-{mock_document.id}", mock_backend_s3 + ) + assert ( + mock_session.add.call_count == 2 + ), "Two objects should be added to the session" + + @pytest.mark.asyncio + async def test_upload_document_empty_content( + self, document_service, mock_session, mock_file, mock_program, mock_backend_s3 + ): + mock_result = AsyncMock() + mock_result.scalars = MagicMock() + mock_result.scalars.return_value.first.side_effect = [ + mock_program, + mock_backend_s3, + ] + mock_session.execute.return_value = mock_result + + document_service.async_session_maker.return_value.__aenter__.return_value = ( + mock_session + ) + mock_file.read = AsyncMock(return_value=None) + + with pytest.raises(BadRequestError) as exc_info: + await document_service.upload_document( + mock_file, TEST_CONSTANTS["PROGRAM_ID"], TEST_CONSTANTS["FILE_TAG"] + ) + assert ( + str(exc_info.value.detail) == TEST_CONSTANTS["EMPTY_CONTENT_ERROR"] + ), "Exception detail should indicate content must not be None" + + @pytest.mark.asyncio + async def test_upload_document_invalid_backend( + self, document_service, mock_session, mock_file, mock_program + ): + mock_backend_invalid = DocumentStoreORM( + id=1, server_env_defaults={"x_backend_type_env_default": "invalid_backend"} + ) + + mock_result = AsyncMock() + mock_result.scalars = MagicMock() + mock_result.scalars.return_value.first.side_effect = [ + mock_program, + mock_backend_invalid, + ] + mock_session.execute.return_value = mock_result + + document_service.async_session_maker.return_value.__aenter__.return_value = ( + mock_session + ) + + result = await document_service.upload_document( + mock_file, TEST_CONSTANTS["PROGRAM_ID"], TEST_CONSTANTS["FILE_TAG"] + ) + + assert ( + result["message"] == TEST_CONSTANTS["INVALID_BACKEND_MESSAGE"] + ), "Invalid backend should return appropriate error message" diff --git a/tests/test_exception.py b/tests/test_exception.py new file mode 100644 index 0000000..102c594 --- /dev/null +++ b/tests/test_exception.py @@ -0,0 +1,33 @@ +import pytest +from openg2p_fastapi_common.errors.http_exceptions import BadRequestError +from openg2p_portal_api.exception import handle_exception + + +class TestExceptionHandler: + @pytest.mark.parametrize( + "error, expected_message", + [ + (ValueError("Test error message"), "Error: Test error message"), + (RuntimeError("Something went wrong"), "Error: Something went wrong"), + (Exception(), "Error: "), + ], + ) + def test_handle_exception(self, error, expected_message): + with pytest.raises(BadRequestError) as exc_info: + handle_exception(error) + + assert expected_message in str( + exc_info.value.detail + ), f"Expected '{expected_message}' in error detail, but got '{exc_info.value.detail}'" + + def test_handle_exception_with_custom_prefix(self): + test_error = RuntimeError("Something went wrong") + custom_prefix = "Custom Error" + expected_message = "Custom Error: Something went wrong" + + with pytest.raises(BadRequestError) as exc_info: + handle_exception(test_error, message_prefix=custom_prefix) + + assert expected_message in str( + exc_info.value.detail + ), f"Expected '{expected_message}' in error detail, but got '{exc_info.value.detail}'" diff --git a/tests/test_form_controller.py b/tests/test_form_controller.py new file mode 100644 index 0000000..3b1814a --- /dev/null +++ b/tests/test_form_controller.py @@ -0,0 +1,179 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest +from openg2p_fastapi_common.errors.http_exceptions import ( + BadRequestError, + UnauthorizedError, +) +from openg2p_portal_api.controllers.form_controller import FormController +from openg2p_portal_api.models.credentials import AuthCredentials +from openg2p_portal_api.models.form import ProgramForm, ProgramRegistrantInfo +from openg2p_portal_api.models.program import Program + + +@pytest.fixture +def mock_form_service(): + service = MagicMock() + service.get_program_form = AsyncMock() + service.create_form_draft = AsyncMock() + service.submit_application_form = AsyncMock() + return service + + +@pytest.fixture +def mock_program_service(): + service = MagicMock() + service.get_program_by_id_service = AsyncMock() + return service + + +@pytest.fixture +def form_controller(mock_form_service, mock_program_service): + controller = FormController() + controller._form_service = mock_form_service + controller._program_service = mock_program_service + return controller + + +@pytest.fixture +def auth_credentials(): + return AuthCredentials(partner_id=1, credentials="test_token") + + +@pytest.fixture +def program_registrant_info(): + return ProgramRegistrantInfo(name="Test User", email="test@example.com") + + +class TestFormController: + @pytest.mark.asyncio + async def test_get_program_form_success(self, form_controller, auth_credentials): + program_id = 1 + expected_form = ProgramForm(id=1, program_id=program_id, status="draft") + form_controller.form_service.get_program_form.return_value = expected_form + + result = await form_controller.get_program_form(program_id, auth_credentials) + + assert result == expected_form, "Retrieved form should match expected form" + form_controller.form_service.get_program_form.assert_called_once_with( + program_id, auth_credentials.partner_id + ), "get_program_form should be called with correct parameters" + + @pytest.mark.asyncio + async def test_get_program_form_unauthorized(self, form_controller): + program_id = 1 + auth_credentials = AuthCredentials(partner_id=0, credentials="test_token") + + with pytest.raises(UnauthorizedError): + await form_controller.get_program_form(program_id, auth_credentials) + + @pytest.mark.asyncio + async def test_create_or_update_form_draft_success( + self, form_controller, auth_credentials, program_registrant_info + ): + program_id = 1 + expected_form = ProgramForm(id=1, program_id=program_id, status="draft") + form_controller.form_service.create_form_draft.return_value = expected_form + + result = await form_controller.create_or_update_form_draft( + program_id, program_registrant_info, auth_credentials + ) + + assert result == expected_form, "Created draft form should match expected form" + form_controller.form_service.create_form_draft.assert_called_once_with( + program_id, program_registrant_info, auth_credentials.partner_id + ), "create_form_draft should be called with correct parameters" + + @pytest.mark.asyncio + async def test_create_or_update_form_draft_unauthorized( + self, form_controller, program_registrant_info + ): + program_id = 1 + auth_credentials = AuthCredentials(partner_id=0, credentials="test_token") + + with pytest.raises(UnauthorizedError): + await form_controller.create_or_update_form_draft( + program_id, program_registrant_info, auth_credentials + ) + + @pytest.mark.asyncio + async def test_submit_form_success( + self, form_controller, auth_credentials, program_registrant_info + ): + program_id = 1 + expected_form = ProgramForm(id=1, program_id=program_id, status="submitted") + Program(id=program_id, name="Test Program", is_portal_form_mapped=True) + form_controller.form_service.submit_application_form.return_value = ( + expected_form + ) + + result = await form_controller.submit_form( + program_id, program_registrant_info, auth_credentials + ) + + assert result == expected_form, "Submitted form should match expected form" + form_controller.program_service.get_program_by_id_service.assert_called_once_with( + program_id, auth_credentials.partner_id + ), "get_program_by_id_service should be called with correct parameters" + form_controller.form_service.submit_application_form.assert_called_once_with( + program_id, program_registrant_info, auth_credentials.partner_id + ), "submit_application_form should be called with correct parameters" + + @pytest.mark.asyncio + async def test_submit_form_unauthorized( + self, form_controller, program_registrant_info + ): + program_id = 1 + auth_credentials = AuthCredentials(partner_id=0, credentials="test_token") + + with pytest.raises(UnauthorizedError): + await form_controller.submit_form( + program_id, program_registrant_info, auth_credentials + ) + + @pytest.mark.asyncio + async def test_submit_form_not_mapped( + self, form_controller, auth_credentials, program_registrant_info + ): + program_id = 1 + program = Program( + id=program_id, name="Test Program", is_portal_form_mapped=False + ) + form_controller.program_service.get_program_by_id_service.return_value = program + + with pytest.raises(BadRequestError): + await form_controller.submit_form( + program_id, program_registrant_info, auth_credentials + ) + + @pytest.mark.asyncio + async def test_get_program_form_not_found(self, form_controller, auth_credentials): + program_id = 1 + form_controller.form_service.get_program_form.side_effect = BadRequestError( + code="G2P-REQ-400", message="Program ID not Found", http_status_code=400 + ) + + with pytest.raises(BadRequestError) as exc: + await form_controller.get_program_form(program_id, auth_credentials) + assert ( + exc.value.message == "Program ID not Found" + ), "Should raise BadRequestError with correct message" + + @pytest.mark.asyncio + async def test_submit_form_draft_error( + self, form_controller, auth_credentials, program_registrant_info + ): + program_id = 1 + form_controller.form_service.create_form_draft.side_effect = BadRequestError( + code="G2P-REQ-400", + message="Error: In creating the draft", + http_status_code=400, + ) + + with pytest.raises(BadRequestError) as exc: + await form_controller.create_or_update_form_draft( + program_id, program_registrant_info, auth_credentials + ) + assert ( + exc.value.message == "Error: In creating the draft" + ), "Should raise BadRequestError with correct message" diff --git a/tests/test_form_service.py b/tests/test_form_service.py new file mode 100644 index 0000000..43f2c80 --- /dev/null +++ b/tests/test_form_service.py @@ -0,0 +1,209 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import HTTPException +from openg2p_portal_api.models.form import ProgramForm +from openg2p_portal_api.models.orm.program_orm import ProgramORM +from openg2p_portal_api.models.orm.program_registrant_info_orm import ( + ProgramRegistrantInfoDraftORM, +) +from openg2p_portal_api.services.form_service import FormService + +TEST_DATA = { + "PROGRAM_ID": 1, + "REGISTRANT_ID": 1, + "MOCK_PROGRAM_NAME": "Test Program", + "MOCK_PROGRAM_DESCRIPTION": "Test Description", + "MOCK_EMAIL": "test@example.com", + "MOCK_USER_NAME": "Test User", +} + + +@pytest.fixture +def form_service(): + service = FormService() + service.membership_service = MagicMock() + service.partner_service = MagicMock() + return service + + +class TestFormService: + @pytest.mark.asyncio + async def test_get_program_form_success(self, form_service): + program_id = TEST_DATA["PROGRAM_ID"] + registrant_id = TEST_DATA["REGISTRANT_ID"] + mock_program = MagicMock() + mock_program.id = program_id + mock_program.name = TEST_DATA["MOCK_PROGRAM_NAME"] + mock_program.description = TEST_DATA["MOCK_PROGRAM_DESCRIPTION"] + mock_program.form.id = 1 + mock_program.form.schema = '{"fields": []}' + + ProgramORM.get_program_form = AsyncMock(return_value=mock_program) + ProgramRegistrantInfoDraftORM.get_draft_reg_info_by_id = AsyncMock( + return_value=None + ) + + result = await form_service.get_program_form(program_id, registrant_id) + + assert isinstance( + result, ProgramForm + ), "Result should be a ProgramForm instance" + assert result.program_id == program_id, "Program ID should match" + assert ( + result.program_name == TEST_DATA["MOCK_PROGRAM_NAME"] + ), "Program name should match" + assert ( + result.program_description == TEST_DATA["MOCK_PROGRAM_DESCRIPTION"] + ), "Program description should match" + + @pytest.mark.asyncio + async def test_get_program_form_not_found(self, form_service): + program_id = 999 + registrant_id = TEST_DATA["REGISTRANT_ID"] + ProgramORM.get_program_form = AsyncMock(return_value=None) + + with pytest.raises(HTTPException) as exc: + await form_service.get_program_form(program_id, registrant_id) + assert exc.value.status_code == 400, "Status code should be 400" + assert exc.value.detail == "Program ID not Found", "Error detail should match" + + @pytest.mark.asyncio + async def test_create_form_draft_new(self, form_service): + program_id = TEST_DATA["PROGRAM_ID"] + registrant_id = TEST_DATA["REGISTRANT_ID"] + form_data = MagicMock() + form_data.program_registrant_info = { + "name": TEST_DATA["MOCK_USER_NAME"], + "email": TEST_DATA["MOCK_EMAIL"], + } + + ProgramRegistrantInfoDraftORM.get_draft_reg_info_by_id = AsyncMock( + return_value=None + ) + + with patch("sqlalchemy.ext.asyncio.AsyncSession.commit") as mock_commit: + result = await form_service.create_form_draft( + program_id, form_data, registrant_id + ) + assert ( + result == "Successfully submitted the draft!!" + ), "Draft submission message should match" + mock_commit.assert_called_once() + + @pytest.mark.asyncio + async def test_submit_application_form_success(self, form_service): + program_id = TEST_DATA["PROGRAM_ID"] + registrant_id = TEST_DATA["REGISTRANT_ID"] + form_data = MagicMock() + form_data.program_registrant_info = { + "name": TEST_DATA["MOCK_USER_NAME"], + "email": TEST_DATA["MOCK_EMAIL"], + } + + mock_program = MagicMock(is_multiple_form_submission=True) + + ProgramRegistrantInfoDraftORM.get_draft_reg_info_by_id = AsyncMock( + return_value=None + ) + form_service.membership_service.check_and_create_mem = AsyncMock( + return_value=123 + ) + form_service.partner_service.update_partner_info = AsyncMock( + return_value=["name"] + ) + + with patch("sqlalchemy.ext.asyncio.AsyncSession.get") as mock_get, patch( + "sqlalchemy.ext.asyncio.AsyncSession.execute" + ) as mock_execute, patch("sqlalchemy.ext.asyncio.AsyncSession.commit"): + mock_get.return_value = mock_program + mock_execute.return_value = MagicMock( + scalars=MagicMock( + return_value=MagicMock(first=MagicMock(return_value=None)) + ) + ) + + result = await form_service.submit_application_form( + program_id, form_data, registrant_id + ) + + assert ( + "Successfully applied into the program! Application ID:" in result + ), "Application submission message should match" + assert len(result.split()[-1]) == 11, "Application ID length should be 11" + form_service.membership_service.check_and_create_mem.assert_called_once_with( + program_id, registrant_id + ) + form_service.partner_service.update_partner_info.assert_called_once() + + @pytest.mark.asyncio + async def test_submit_application_form_multiple_not_allowed(self, form_service): + program_id = TEST_DATA["PROGRAM_ID"] + registrant_id = TEST_DATA["REGISTRANT_ID"] + form_data = MagicMock() + form_data.program_registrant_info = {"name": TEST_DATA["MOCK_USER_NAME"]} + + mock_program = MagicMock(is_multiple_form_submission=False) + mock_application = MagicMock() + mock_result = MagicMock( + scalars=MagicMock( + return_value=MagicMock(first=MagicMock(return_value=mock_application)) + ) + ) + + with patch("sqlalchemy.ext.asyncio.AsyncSession.get") as mock_get, patch( + "sqlalchemy.ext.asyncio.AsyncSession.execute" + ) as mock_execute: + mock_get.return_value = mock_program + mock_execute.return_value = mock_result + + result = await form_service.submit_application_form( + program_id, form_data, registrant_id + ) + + assert ( + result + == "Error: Multiple form submissions are not allowed for this program." + ), "Error message for multiple submissions should match" + + def test_compute_application_id(self, form_service): + today = datetime.today() + expected_prefix = today.strftime("%d%m%y") + + result = form_service._compute_application_id() + + assert result.startswith( + expected_prefix + ), "Application ID should start with today's date" + assert len(result) == 11, "Application ID length should be 11" + + def test_clean_program_registrant_info(self, form_service): + program_info = {"name": "Test", "email": TEST_DATA["MOCK_EMAIL"], "age": 25} + updated_fields = ["name", "email"] + + result = form_service.clean_program_registrant_info( + program_info, updated_fields + ) + + assert "name" not in result, "Name should be removed from result" + assert "email" not in result, "Email should be removed from result" + assert result == {"age": 25}, "Result should only contain age" + + def test_clean_program_registrant_info_empty(self, form_service): + program_info = {} + updated_fields = ["name", "email"] + + result = form_service.clean_program_registrant_info( + program_info, updated_fields + ) + assert result == {}, "Result should be empty" + + def test_clean_program_registrant_info_no_updates(self, form_service): + program_info = {"name": "Test", "email": TEST_DATA["MOCK_EMAIL"]} + updated_fields = [] + + result = form_service.clean_program_registrant_info( + program_info, updated_fields + ) + assert result == program_info, "Result should match original program info" diff --git a/tests/test_membership_service.py b/tests/test_membership_service.py new file mode 100644 index 0000000..db63c8f --- /dev/null +++ b/tests/test_membership_service.py @@ -0,0 +1,88 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from openg2p_portal_api.models.orm.program_membership_orm import ProgramMembershipORM +from openg2p_portal_api.services.membership_service import MembershipService +from sqlalchemy.exc import IntegrityError + + +@pytest.fixture +def membership_service(): + return MembershipService() + + +@pytest.fixture +def mock_session(): + session = AsyncMock() + async_session = AsyncMock() + async_session.__aenter__.return_value = session + async_session.__aexit__.return_value = None + return session, async_session + + +@pytest.fixture +def mock_session_maker(mock_session): + session, async_session = mock_session + with patch( + "openg2p_portal_api.services.membership_service.async_sessionmaker", + return_value=lambda: async_session, + ) as session_maker: + yield session_maker + + +class TestMembershipService: + @pytest.mark.asyncio + async def test_check_and_create_mem_existing(self): + service = MembershipService() + mock_membership = MagicMock(id=1) + + with patch.object( + ProgramMembershipORM, + "get_membership_by_id", + new_callable=AsyncMock, + return_value=mock_membership, + ): + result = await service.check_and_create_mem(1, 1) + assert ( + result == 1 + ), "Should return existing membership ID when membership already exists" + + @pytest.mark.asyncio + async def test_check_and_create_mem_new_success( + self, mock_session, mock_session_maker + ): + service = MembershipService() + session, _ = mock_session + + with patch.object( + ProgramMembershipORM, + "get_membership_by_id", + new_callable=AsyncMock, + return_value=None, + ): + session.refresh = AsyncMock() + + await service.check_and_create_mem(1, 1) + session.add.assert_called_once(), "Should add new membership to session" + session.commit.assert_called_once(), "Should commit the new membership" + session.refresh.assert_called_once(), "Should refresh the session after commit" + + @pytest.mark.asyncio + async def test_check_and_create_mem_integrity_error( + self, mock_session, mock_session_maker + ): + service = MembershipService() + session, _ = mock_session + + with patch.object( + ProgramMembershipORM, + "get_membership_by_id", + new_callable=AsyncMock, + return_value=None, + ): + session.commit.side_effect = IntegrityError(None, None, None) + + result = await service.check_and_create_mem(1, 1) + assert ( + result == "Could not add to registrant to program!!" + ), "Should return error message when IntegrityError occurs" diff --git a/tests/test_oauth_controller.py b/tests/test_oauth_controller.py new file mode 100644 index 0000000..9fb9e6a --- /dev/null +++ b/tests/test_oauth_controller.py @@ -0,0 +1,155 @@ +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from fastapi import Request +from openg2p_fastapi_auth.models.orm.login_provider import LoginProviderTypes +from openg2p_fastapi_common.utils import cookie_utils +from openg2p_portal_api.controllers.oauth_controller import OAuthController +from openg2p_portal_api.models.orm.auth_oauth_provider import AuthOauthProviderORM + +TEST_CONSTANTS = { + "CLIENT_ID": "test_client_id", + "CLIENT_SECRET": "test_client_secret", + "REDIRECT_URI": "https://example.com/callback", + "DASHBOARD_URL": "https://example.com/dashboard", + "TOKEN_ENDPOINT": "https://example.com/token", + "CODE": "test_code", + "CODE_VERIFIER": "test_code_verifier", + "PROVIDER_ID": "test_login_provider_id", + "SUB": "test_user", + "ID_TYPE": "test_type", +} + +ASSERT_MESSAGES = { + "NOT_HTTPX_RESPONSE": "Response should not be an instance of httpx.Response", + "LOCATION_HEADER": "Response should contain location header", + "DASHBOARD_REDIRECT": "Should redirect to dashboard URL", + "MOCK_RESPONSE": "Should return mock response for empty state", +} + + +@pytest.fixture +def oauth_controller(): + with patch.object(OAuthController, "auth_controller", new=MagicMock()) as mock_auth: + controller = OAuthController() + controller.partner_service = MagicMock() + mock_auth.return_value = MagicMock() + return controller + + +@pytest.fixture +def mock_login_provider(): + return MagicMock( + type=LoginProviderTypes.oauth2_auth_code, + provider="test_provider", + authorization_parameters={ + "client_id": TEST_CONSTANTS["CLIENT_ID"], + "client_secret": TEST_CONSTANTS["CLIENT_SECRET"], + "redirect_uri": TEST_CONSTANTS["REDIRECT_URI"], + "scope": "read write", + "authorize_endpoint": "https://example.com/auth", + "token_endpoint": TEST_CONSTANTS["TOKEN_ENDPOINT"], + "validate_endpoint": "https://example.com/validate", + "jwks_endpoint": "https://example.com/jwks", + "code_verifier": TEST_CONSTANTS["CODE_VERIFIER"], + "enable_pkce": True, + "client_assertion_type": "client_secret", + }, + ) + + +class TestOAuthController: + class TestOAuthCallback: + @pytest.mark.asyncio + async def test_successful_oauth_callback_with_valid_state( + self, oauth_controller, mock_login_provider + ): + mock_request = MagicMock(spec=Request) + mock_request.query_params = { + "state": json.dumps( + { + "p": TEST_CONSTANTS["PROVIDER_ID"], + "r": TEST_CONSTANTS["DASHBOARD_URL"], + } + ), + "code": TEST_CONSTANTS["CODE"], + } + + cookie_utils.get_response_cookies = MagicMock( + side_effect=[["access_token"], ["id_token"]] + ) + + oauth_controller.auth_controller.get_oauth_validation_data = AsyncMock( + return_value={"sub": TEST_CONSTANTS["SUB"]} + ) + oauth_controller.auth_controller.get_login_provider_db_by_id = AsyncMock( + return_value=mock_login_provider + ) + + AuthOauthProviderORM.get_auth_id_type_config = AsyncMock( + return_value={"id_type": TEST_CONSTANTS["ID_TYPE"]} + ) + oauth_controller.partner_service.check_and_create_partner = AsyncMock() + + mock_response = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = { + "access_token": "test_access_token", + "id_token": "test_id_token", + } + mock_response.is_success = True + + with patch("httpx.post", return_value=mock_response) as mock_httpx_post: + result = await oauth_controller.oauth_callback(mock_request) + + assert not isinstance(result, httpx.Response), ASSERT_MESSAGES[ + "NOT_HTTPX_RESPONSE" + ] + assert "location" in result.headers, ASSERT_MESSAGES["LOCATION_HEADER"] + assert ( + result.headers["location"] == TEST_CONSTANTS["DASHBOARD_URL"] + ), ASSERT_MESSAGES["DASHBOARD_REDIRECT"] + + oauth_controller.auth_controller.get_oauth_validation_data.assert_called_once_with( + auth="access_token", + id_token="id_token", + provider=mock_login_provider, + ) + + AuthOauthProviderORM.get_auth_id_type_config.assert_called_once_with( + id=TEST_CONSTANTS["PROVIDER_ID"] + ) + + oauth_controller.partner_service.check_and_create_partner.assert_called_once_with( + {"sub": TEST_CONSTANTS["SUB"]}, + id_type_config={"id_type": TEST_CONSTANTS["ID_TYPE"]}, + ) + + mock_httpx_post.assert_called_once_with( + TEST_CONSTANTS["TOKEN_ENDPOINT"], + auth=(TEST_CONSTANTS["CLIENT_ID"], TEST_CONSTANTS["CLIENT_SECRET"]), + data={ + "client_id": TEST_CONSTANTS["CLIENT_ID"], + "grant_type": "authorization_code", + "redirect_uri": TEST_CONSTANTS["REDIRECT_URI"], + "code": TEST_CONSTANTS["CODE"], + "code_verifier": TEST_CONSTANTS["CODE_VERIFIER"], + }, + ) + + @pytest.mark.asyncio + async def test_oauth_callback_empty_state(self, oauth_controller): + mock_request = MagicMock(spec=Request) + mock_request.query_params = {"state": "{}"} + mock_response = MagicMock() + + with patch.object( + OAuthController, "oauth_callback", new_callable=AsyncMock + ) as mock_super_callback: + mock_super_callback.return_value = mock_response + result = await oauth_controller.oauth_callback(mock_request) + + assert result == mock_response, ASSERT_MESSAGES["MOCK_RESPONSE"] + mock_super_callback.assert_called_once_with(mock_request) diff --git a/tests/test_partner_service.py b/tests/test_partner_service.py new file mode 100644 index 0000000..c601902 --- /dev/null +++ b/tests/test_partner_service.py @@ -0,0 +1,261 @@ +from datetime import date +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from openg2p_fastapi_common.errors.http_exceptions import InternalServerError +from openg2p_portal_api.models.orm.partner_orm import PartnerORM +from openg2p_portal_api.services.partner_service import PartnerService +from sqlalchemy import Engine +from sqlalchemy.ext.asyncio import AsyncSession + +VALID_PARTNER_DATA = { + "sub": "12345", + "name": "John Middle Doe", + "email": "john@example.com", + "gender": "male", + "birthdate": "1990/01/01", + "phone": "1234567890", +} + +VALID_ID_TYPE_CONFIG = { + "g2p_id_type": "national_id", + "company_id": 1, + "token_map": "name:name email:email gender:gender birthdate:birthdate phone:phone", + "date_format": "%Y/%m/%d", +} + + +@pytest.fixture +def mock_engine(): + engine = MagicMock(spec=Engine) + engine.execution_options.return_value = engine + return engine + + +@pytest.fixture +def mock_session(mock_engine): + session = AsyncMock(spec=AsyncSession) + session.get_bind = MagicMock(return_value=mock_engine) + session.bind = mock_engine + + sync_session = MagicMock() + sync_session.get_bind = MagicMock(return_value=mock_engine) + session.sync_session = sync_session + + session.commit = AsyncMock() + session.flush = AsyncMock() + session.add = AsyncMock() + session.refresh = AsyncMock() + session.close = AsyncMock() + + async_context = AsyncMock() + async_context.__aenter__.return_value = session + async_context.__aexit__.return_value = None + + return session, async_context + + +@pytest.fixture +def partner_service(mock_session, mock_engine): + with patch("openg2p_portal_api.services.partner_service.dbengine") as mock_dbengine: + mock_dbengine.get.return_value = mock_engine + return PartnerService() + + +class TestPartnerService: + @pytest.mark.asyncio + async def test_check_and_create_partner_success( + self, partner_service, mock_session + ): + expected_fields = ["name", "email", "phone", "gender", "birthdate"] + + with patch( + "openg2p_portal_api.models.orm.reg_id_orm.RegIDORM.get_partner_by_reg_id", + new_callable=AsyncMock, + return_value=None, + ), patch.object( + partner_service, + "get_partner_fields", + new_callable=AsyncMock, + return_value=expected_fields, + ), patch( + "openg2p_portal_api.services.partner_service.async_sessionmaker", + return_value=lambda: mock_session[1], + ): + await partner_service.check_and_create_partner( + VALID_PARTNER_DATA, VALID_ID_TYPE_CONFIG + ) + + session = mock_session[0] + self._verify_partner_creation(session) + + def _verify_partner_creation(self, session): + assert ( + session.add.call_count == 3 + ), "Session add should be called exactly 3 times" + session.commit.assert_called_once(), "Session commit should be called exactly once" + + partner_call = session.add.call_args_list[0] + created_partner = partner_call[0][0] + + assert isinstance( + created_partner, PartnerORM + ), "Created object should be instance of PartnerORM" + assert ( + created_partner.name == "DOE, JOHN MIDDLE " + ), "Partner name should be properly formatted" + assert ( + created_partner.email == "john@example.com" + ), "Partner email should match input" + assert created_partner.phone == "1234567890", "Partner phone should match input" + assert ( + created_partner.gender == "Male" + ), "Partner gender should be properly capitalized" + assert ( + created_partner.is_registrant is True + ), "Partner should be marked as registrant" + assert created_partner.active is True, "Partner should be marked as active" + assert created_partner.company_id == 1, "Partner company_id should match config" + + @pytest.mark.asyncio + async def test_check_and_create_partner_validation(self, partner_service): + test_cases = [ + (None, "ID Type not configured"), + ({}, "ID Type not configured"), + ] + + for config, expected_error in test_cases: + with pytest.raises(InternalServerError) as exc_info: + await partner_service.check_and_create_partner({}, config) + assert expected_error in str( + exc_info.value + ), f"Expected error message '{expected_error}' not found" + + @pytest.mark.parametrize( + "input_data, expected_output", + [ + (("Doe", "John", "Middle"), "DOE, JOHN MIDDLE "), + (("Smith", "Jane", ""), "SMITH, JANE "), + (("", "", ""), ""), + ], + ) + def test_create_partner_process_name( + self, partner_service, input_data, expected_output + ): + result = partner_service.create_partner_process_name(*input_data) + assert ( + result == expected_output + ), f"Name processing failed for input {input_data}" + + @pytest.mark.parametrize( + "input_gender, expected_output", + [ + ("male", "Male"), + ("female", "Female"), + ("", ""), + ("other", "Other"), + ], + ) + def test_create_partner_process_gender( + self, partner_service, input_gender, expected_output + ): + result = partner_service.create_partner_process_gender(input_gender) + assert ( + result == expected_output + ), f"Gender processing failed for input '{input_gender}'" + + def test_create_partner_process_gender_none(self, partner_service): + with pytest.raises(AttributeError): + partner_service.create_partner_process_gender(None) + + @pytest.mark.parametrize( + "date_str, date_format, expected_result", + [ + ("1990/01/01", "%Y/%m/%d", date(1990, 1, 1)), + (None, "%Y/%m/%d", None), + ], + ) + def test_create_partner_process_birthdate( + self, partner_service, date_str, date_format, expected_result + ): + result = partner_service.create_partner_process_birthdate(date_str, date_format) + assert ( + result == expected_result + ), f"Birthdate processing failed for input '{date_str}'" + + def test_create_partner_process_birthdate_invalid(self, partner_service): + with pytest.raises(ValueError, match=".*"): + partner_service.create_partner_process_birthdate("invalid_date", "%Y/%m/%d") + + @pytest.mark.asyncio + async def test_get_partner_fields(self, partner_service): + expected_fields = ["name", "email", "phone"] + + with patch.object( + PartnerORM, + "get_partner_fields", + new_callable=AsyncMock, + return_value=expected_fields, + ): + result = await partner_service.get_partner_fields() + assert ( + result == expected_fields + ), "Initial partner fields should match expected fields" + + cached_result = await partner_service.get_partner_fields() + assert ( + cached_result == expected_fields + ), "Cached partner fields should match expected fields" + PartnerORM.get_partner_fields.assert_called_once(), "get_partner_fields should be called only once due to caching" + + def test_create_partner_process_other_fields(self, partner_service): + validation = { + "field1": "value1", + "field2": {"key": "value"}, + "field3": ["item1", "item2"], + "field4": "value4", + } + mapping = "field1: map1 field2: map2 field3: map3" + partner_fields = ["field1", "field2", "field3"] + + result = partner_service.create_partner_process_other_fields( + validation, mapping, partner_fields + ) + + assert result["field1"] == "value1", "Simple field mapping should be preserved" + assert "field2" in result, "Complex field should be included" + assert "field3" in result, "Array field should be included" + assert "field4" not in result, "Unmapped field should be excluded" + assert isinstance( + result["field2"], str + ), "Complex field should be converted to string" + assert isinstance( + result["field3"], str + ), "Array field should be converted to string" + + @pytest.mark.asyncio + async def test_update_partner_info(self, partner_service): + session_mock = AsyncMock() + partner_id = "123" + data = { + "name": "New Name", + "email": "new@example.com", + "invalid_field": "value", + } + + with patch.object( + PartnerORM, "get_partner_fields", return_value=["name", "email"] + ): + updated = await partner_service.update_partner_info( + partner_id, data, session_mock + ) + + assert "name" in updated, "Valid field 'name' should be included in update" + assert ( + "email" in updated + ), "Valid field 'email' should be included in update" + assert ( + "invalid_field" not in updated + ), "Invalid field should be excluded from update" + session_mock.execute.assert_called_once(), "Session execute should be called once" + session_mock.commit.assert_called_once(), "Session commit should be called once" diff --git a/tests/test_program_controller.py b/tests/test_program_controller.py new file mode 100644 index 0000000..c0f7bbc --- /dev/null +++ b/tests/test_program_controller.py @@ -0,0 +1,174 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest +from openg2p_portal_api.controllers.program_controller import ProgramController +from openg2p_portal_api.models.credentials import AuthCredentials +from openg2p_portal_api.models.program import ( + ApplicationDetails, + BenefitDetails, + Program, + ProgramSummary, +) +from openg2p_portal_api.services.program_service import ProgramService + +TEST_DATA = { + "PARTNER_ID": 1, + "PROGRAM_ID": 1, + "TOKEN": "test_token", + "PROGRAM_NAME": "Test Program", + "DATE": "2023-01-01", +} + + +@pytest.fixture +def mock_program_service(): + service = MagicMock(spec=ProgramService) + service.get_all_program_service = AsyncMock() + service.get_program_by_id_service = AsyncMock() + service.get_program_summary_service = AsyncMock() + service.get_application_details_service = AsyncMock() + service.get_benefit_details_service = AsyncMock() + return service + + +@pytest.fixture +def program_controller(mock_program_service): + controller = ProgramController() + controller._program_service = mock_program_service + return controller + + +@pytest.fixture +def auth_credentials(): + return AuthCredentials( + partner_id=TEST_DATA["PARTNER_ID"], credentials=TEST_DATA["TOKEN"] + ) + + +class TestProgramController: + @pytest.mark.asyncio + async def test_get_programs_success(self, program_controller, auth_credentials): + expected_programs = [ + Program(id=TEST_DATA["PROGRAM_ID"], name=f"{TEST_DATA['PROGRAM_NAME']} 1"), + Program(id=2, name=f"{TEST_DATA['PROGRAM_NAME']} 2"), + ] + program_controller.program_service.get_all_program_service.return_value = ( + expected_programs + ) + result = await program_controller.get_programs(auth_credentials) + assert result == expected_programs, "Should return list of programs as expected" + program_controller.program_service.get_all_program_service.assert_called_once_with( + auth_credentials.partner_id + ), "Should call get_all_program_service with correct partner_id" + + @pytest.mark.asyncio + async def test_get_program_by_id_success( + self, program_controller, auth_credentials + ): + expected_program = Program( + id=TEST_DATA["PROGRAM_ID"], name=TEST_DATA["PROGRAM_NAME"] + ) + program_controller.program_service.get_program_by_id_service.return_value = ( + expected_program + ) + result = await program_controller.get_program_by_id( + TEST_DATA["PROGRAM_ID"], auth_credentials + ) + assert result == expected_program, "Should return the expected program" + program_controller.program_service.get_program_by_id_service.assert_called_once_with( + TEST_DATA["PROGRAM_ID"], auth_credentials.partner_id + ), "Should call get_program_by_id_service with correct program_id and partner_id" + + @pytest.mark.asyncio + async def test_get_program_summary_success( + self, program_controller, auth_credentials + ): + expected_summary = [ + ProgramSummary( + program_name=f"{TEST_DATA['PROGRAM_NAME']} 1", + enrollment_status="Active", + total_funds_awaited=1000, + total_funds_received=500, + ) + ] + program_controller.program_service.get_program_summary_service.return_value = ( + expected_summary + ) + result = await program_controller.get_program_summary(auth_credentials) + assert result == expected_summary, "Should return the expected program summary" + program_controller.program_service.get_program_summary_service.assert_called_once_with( + auth_credentials.partner_id + ), "Should call get_program_summary_service with correct partner_id" + + @pytest.mark.asyncio + async def test_get_application_details_success( + self, program_controller, auth_credentials + ): + expected_details = [ + ApplicationDetails( + program_name=f"{TEST_DATA['PROGRAM_NAME']} 1", + application_id=TEST_DATA["PROGRAM_ID"], + date_applied=TEST_DATA["DATE"], + application_status="Pending", + ) + ] + program_controller.program_service.get_application_details_service.return_value = ( + expected_details + ) + result = await program_controller.get_application_details(auth_credentials) + assert ( + result == expected_details + ), "Should return the expected application details" + program_controller.program_service.get_application_details_service.assert_called_once_with( + auth_credentials.partner_id + ), "Should call get_application_details_service with correct partner_id" + + @pytest.mark.asyncio + async def test_get_benefit_details_success( + self, program_controller, auth_credentials + ): + expected_details = [ + BenefitDetails( + program_name=f"{TEST_DATA['PROGRAM_NAME']} 1", + enrollment_status="Active", + funds_awaited=1000, + funds_received=500, + entitlement_ref_no=TEST_DATA["PROGRAM_ID"], + ) + ] + program_controller.program_service.get_benefit_details_service.return_value = ( + expected_details + ) + result = await program_controller.get_benefit_details(auth_credentials) + assert result == expected_details, "Should return the expected benefit details" + program_controller.program_service.get_benefit_details_service.assert_called_once_with( + auth_credentials.partner_id + ), "Should call get_benefit_details_service with correct partner_id" + + @pytest.mark.asyncio + async def test_get_programs_error(self, program_controller, auth_credentials): + program_controller.program_service.get_all_program_service.side_effect = ( + Exception("Failed to fetch programs from database") + ) + + with pytest.raises(Exception) as exc_info: + await program_controller.get_programs(auth_credentials) + + assert ( + str(exc_info.value) == "Failed to fetch programs from database" + ), "Should raise exception with appropriate error message" + + @pytest.mark.asyncio + async def test_get_program_by_id_error(self, program_controller, auth_credentials): + program_controller.program_service.get_program_by_id_service.side_effect = ( + Exception("Program not found in database") + ) + + with pytest.raises(Exception) as exc_info: + await program_controller.get_program_by_id( + TEST_DATA["PROGRAM_ID"], auth_credentials + ) + + assert ( + str(exc_info.value) == "Program not found in database" + ), "Should raise exception when program is not found" diff --git a/tests/test_program_service.py b/tests/test_program_service.py index 5a2f32b..53edd31 100644 --- a/tests/test_program_service.py +++ b/tests/test_program_service.py @@ -1,5 +1,262 @@ -from openg2p_portal_api.app import Initializer +from datetime import datetime +from unittest.mock import MagicMock +import pytest +from openg2p_portal_api.models.orm.program_orm import ProgramORM +from openg2p_portal_api.models.orm.program_registrant_info_orm import ( + ProgramRegistrantInfoORM, +) +from openg2p_portal_api.services.program_service import ProgramService -def test_program_service(): - Initializer() +TEST_DATA = { + "PROGRAM": { + "ID": 1, + "PARTNER_ID": 1, + "NAME": "Test Program", + "DESC": "Description of the program", + "CREATE_DATE": datetime(2024, 1, 1), + "SELF_SERVICE_PORTAL_FORM": True, + "IS_MULTIPLE_FORM_SUBMISSION": False, + }, + "STATUS": { + "NOT_APPLIED": "Not Applied", + "ACTIVE": "active", + "NO_APPLICATION": "Not submitted any application", + }, + "ERROR": {"PROGRAM_NOT_FOUND": "Program with ID 1 not found."}, + "TEST": { + "KEYWORD": "Test", + "NONEXISTENT": "NonExistent", + "BENEFIT": "Test Benefit", + "APPLICATION": "Test Application", + }, +} + + +@pytest.fixture +def program_mock(): + program = MagicMock() + program.id = TEST_DATA["PROGRAM"]["ID"] + program.name = TEST_DATA["PROGRAM"]["NAME"] + program.description = TEST_DATA["PROGRAM"]["DESC"] + program.create_date = TEST_DATA["PROGRAM"]["CREATE_DATE"] + program.self_service_portal_form = TEST_DATA["PROGRAM"]["SELF_SERVICE_PORTAL_FORM"] + program.is_multiple_form_submission = TEST_DATA["PROGRAM"][ + "IS_MULTIPLE_FORM_SUBMISSION" + ] + program.membership = [] + program.configure_mock( + **{ + "program_name": TEST_DATA["PROGRAM"]["NAME"], + "enrollment_status": TEST_DATA["STATUS"]["ACTIVE"], + "application_status": TEST_DATA["STATUS"]["ACTIVE"], + } + ) + return program + + +@pytest.fixture +def program_reg_info_mock(): + reg_info = MagicMock() + reg_info.partner_id = TEST_DATA["PROGRAM"]["PARTNER_ID"] + reg_info.state = TEST_DATA["STATUS"]["ACTIVE"] + return reg_info + + +class TestProgramService: + @pytest.mark.asyncio + async def test_get_all_programs_empty_result(self, mocker): + mocker.patch.object(ProgramORM, "get_all_programs", return_value=[]) + program_service = ProgramService() + programs = await program_service.get_all_program_service( + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"] + ) + assert programs == [], "Expected empty list when no programs are found" + + @pytest.mark.asyncio + async def test_get_all_programs_success(self, mocker, program_mock): + mocker.patch.object(ProgramORM, "get_all_programs", return_value=[program_mock]) + program_service = ProgramService() + programs = await program_service.get_all_program_service( + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"] + ) + assert len(programs) == 1, "Expected exactly one program in response" + assert ( + programs[0].id == TEST_DATA["PROGRAM"]["ID"] + ), "Program ID does not match expected value" + assert ( + programs[0].name == TEST_DATA["PROGRAM"]["NAME"] + ), "Program name does not match expected value" + assert ( + programs[0].create_date == TEST_DATA["PROGRAM"]["CREATE_DATE"] + ), "Program create date does not match expected value" + assert ( + programs[0].self_service_portal_form + == TEST_DATA["PROGRAM"]["SELF_SERVICE_PORTAL_FORM"] + ), "Self service portal form flag does not match expected value" + assert ( + programs[0].is_multiple_form_submission + == TEST_DATA["PROGRAM"]["IS_MULTIPLE_FORM_SUBMISSION"] + ), "Multiple form submission flag does not match expected value" + + @pytest.mark.asyncio + async def test_get_all_programs_with_membership( + self, mocker, program_mock, program_reg_info_mock + ): + program_mock.membership = [program_reg_info_mock] + mocker.patch.object(ProgramORM, "get_all_programs", return_value=[program_mock]) + mocker.patch.object( + ProgramRegistrantInfoORM, + "get_latest_reg_info", + return_value=program_reg_info_mock, + ) + + program_service = ProgramService() + programs = await program_service.get_all_program_service( + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"] + ) + + assert len(programs) == 1, "Expected exactly one program in response" + assert programs[0].has_applied is True, "Program should be marked as applied" + assert ( + programs[0].state == TEST_DATA["STATUS"]["ACTIVE"] + ), "Program state should be active" + assert ( + programs[0].last_application_status == TEST_DATA["STATUS"]["ACTIVE"] + ), "Last application status should be active" + + @pytest.mark.asyncio + async def test_get_program_by_id_with_membership( + self, mocker, program_mock, program_reg_info_mock + ): + program_mock.membership = [program_reg_info_mock] + mocker.patch.object( + ProgramORM, "get_all_by_program_id", return_value=program_mock + ) + mocker.patch.object( + ProgramRegistrantInfoORM, + "get_latest_reg_info", + return_value=program_reg_info_mock, + ) + + program_service = ProgramService() + program = await program_service.get_program_by_id_service( + programid=TEST_DATA["PROGRAM"]["ID"], + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"], + ) + + assert ( + program.id == TEST_DATA["PROGRAM"]["ID"] + ), "Program ID does not match expected value" + assert program.has_applied is True, "Program should be marked as applied" + assert ( + program.state == TEST_DATA["STATUS"]["ACTIVE"] + ), "Program state should be active" + assert ( + program.last_application_status == TEST_DATA["STATUS"]["ACTIVE"] + ), "Last application status should be active" + + @pytest.mark.asyncio + async def test_get_program_by_id_no_membership(self, mocker, program_mock): + program_mock.membership = [] + mocker.patch.object( + ProgramORM, "get_all_by_program_id", return_value=program_mock + ) + + program_service = ProgramService() + program = await program_service.get_program_by_id_service( + programid=TEST_DATA["PROGRAM"]["ID"], + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"], + ) + + assert ( + program.id == TEST_DATA["PROGRAM"]["ID"] + ), "Program ID does not match expected value" + assert program.has_applied is False, "Program should be marked as not applied" + assert ( + program.state == TEST_DATA["STATUS"]["NOT_APPLIED"] + ), "Program state should be 'Not Applied'" + assert ( + program.last_application_status == TEST_DATA["STATUS"]["NO_APPLICATION"] + ), "Last application status should be 'Not submitted any application'" + + @pytest.mark.asyncio + async def test_get_program_by_id_not_found(self, mocker): + mocker.patch.object(ProgramORM, "get_all_by_program_id", return_value=None) + + program_service = ProgramService() + result = await program_service.get_program_by_id_service( + programid=TEST_DATA["PROGRAM"]["ID"], + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"], + ) + assert result == { + "message": TEST_DATA["ERROR"]["PROGRAM_NOT_FOUND"] + }, "Should return appropriate error message when program not found" + + @pytest.mark.asyncio + async def test_get_program_by_key_empty_result(self, mocker): + mocker.patch.object(ProgramORM, "get_all_program_by_keyword", return_value=[]) + program_service = ProgramService() + programs = await program_service.get_program_by_key_service( + keyword=TEST_DATA["TEST"]["NONEXISTENT"] + ) + assert programs == [], "Expected empty list when no programs match keyword" + + @pytest.mark.asyncio + async def test_get_program_by_key_success(self, mocker, program_mock): + mocker.patch.object( + ProgramORM, "get_all_program_by_keyword", return_value=[program_mock] + ) + program_service = ProgramService() + programs = await program_service.get_program_by_key_service( + keyword=TEST_DATA["TEST"]["KEYWORD"] + ) + assert len(programs) == 1, "Expected exactly one program matching keyword" + assert ( + programs[0].id == TEST_DATA["PROGRAM"]["ID"] + ), "Program ID does not match expected value" + assert ( + programs[0].name == TEST_DATA["PROGRAM"]["NAME"] + ), "Program name does not match expected value" + + @pytest.mark.asyncio + async def test_get_program_summary(self, mocker, program_mock): + mocker.patch.object( + ProgramORM, "get_program_summary", return_value=[program_mock] + ) + program_service = ProgramService() + summaries = await program_service.get_program_summary_service( + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"] + ) + assert len(summaries) == 1, "Expected exactly one program summary" + assert ( + summaries[0].program_name == TEST_DATA["PROGRAM"]["NAME"] + ), "Program name in summary does not match expected value" + + @pytest.mark.asyncio + async def test_get_application_details(self, mocker, program_mock): + mocker.patch.object( + ProgramORM, "get_application_details", return_value=[program_mock] + ) + program_service = ProgramService() + details = await program_service.get_application_details_service( + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"] + ) + assert len(details) == 1, "Expected exactly one application detail" + assert ( + details[0].application_id == TEST_DATA["PROGRAM"]["ID"] + ), "Application ID does not match expected value" + + @pytest.mark.asyncio + async def test_get_benefit_details(self, mocker, program_mock): + mocker.patch.object( + ProgramORM, "get_benefit_details", return_value=[program_mock] + ) + program_service = ProgramService() + benefits = await program_service.get_benefit_details_service( + partnerid=TEST_DATA["PROGRAM"]["PARTNER_ID"] + ) + assert len(benefits) == 1, "Expected exactly one benefit detail" + assert ( + benefits[0].entitlement_reference_number == TEST_DATA["PROGRAM"]["ID"] + ), "Entitlement reference number does not match expected value" From 7c9700dbb6a2f54e5a77e872acddadcc5baa9a6d Mon Sep 17 00:00:00 2001 From: Niranjan Kumar Date: Mon, 11 Nov 2024 10:22:49 +0530 Subject: [PATCH 2/4] test: implement unit tests for controller and service --- tests/test_document_file_service.py | 2 +- tests/test_partner_service.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_document_file_service.py b/tests/test_document_file_service.py index 5417a90..98ec330 100644 --- a/tests/test_document_file_service.py +++ b/tests/test_document_file_service.py @@ -237,7 +237,7 @@ async def test_upload_document_s3( result["message"] == TEST_CONSTANTS["SUCCESS_MESSAGE"] ), "S3 upload should return success message" mock_s3_storage.assert_called_once_with( - mock_file, f"test-pdf-{mock_document.id}", mock_backend_s3 + mock_file, str(f"test-pdf-{mock_document.id}"), mock_backend_s3 ) assert ( mock_session.add.call_count == 2 diff --git a/tests/test_partner_service.py b/tests/test_partner_service.py index c601902..dbabcd3 100644 --- a/tests/test_partner_service.py +++ b/tests/test_partner_service.py @@ -10,6 +10,7 @@ VALID_PARTNER_DATA = { "sub": "12345", + "user_id": "user123", "name": "John Middle Doe", "email": "john@example.com", "gender": "male", From caedf09571dce68aebcc06ecd3ea59022b6e8a26 Mon Sep 17 00:00:00 2001 From: Niranjan Kumar Date: Mon, 11 Nov 2024 11:16:29 +0530 Subject: [PATCH 3/4] update the slugify to python-slugify --- pyproject.toml | 2 +- src/openg2p_portal_api/services/document_file_service.py | 4 ++-- test-requirements.txt | 1 + tests/test_partner_service.py | 2 +- 4 files changed, 5 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index abbfc3f..6ac738d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,7 +19,7 @@ dependencies = [ "openg2p-fastapi-common", "openg2p-fastapi-auth", "boto3", - "slugify", + "python-slugify>=8.0.0", ] dynamic = ["version"] diff --git a/src/openg2p_portal_api/services/document_file_service.py b/src/openg2p_portal_api/services/document_file_service.py index 93dad9d..8aa0bde 100644 --- a/src/openg2p_portal_api/services/document_file_service.py +++ b/src/openg2p_portal_api/services/document_file_service.py @@ -7,7 +7,7 @@ from openg2p_fastapi_common.context import dbengine from openg2p_fastapi_common.errors.http_exceptions import BadRequestError from openg2p_fastapi_common.service import BaseService -from slugify import slugify +from slugify import slugify as python_slugify from sqlalchemy import select from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import async_sessionmaker @@ -102,7 +102,7 @@ async def upload_document(self, file, programid: int, file_tag: str): await session.refresh(new_file) # Generate slugified filename - slugified_filename = slugify(name) + slugified_filename = python_slugify(name) file_id = await get_file_id_by_slug(self) final_filename = f"{slugified_filename}-{file_id}" diff --git a/test-requirements.txt b/test-requirements.txt index 212e4cd..1a8c823 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -2,5 +2,6 @@ pytest-cov pytest-asyncio pytest pytest-mock +python-slugify>=8.0.0 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 diff --git a/tests/test_partner_service.py b/tests/test_partner_service.py index dbabcd3..8983d12 100644 --- a/tests/test_partner_service.py +++ b/tests/test_partner_service.py @@ -21,7 +21,7 @@ VALID_ID_TYPE_CONFIG = { "g2p_id_type": "national_id", "company_id": 1, - "token_map": "name:name email:email gender:gender birthdate:birthdate phone:phone", + "token_map": "name:name email:email gender:gender birthdate:birthdate phone:phone user_id:user_id", "date_format": "%Y/%m/%d", } From 55f4b3a3d51c2633063fbcbb40812097d7719bd1 Mon Sep 17 00:00:00 2001 From: Niranjan Kumar Date: Mon, 11 Nov 2024 11:35:01 +0530 Subject: [PATCH 4/4] test update of test_membership_service.py --- tests/test_membership_service.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/tests/test_membership_service.py b/tests/test_membership_service.py index db63c8f..38f0716 100644 --- a/tests/test_membership_service.py +++ b/tests/test_membership_service.py @@ -6,11 +6,6 @@ from sqlalchemy.exc import IntegrityError -@pytest.fixture -def membership_service(): - return MembershipService() - - @pytest.fixture def mock_session(): session = AsyncMock()