Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

tests: implement unit tests for controllers and services #29

Merged
merged 4 commits into from
Dec 9, 2024
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
tests: add unit tests for controllers and services
Q-Niranjan committed Nov 8, 2024

Verified

This commit was signed with the committer’s verified signature.
emma-sg Emma Segal-Grossman
commit 9bbe398795e2ad3097a03d94129c5f3bcfd78793
3 changes: 3 additions & 0 deletions test-requirements.txt
Original file line number Diff line number Diff line change
@@ -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
312 changes: 312 additions & 0 deletions tests/test_auth_controller.py
Original file line number Diff line number Diff line change
@@ -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 = "[email protected]"
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 == "[email protected]"
), "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="[email protected]"
)

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"
85 changes: 85 additions & 0 deletions tests/test_discovery_controller.py
Original file line number Diff line number Diff line change
@@ -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"
132 changes: 132 additions & 0 deletions tests/test_document_file_controller.py
Original file line number Diff line number Diff line change
@@ -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."
297 changes: 297 additions & 0 deletions tests/test_document_file_service.py
Original file line number Diff line number Diff line change
@@ -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"
33 changes: 33 additions & 0 deletions tests/test_exception.py
Original file line number Diff line number Diff line change
@@ -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}'"
179 changes: 179 additions & 0 deletions tests/test_form_controller.py
Original file line number Diff line number Diff line change
@@ -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"
209 changes: 209 additions & 0 deletions tests/test_form_service.py
Original file line number Diff line number Diff line change
@@ -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"
88 changes: 88 additions & 0 deletions tests/test_membership_service.py
Original file line number Diff line number Diff line change
@@ -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"
155 changes: 155 additions & 0 deletions tests/test_oauth_controller.py
Original file line number Diff line number Diff line change
@@ -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)
261 changes: 261 additions & 0 deletions tests/test_partner_service.py
Original file line number Diff line number Diff line change
@@ -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"
174 changes: 174 additions & 0 deletions tests/test_program_controller.py
Original file line number Diff line number Diff line change
@@ -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"
263 changes: 260 additions & 3 deletions tests/test_program_service.py
Original file line number Diff line number Diff line change
@@ -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"