0207c175ba
This commit introduces a new endpoint for updating chores of any type, allowing conversions between personal and group chores while enforcing permission checks. Additionally, structured logging has been implemented through a new middleware, improving request tracing and logging details for better monitoring and debugging. These changes aim to enhance the functionality and maintainability of the chore management system.
438 lines
19 KiB
Python
438 lines
19 KiB
Python
import pytest
|
|
from decimal import Decimal
|
|
from datetime import datetime, timezone
|
|
from typing import AsyncGenerator, List
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import (
|
|
User,
|
|
Group,
|
|
Expense,
|
|
ExpenseSplit,
|
|
SettlementActivity,
|
|
ExpenseSplitStatusEnum,
|
|
ExpenseOverallStatusEnum,
|
|
SplitTypeEnum,
|
|
UserRoleEnum
|
|
)
|
|
from app.crud.settlement_activity import (
|
|
create_settlement_activity,
|
|
get_settlement_activity_by_id,
|
|
get_settlement_activities_for_split,
|
|
update_expense_split_status, # For direct testing if needed
|
|
update_expense_overall_status # For direct testing if needed
|
|
)
|
|
from app.schemas.settlement_activity import SettlementActivityCreate as SettlementActivityCreateSchema
|
|
from app.core.exceptions import OverpaymentError
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_user1(db_session: AsyncSession) -> User:
|
|
user = User(email="user1@example.com", name="Test User 1", hashed_password="password1")
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.fixture
|
|
async def test_user2(db_session: AsyncSession) -> User:
|
|
user = User(email="user2@example.com", name="Test User 2", hashed_password="password2")
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.fixture
|
|
async def test_group(db_session: AsyncSession, test_user1: User) -> Group:
|
|
group = Group(name="Test Group", created_by_id=test_user1.id)
|
|
db_session.add(group)
|
|
await db_session.commit()
|
|
# Add user1 as owner and user2 as member (can be done in specific tests if needed)
|
|
await db_session.refresh(group)
|
|
return group
|
|
|
|
@pytest.fixture
|
|
async def test_expense(db_session: AsyncSession, test_user1: User, test_group: Group) -> Expense:
|
|
expense = Expense(
|
|
description="Test Expense for Settlement",
|
|
total_amount=Decimal("20.00"),
|
|
currency="USD",
|
|
expense_date=datetime.now(timezone.utc),
|
|
split_type=SplitTypeEnum.EQUAL,
|
|
group_id=test_group.id,
|
|
paid_by_user_id=test_user1.id,
|
|
created_by_user_id=test_user1.id,
|
|
overall_settlement_status=ExpenseOverallStatusEnum.unpaid # Initial status
|
|
)
|
|
db_session.add(expense)
|
|
await db_session.commit()
|
|
await db_session.refresh(expense)
|
|
return expense
|
|
|
|
@pytest.fixture
|
|
async def test_expense_split_user2_owes(db_session: AsyncSession, test_expense: Expense, test_user2: User) -> ExpenseSplit:
|
|
# User2 owes 10.00 to User1 (who paid the expense)
|
|
split = ExpenseSplit(
|
|
expense_id=test_expense.id,
|
|
user_id=test_user2.id,
|
|
owed_amount=Decimal("10.00"),
|
|
status=ExpenseSplitStatusEnum.unpaid # Initial status
|
|
)
|
|
db_session.add(split)
|
|
await db_session.commit()
|
|
await db_session.refresh(split)
|
|
return split
|
|
|
|
@pytest.fixture
|
|
async def test_expense_split_user1_owes_self_for_completeness(db_session: AsyncSession, test_expense: Expense, test_user1: User) -> ExpenseSplit:
|
|
# User1's own share (owes 10.00 to self, effectively settled)
|
|
# This is often how splits are represented, even for the payer
|
|
split = ExpenseSplit(
|
|
expense_id=test_expense.id,
|
|
user_id=test_user1.id,
|
|
owed_amount=Decimal("10.00"), # User1's share of the 20.00 expense
|
|
status=ExpenseSplitStatusEnum.unpaid # Initial status, though payer's own share might be considered paid by some logic
|
|
)
|
|
db_session.add(split)
|
|
await db_session.commit()
|
|
await db_session.refresh(split)
|
|
return split
|
|
|
|
|
|
# --- Tests for create_settlement_activity ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_full_payment(
|
|
db_session: AsyncSession,
|
|
test_user1: User, # Creator of activity, Payer of expense
|
|
test_user2: User, # Payer of this settlement activity (settling their debt)
|
|
test_expense: Expense,
|
|
test_expense_split_user2_owes: ExpenseSplit,
|
|
test_expense_split_user1_owes_self_for_completeness: ExpenseSplit # User1's own share
|
|
):
|
|
# Scenario: User2 fully pays their 10.00 share.
|
|
# User1's share is also part of the expense. Let's assume it's 'paid' by default or handled separately.
|
|
# For this test, we focus on User2's split.
|
|
# To make overall expense paid, User1's split also needs to be considered paid.
|
|
# We can manually update User1's split status to paid for this test case.
|
|
test_expense_split_user1_owes_self_for_completeness.status = ExpenseSplitStatusEnum.paid
|
|
test_expense_split_user1_owes_self_for_completeness.paid_at = datetime.now(timezone.utc)
|
|
db_session.add(test_expense_split_user1_owes_self_for_completeness)
|
|
await db_session.commit()
|
|
await db_session.refresh(test_expense_split_user1_owes_self_for_completeness)
|
|
await db_session.refresh(test_expense) # Refresh expense to reflect split status change
|
|
|
|
|
|
activity_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id, # User2 is paying their share
|
|
amount_paid=Decimal("10.00")
|
|
)
|
|
|
|
created_activity = await create_settlement_activity(
|
|
db=db_session,
|
|
settlement_activity_in=activity_data,
|
|
current_user_id=test_user2.id # User2 is recording their own payment
|
|
)
|
|
|
|
assert created_activity is not None
|
|
assert created_activity.expense_split_id == test_expense_split_user2_owes.id
|
|
assert created_activity.paid_by_user_id == test_user2.id
|
|
assert created_activity.amount_paid == Decimal("10.00")
|
|
assert created_activity.created_by_user_id == test_user2.id
|
|
|
|
await db_session.refresh(test_expense_split_user2_owes)
|
|
await db_session.refresh(test_expense) # Refresh to get updated overall_status
|
|
|
|
assert test_expense_split_user2_owes.status == ExpenseSplitStatusEnum.paid
|
|
assert test_expense_split_user2_owes.paid_at is not None
|
|
|
|
# Check parent expense status
|
|
# This depends on all splits being paid for the expense to be fully paid.
|
|
assert test_expense.overall_settlement_status == ExpenseOverallStatusEnum.paid
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_partial_payment(
|
|
db_session: AsyncSession,
|
|
test_user1: User, # Creator of activity
|
|
test_user2: User, # Payer of this settlement activity
|
|
test_expense: Expense,
|
|
test_expense_split_user2_owes: ExpenseSplit
|
|
):
|
|
activity_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("5.00")
|
|
)
|
|
|
|
created_activity = await create_settlement_activity(
|
|
db=db_session,
|
|
settlement_activity_in=activity_data,
|
|
current_user_id=test_user2.id # User2 records their payment
|
|
)
|
|
|
|
assert created_activity is not None
|
|
assert created_activity.amount_paid == Decimal("5.00")
|
|
|
|
await db_session.refresh(test_expense_split_user2_owes)
|
|
await db_session.refresh(test_expense)
|
|
|
|
assert test_expense_split_user2_owes.status == ExpenseSplitStatusEnum.partially_paid
|
|
assert test_expense_split_user2_owes.paid_at is None
|
|
assert test_expense.overall_settlement_status == ExpenseOverallStatusEnum.partially_paid # Assuming other splits are unpaid or partially paid
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_multiple_payments_to_full(
|
|
db_session: AsyncSession,
|
|
test_user1: User,
|
|
test_user2: User,
|
|
test_expense: Expense,
|
|
test_expense_split_user2_owes: ExpenseSplit,
|
|
test_expense_split_user1_owes_self_for_completeness: ExpenseSplit # User1's own share
|
|
):
|
|
# Assume user1's share is already 'paid' for overall expense status testing
|
|
test_expense_split_user1_owes_self_for_completeness.status = ExpenseSplitStatusEnum.paid
|
|
test_expense_split_user1_owes_self_for_completeness.paid_at = datetime.now(timezone.utc)
|
|
db_session.add(test_expense_split_user1_owes_self_for_completeness)
|
|
await db_session.commit()
|
|
|
|
# First partial payment
|
|
activity_data1 = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("3.00")
|
|
)
|
|
await create_settlement_activity(db=db_session, settlement_activity_in=activity_data1, current_user_id=test_user2.id)
|
|
|
|
await db_session.refresh(test_expense_split_user2_owes)
|
|
await db_session.refresh(test_expense)
|
|
assert test_expense_split_user2_owes.status == ExpenseSplitStatusEnum.partially_paid
|
|
assert test_expense.overall_settlement_status == ExpenseOverallStatusEnum.partially_paid
|
|
|
|
# Second payment completing the amount
|
|
activity_data2 = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("7.00") # 3.00 + 7.00 = 10.00
|
|
)
|
|
await create_settlement_activity(db=db_session, settlement_activity_in=activity_data2, current_user_id=test_user2.id)
|
|
|
|
await db_session.refresh(test_expense_split_user2_owes)
|
|
await db_session.refresh(test_expense)
|
|
assert test_expense_split_user2_owes.status == ExpenseSplitStatusEnum.paid
|
|
assert test_expense_split_user2_owes.paid_at is not None
|
|
assert test_expense.overall_settlement_status == ExpenseOverallStatusEnum.paid
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_invalid_split_id(
|
|
db_session: AsyncSession, test_user1: User
|
|
):
|
|
activity_data = SettlementActivityCreateSchema(
|
|
expense_split_id=99999, # Non-existent
|
|
paid_by_user_id=test_user1.id,
|
|
amount_paid=Decimal("10.00")
|
|
)
|
|
# The CRUD function returns None for not found related objects
|
|
result = await create_settlement_activity(db=db_session, settlement_activity_in=activity_data, current_user_id=test_user1.id)
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_invalid_paid_by_user_id(
|
|
db_session: AsyncSession, test_user1: User, test_expense_split_user2_owes: ExpenseSplit
|
|
):
|
|
activity_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=99999, # Non-existent
|
|
amount_paid=Decimal("10.00")
|
|
)
|
|
result = await create_settlement_activity(db=db_session, settlement_activity_in=activity_data, current_user_id=test_user1.id)
|
|
assert result is None
|
|
|
|
|
|
# --- Tests for get_settlement_activity_by_id ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_settlement_activity_by_id_found(
|
|
db_session: AsyncSession, test_user2: User, test_expense_split_user2_owes: ExpenseSplit
|
|
):
|
|
activity_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("5.00")
|
|
)
|
|
created = await create_settlement_activity(db=db_session, settlement_activity_in=activity_data, current_user_id=test_user2.id)
|
|
assert created is not None
|
|
|
|
fetched = await get_settlement_activity_by_id(db=db_session, settlement_activity_id=created.id)
|
|
assert fetched is not None
|
|
assert fetched.id == created.id
|
|
assert fetched.amount_paid == Decimal("5.00")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_settlement_activity_by_id_not_found(db_session: AsyncSession):
|
|
fetched = await get_settlement_activity_by_id(db=db_session, settlement_activity_id=99999)
|
|
assert fetched is None
|
|
|
|
|
|
# --- Tests for get_settlement_activities_for_split ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_settlement_activities_for_split_multiple_found(
|
|
db_session: AsyncSession, test_user2: User, test_expense_split_user2_owes: ExpenseSplit
|
|
):
|
|
act1_data = SettlementActivityCreateSchema(expense_split_id=test_expense_split_user2_owes.id, paid_by_user_id=test_user2.id, amount_paid=Decimal("2.00"))
|
|
act2_data = SettlementActivityCreateSchema(expense_split_id=test_expense_split_user2_owes.id, paid_by_user_id=test_user2.id, amount_paid=Decimal("3.00"))
|
|
|
|
await create_settlement_activity(db=db_session, settlement_activity_in=act1_data, current_user_id=test_user2.id)
|
|
await create_settlement_activity(db=db_session, settlement_activity_in=act2_data, current_user_id=test_user2.id)
|
|
|
|
activities: List[SettlementActivity] = await get_settlement_activities_for_split(db=db_session, expense_split_id=test_expense_split_user2_owes.id)
|
|
assert len(activities) == 2
|
|
amounts = sorted([act.amount_paid for act in activities])
|
|
assert amounts == [Decimal("2.00"), Decimal("3.00")]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_settlement_activities_for_split_none_found(
|
|
db_session: AsyncSession, test_expense_split_user2_owes: ExpenseSplit # A split with no activities
|
|
):
|
|
activities: List[SettlementActivity] = await get_settlement_activities_for_split(db=db_session, expense_split_id=test_expense_split_user2_owes.id)
|
|
assert len(activities) == 0
|
|
|
|
# Note: Direct tests for helper functions update_expense_split_status and update_expense_overall_status
|
|
# could be added if complex logic within them isn't fully covered by create_settlement_activity tests.
|
|
# However, their effects are validated through the main CRUD function here.
|
|
# For example, to test update_expense_split_status directly:
|
|
# 1. Create an ExpenseSplit.
|
|
# 2. Create one or more SettlementActivity instances directly in the DB session for that split.
|
|
# 3. Call await update_expense_split_status(db_session, expense_split_id=split.id).
|
|
# 4. Assert the split.status and split.paid_at are as expected.
|
|
# Similar for update_expense_overall_status by setting up multiple splits.
|
|
# For now, relying on indirect testing via create_settlement_activity.
|
|
|
|
# More tests can be added for edge cases, such as:
|
|
# - Overpayment (current logic in update_expense_split_status treats >= owed_amount as 'paid').
|
|
# - Different users creating the activity vs. paying for it (permission aspects, though that's more for API tests).
|
|
# - Interactions with different expense split types if that affects status updates.
|
|
# - Ensuring `overall_settlement_status` correctly reflects if one split is paid, another is unpaid, etc.
|
|
# (e.g. test_expense_split_user1_owes_self_for_completeness is set to unpaid initially).
|
|
# A test case where one split becomes 'paid' but another remains 'unpaid' should result in 'partially_paid' for the expense.
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_overall_status_becomes_partially_paid(
|
|
db_session: AsyncSession,
|
|
test_user1: User,
|
|
test_user2: User,
|
|
test_expense: Expense, # Overall status is initially unpaid
|
|
test_expense_split_user2_owes: ExpenseSplit, # User2's split, initially unpaid
|
|
test_expense_split_user1_owes_self_for_completeness: ExpenseSplit # User1's split, also initially unpaid
|
|
):
|
|
# Sanity check: both splits and expense are unpaid initially
|
|
assert test_expense_split_user2_owes.status == ExpenseSplitStatusEnum.unpaid
|
|
assert test_expense_split_user1_owes_self_for_completeness.status == ExpenseSplitStatusEnum.unpaid
|
|
assert test_expense.overall_settlement_status == ExpenseOverallStatusEnum.unpaid
|
|
|
|
# User2 fully pays their 10.00 share.
|
|
activity_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id, # User2 is paying their share
|
|
amount_paid=Decimal("10.00")
|
|
)
|
|
|
|
await create_settlement_activity(
|
|
db=db_session,
|
|
settlement_activity_in=activity_data,
|
|
current_user_id=test_user2.id # User2 is recording their own payment
|
|
)
|
|
|
|
await db_session.refresh(test_expense_split_user2_owes)
|
|
await db_session.refresh(test_expense_split_user1_owes_self_for_completeness) # Ensure its status is current
|
|
await db_session.refresh(test_expense)
|
|
|
|
assert test_expense_split_user2_owes.status == ExpenseSplitStatusEnum.paid
|
|
assert test_expense_split_user1_owes_self_for_completeness.status == ExpenseSplitStatusEnum.unpaid # User1's split is still unpaid
|
|
|
|
# Since one split is paid and the other is unpaid, the overall expense status should be partially_paid
|
|
assert test_expense.overall_settlement_status == ExpenseOverallStatusEnum.partially_paid
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_overpayment_protection(
|
|
db_session: AsyncSession, test_user2: User, test_expense_split_user2_owes: ExpenseSplit
|
|
):
|
|
"""Test that settlement activities prevent overpayment beyond owed amount."""
|
|
# Test split owes 10.00, attempt to pay 15.00 directly
|
|
activity_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("15.00") # More than the 10.00 owed
|
|
)
|
|
|
|
# Should raise OverpaymentError
|
|
with pytest.raises(OverpaymentError):
|
|
await create_settlement_activity(
|
|
db=db_session,
|
|
settlement_activity_in=activity_data,
|
|
current_user_id=test_user2.id
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_settlement_activity_overpayment_protection_partial(
|
|
db_session: AsyncSession, test_user2: User, test_expense_split_user2_owes: ExpenseSplit
|
|
):
|
|
"""Test overpayment protection with multiple payments."""
|
|
# First payment of 7.00
|
|
activity1_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("7.00")
|
|
)
|
|
|
|
activity1 = await create_settlement_activity(
|
|
db=db_session,
|
|
settlement_activity_in=activity1_data,
|
|
current_user_id=test_user2.id
|
|
)
|
|
assert activity1 is not None
|
|
|
|
# Second payment of 5.00 should be rejected (7 + 5 = 12 > 10 owed)
|
|
activity2_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("5.00")
|
|
)
|
|
|
|
with pytest.raises(OverpaymentError):
|
|
await create_settlement_activity(
|
|
db=db_session,
|
|
settlement_activity_in=activity2_data,
|
|
current_user_id=test_user2.id
|
|
)
|
|
|
|
# But a payment of 3.00 should work (7 + 3 = 10 = exact amount owed)
|
|
activity3_data = SettlementActivityCreateSchema(
|
|
expense_split_id=test_expense_split_user2_owes.id,
|
|
paid_by_user_id=test_user2.id,
|
|
amount_paid=Decimal("3.00")
|
|
)
|
|
|
|
activity3 = await create_settlement_activity(
|
|
db=db_session,
|
|
settlement_activity_in=activity3_data,
|
|
current_user_id=test_user2.id
|
|
)
|
|
assert activity3 is not None
|
|
|
|
# Example of a placeholder for db_session fixture if not provided by conftest.py
|
|
# @pytest.fixture
|
|
# async def db_session() -> AsyncGenerator[AsyncSession, None]:
|
|
# # This needs to be implemented based on your test database setup
|
|
# # e.g., using a test-specific database and creating a new session per test
|
|
# # from app.database import SessionLocal # Assuming SessionLocal is your session factory
|
|
# # async with SessionLocal() as session:
|
|
# # async with session.begin(): # Start a transaction
|
|
# # yield session
|
|
# # # Transaction will be rolled back here after the test
|
|
# pass # Replace with actual implementation if needed
|