373 lines
15 KiB
Python
373 lines
15 KiB
Python
import pytest
|
|
from fastapi import status
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from typing import Callable, Dict, Any
|
|
|
|
from app.models import User as UserModel, Group as GroupModel, List as ListModel
|
|
from app.schemas.expense import ExpenseCreate
|
|
from app.core.config import settings
|
|
|
|
# Helper to create a URL for an endpoint
|
|
API_V1_STR = settings.API_V1_STR
|
|
|
|
def expense_url(endpoint: str = "") -> str:
|
|
return f"{API_V1_STR}/financials/expenses{endpoint}"
|
|
|
|
def settlement_url(endpoint: str = "") -> str:
|
|
return f"{API_V1_STR}/financials/settlements{endpoint}"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_new_expense_success_list_context(
|
|
client: AsyncClient,
|
|
db_session: AsyncSession, # Assuming a fixture for db session
|
|
normal_user_token_headers: Dict[str, str], # Assuming a fixture for user auth
|
|
test_user: UserModel, # Assuming a fixture for a test user
|
|
test_list_user_is_member: ListModel, # Assuming a fixture for a list user is member of
|
|
) -> None:
|
|
"""
|
|
Test successful creation of a new expense linked to a list.
|
|
"""
|
|
expense_data = ExpenseCreate(
|
|
description="Test Expense for List",
|
|
amount=100.00,
|
|
currency="USD",
|
|
paid_by_user_id=test_user.id,
|
|
list_id=test_list_user_is_member.id,
|
|
group_id=None, # group_id should be derived from list if list is in a group
|
|
# category_id: Optional[int] = None # Assuming category is optional
|
|
# expense_date: Optional[date] = None # Assuming date is optional
|
|
# splits: Optional[List[SplitCreate]] = [] # Assuming splits are optional for now
|
|
)
|
|
|
|
response = await client.post(
|
|
expense_url(),
|
|
headers=normal_user_token_headers,
|
|
json=expense_data.model_dump(exclude_unset=True)
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
content = response.json()
|
|
assert content["description"] == expense_data.description
|
|
assert content["amount"] == expense_data.amount
|
|
assert content["currency"] == expense_data.currency
|
|
assert content["paid_by_user_id"] == test_user.id
|
|
assert content["list_id"] == test_list_user_is_member.id
|
|
# If test_list_user_is_member has a group_id, it should be set in the response
|
|
if test_list_user_is_member.group_id:
|
|
assert content["group_id"] == test_list_user_is_member.group_id
|
|
else:
|
|
assert content["group_id"] is None
|
|
assert "id" in content
|
|
assert "created_at" in content
|
|
assert "updated_at" in content
|
|
assert "version" in content
|
|
assert content["version"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_new_expense_success_group_context(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
test_user: UserModel,
|
|
test_group_user_is_member: GroupModel, # Assuming a fixture for a group user is member of
|
|
) -> None:
|
|
"""
|
|
Test successful creation of a new expense linked directly to a group.
|
|
"""
|
|
expense_data = ExpenseCreate(
|
|
description="Test Expense for Group",
|
|
amount=50.00,
|
|
currency="EUR",
|
|
paid_by_user_id=test_user.id,
|
|
group_id=test_group_user_is_member.id,
|
|
list_id=None,
|
|
)
|
|
|
|
response = await client.post(
|
|
expense_url(),
|
|
headers=normal_user_token_headers,
|
|
json=expense_data.model_dump(exclude_unset=True)
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
content = response.json()
|
|
assert content["description"] == expense_data.description
|
|
assert content["paid_by_user_id"] == test_user.id
|
|
assert content["group_id"] == test_group_user_is_member.id
|
|
assert content["list_id"] is None
|
|
assert content["version"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_new_expense_fail_no_list_or_group(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
test_user: UserModel,
|
|
) -> None:
|
|
"""
|
|
Test expense creation fails if neither list_id nor group_id is provided.
|
|
"""
|
|
expense_data = ExpenseCreate(
|
|
description="Test Invalid Expense",
|
|
amount=10.00,
|
|
currency="USD",
|
|
paid_by_user_id=test_user.id,
|
|
list_id=None,
|
|
group_id=None,
|
|
)
|
|
|
|
response = await client.post(
|
|
expense_url(),
|
|
headers=normal_user_token_headers,
|
|
json=expense_data.model_dump(exclude_unset=True)
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
|
content = response.json()
|
|
assert "Expense must be linked to a list_id or group_id" in content["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_new_expense_fail_paid_by_other_not_owner(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str], # User is member, not owner
|
|
test_user: UserModel, # This is the current_user (member)
|
|
test_group_user_is_member: GroupModel, # Group the current_user is a member of
|
|
another_user_in_group: UserModel, # Another user in the same group
|
|
# Ensure test_user is NOT an owner of test_group_user_is_member for this test
|
|
) -> None:
|
|
"""
|
|
Test creation fails if paid_by_user_id is another user, and current_user is not a group owner.
|
|
Assumes normal_user_token_headers belongs to a user who is a member but not an owner of test_group_user_is_member.
|
|
"""
|
|
expense_data = ExpenseCreate(
|
|
description="Expense paid by other",
|
|
amount=75.00,
|
|
currency="GBP",
|
|
paid_by_user_id=another_user_in_group.id, # Paid by someone else
|
|
group_id=test_group_user_is_member.id,
|
|
list_id=None,
|
|
)
|
|
|
|
response = await client.post(
|
|
expense_url(),
|
|
headers=normal_user_token_headers, # Current user is a member, not owner
|
|
json=expense_data.model_dump(exclude_unset=True)
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
content = response.json()
|
|
assert "Only group owners can create expenses paid by others" in content["detail"]
|
|
|
|
# --- Add tests for other endpoints below ---
|
|
# GET /expenses/{expense_id}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_success(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
test_user: UserModel,
|
|
# Assume an existing expense created by test_user or in a group/list they have access to
|
|
# This would typically be created by another test or a fixture
|
|
created_expense: ExpensePublic, # Assuming a fixture that provides a created expense
|
|
) -> None:
|
|
"""
|
|
Test successfully retrieving an existing expense.
|
|
User has access either by being the payer, or via list/group membership.
|
|
"""
|
|
response = await client.get(
|
|
expense_url(f"/{created_expense.id}"),
|
|
headers=normal_user_token_headers
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
content = response.json()
|
|
assert content["id"] == created_expense.id
|
|
assert content["description"] == created_expense.description
|
|
assert content["amount"] == created_expense.amount
|
|
assert content["paid_by_user_id"] == created_expense.paid_by_user_id
|
|
if created_expense.list_id:
|
|
assert content["list_id"] == created_expense.list_id
|
|
if created_expense.group_id:
|
|
assert content["group_id"] == created_expense.group_id
|
|
|
|
# TODO: Add more tests for get_expense:
|
|
# - expense not found -> 404
|
|
# - user has no access (not payer, not in list, not in group if applicable) -> 403
|
|
# - expense in list, user has list access
|
|
# - expense in group, user has group access
|
|
# - expense personal (no list, no group), user is payer
|
|
# - expense personal (no list, no group), user is NOT payer -> 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_not_found(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
) -> None:
|
|
"""
|
|
Test retrieving a non-existent expense results in 404.
|
|
"""
|
|
non_existent_expense_id = 9999999
|
|
response = await client.get(
|
|
expense_url(f"/{non_existent_expense_id}"),
|
|
headers=normal_user_token_headers
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
content = response.json()
|
|
assert "not found" in content["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_forbidden_personal_expense_other_user(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str], # Belongs to test_user
|
|
# Fixture for an expense paid by another_user, not linked to any list/group test_user has access to
|
|
personal_expense_of_another_user: ExpensePublic
|
|
) -> None:
|
|
"""
|
|
Test retrieving a personal expense of another user (no shared list/group) results in 403.
|
|
"""
|
|
response = await client.get(
|
|
expense_url(f"/{personal_expense_of_another_user.id}"),
|
|
headers=normal_user_token_headers # Current user querying
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
content = response.json()
|
|
assert "Not authorized to view this expense" in content["detail"]
|
|
|
|
# GET /lists/{list_id}/expenses
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_list_expenses_success(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
test_user: UserModel,
|
|
test_list_user_is_member: ListModel, # List the user is a member of
|
|
# Assume some expenses have been created for this list by a fixture or previous tests
|
|
) -> None:
|
|
"""
|
|
Test successfully listing expenses for a list the user has access to.
|
|
"""
|
|
response = await client.get(
|
|
f"{API_V1_STR}/financials/lists/{test_list_user_is_member.id}/expenses",
|
|
headers=normal_user_token_headers
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
content = response.json()
|
|
assert isinstance(content, list)
|
|
for expense_item in content: # Renamed from expense to avoid conflict if a fixture is named expense
|
|
assert expense_item["list_id"] == test_list_user_is_member.id
|
|
|
|
# TODO: Add more tests for list_list_expenses:
|
|
# - list not found -> 404 (ListNotFoundError from check_list_access_for_financials)
|
|
# - user has no access to list -> 403 (ListPermissionError from check_list_access_for_financials)
|
|
# - list exists but has no expenses -> empty list, 200 OK
|
|
# - test pagination (skip, limit)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_list_expenses_list_not_found(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
) -> None:
|
|
"""
|
|
Test listing expenses for a non-existent list results in 404 (or appropriate error from permission check).
|
|
The check_list_access_for_financials raises ListNotFoundError, which might be caught and raised as 404.
|
|
The endpoint itself also has a get for ListModel, which would 404 first if permission check passed (not possible here).
|
|
Based on financials.py, ListNotFoundError is raised by check_list_access_for_financials.
|
|
This should translate to a 404 or a 403 if ListPermissionError wraps it with an action.
|
|
The current ListPermissionError in check_list_access_for_financials re-raises ListNotFoundError if that's the cause.
|
|
ListNotFoundError is a custom exception often mapped to 404.
|
|
Let's assume ListNotFoundError results in a 404 response from an exception handler.
|
|
"""
|
|
non_existent_list_id = 99999
|
|
response = await client.get(
|
|
f"{API_V1_STR}/financials/lists/{non_existent_list_id}/expenses",
|
|
headers=normal_user_token_headers
|
|
)
|
|
# The ListNotFoundError is raised by the check_list_access_for_financials helper,
|
|
# which is then re-raised. FastAPI default exception handlers or custom ones
|
|
# would convert this to an HTTP response. Typically NotFoundError -> 404.
|
|
# If ListPermissionError catches it and re-raises it specifically, it might be 403.
|
|
# From the code: `except ListNotFoundError: raise` means it propagates.
|
|
# Let's assume a global handler for NotFoundError derived exceptions leads to 404.
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
# The actual detail might vary based on how ListNotFoundError is handled by FastAPI
|
|
# For now, we check the status code. If financials.py maps it differently, this will need adjustment.
|
|
# Based on `raise ListNotFoundError(expense_in.list_id)` in create_new_expense, and if that leads to 400,
|
|
# this might be inconsistent. However, `check_list_access_for_financials` just re-raises ListNotFoundError.
|
|
# Let's stick to expecting 404 for a direct not found error from a path parameter.
|
|
content = response.json()
|
|
assert "list not found" in content["detail"].lower() # Common detail for not found errors
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_list_expenses_no_access(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str], # User who will attempt access
|
|
test_list_user_not_member: ListModel, # A list current user is NOT a member of
|
|
) -> None:
|
|
"""
|
|
Test listing expenses for a list the user does not have access to (403 Forbidden).
|
|
"""
|
|
response = await client.get(
|
|
f"{API_V1_STR}/financials/lists/{test_list_user_not_member.id}/expenses",
|
|
headers=normal_user_token_headers
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
content = response.json()
|
|
assert f"User does not have permission to access financial data for list {test_list_user_not_member.id}" in content["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_list_expenses_empty(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
test_list_user_is_member_no_expenses: ListModel, # List user is member of, but has no expenses
|
|
) -> None:
|
|
"""
|
|
Test listing expenses for an accessible list that has no expenses (empty list, 200 OK).
|
|
"""
|
|
response = await client.get(
|
|
f"{API_V1_STR}/financials/lists/{test_list_user_is_member_no_expenses.id}/expenses",
|
|
headers=normal_user_token_headers
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
content = response.json()
|
|
assert isinstance(content, list)
|
|
assert len(content) == 0
|
|
|
|
# GET /groups/{group_id}/expenses
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_group_expenses_success(
|
|
client: AsyncClient,
|
|
normal_user_token_headers: Dict[str, str],
|
|
test_user: UserModel,
|
|
test_group_user_is_member: GroupModel, # Group the user is a member of
|
|
# Assume some expenses have been created for this group by a fixture or previous tests
|
|
) -> None:
|
|
"""
|
|
Test successfully listing expenses for a group the user has access to.
|
|
"""
|
|
response = await client.get(
|
|
f"{API_V1_STR}/financials/groups/{test_group_user_is_member.id}/expenses",
|
|
headers=normal_user_token_headers
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
content = response.json()
|
|
assert isinstance(content, list)
|
|
# Further assertions can be made here, e.g., checking if all expenses belong to the group
|
|
for expense_item in content:
|
|
assert expense_item["group_id"] == test_group_user_is_member.id
|
|
# Expenses in a group might also have a list_id if they were added via a list belonging to that group
|
|
|
|
# TODO: Add more tests for list_group_expenses:
|
|
# - group not found -> 404 (GroupNotFoundError from check_group_membership)
|
|
# - user has no access to group (not a member) -> 403 (GroupMembershipError from check_group_membership)
|
|
# - group exists but has no expenses -> empty list, 200 OK
|
|
# - test pagination (skip, limit)
|
|
|
|
# PUT /expenses/{expense_id}
|
|
# DELETE /expenses/{expense_id}
|
|
|
|
# GET /settlements/{settlement_id}
|
|
# POST /settlements
|
|
# GET /groups/{group_id}/settlements
|
|
# PUT /settlements/{settlement_id}
|
|
# DELETE /settlements/{settlement_id}
|
|
|
|
pytest.skip("Still implementing other tests", allow_module_level=True) |