import pytest import csv import json import io from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.main import app from app.core.database import get_db, Base from app.models.trip import Trip from app.models.participant import Participant from app.models.expense import Expense, ExpenseSplit # Test database setup SQLALCHEMY_DATABASE_URL = "sqlite:///./test_export.db" engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}) TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) def override_get_db(): try: db = TestingSessionLocal() yield db finally: db.close() app.dependency_overrides[get_db] = override_get_db client = TestClient(app) def setup_database(): """Setup test database""" Base.metadata.create_all(bind=engine) def cleanup_database(): """Clean up test database""" Base.metadata.drop_all(bind=engine) def create_test_trip(): """Helper function to create a test trip""" trip_data = { "name": "Test Trip Paris", "creator_name": "John Doe", "currency_code": "USD" } response = client.post("/api/trips/", json=trip_data) return response.json() def create_test_participant(trip_id, name="Test Participant"): """Helper function to create a test participant""" participant_data = {"name": name} response = client.post(f"/api/participants/{trip_id}/participants", json=participant_data) return response.json() def create_test_expense(trip_id, paid_by_id, splits): """Helper function to create a test expense""" expense_data = { "description": "Test Expense", "amount": 100.00, "currency_code": "USD", "exchange_rate": 1.0, "category": "Food & Dining", "paid_by_id": paid_by_id, "splits": splits } response = client.post(f"/api/expenses/{trip_id}/expenses", json=expense_data) return response.json() class TestExportAPI: """Test suite for Export API endpoints""" @pytest.fixture(autouse=True) def setup(self): """Setup for each test""" setup_database() yield cleanup_database() def test_export_expenses_csv(self): """Test exporting expenses as CSV""" # Create test data trip = create_test_trip() participant1 = create_test_participant(trip["id"], "Alice") participant2 = create_test_participant(trip["id"], "Bob") # Create expenses expense1 = create_test_expense( trip["id"], participant1["id"], [ {"participant_id": participant1["id"], "percentage": 50.0}, {"participant_id": participant2["id"], "percentage": 50.0} ] ) expense2 = create_test_expense( trip["id"], participant2["id"], [ {"participant_id": participant1["id"], "percentage": 100.0} ] ) # Export expenses response = client.get(f"/api/export/{trip['id']}/expenses/csv") assert response.status_code == 200 assert response.headers["content-type"] == "text/csv; charset=utf-8" # Check content disposition header content_disposition = response.headers["content-disposition"] assert "attachment" in content_disposition assert f"expenses_trip_{trip['id']}_{trip['share_code']}.csv" in content_disposition # Parse CSV content csv_content = response.content.decode('utf-8') csv_reader = csv.reader(io.StringIO(csv_content)) rows = list(csv_reader) # Check headers expected_headers = [ "ID", "Description", "Amount", "Currency", "Amount in Trip Currency", "Category", "Paid By", "Date", "Created At" ] assert rows[0] == expected_headers # Check data rows (should have 2 expenses) assert len(rows) == 3 # Header + 2 expense rows # Find expense rows by description expense_rows = {row[1]: row for row in rows[1:]} # Use description as key assert "Test Expense" in expense_rows def test_export_participants_csv(self): """Test exporting participants as CSV""" # Create test data trip = create_test_trip() participant1 = create_test_participant(trip["id"], "Alice") participant2 = create_test_participant(trip["id"], "Bob") # Export participants response = client.get(f"/api/export/{trip['id']}/participants/csv") assert response.status_code == 200 assert response.headers["content-type"] == "text/csv; charset=utf-8" # Parse CSV content csv_content = response.content.decode('utf-8') csv_reader = csv.reader(io.StringIO(csv_content)) rows = list(csv_reader) # Check headers expected_headers = ["ID", "Name", "Is Creator", "Joined At"] assert rows[0] == expected_headers # Check data rows assert len(rows) == 3 # Header + 2 participants # Find participant rows by name participant_rows = {row[1]: row for row in rows[1:]} # Use name as key assert "Alice" in participant_rows assert "Bob" in participant_rows assert "John Doe" in participant_rows # Creator def test_export_trip_summary_json(self): """Test exporting trip summary as JSON""" # Create test data trip = create_test_trip() participant1 = create_test_participant(trip["id"], "Alice") participant2 = create_test_participant(trip["id"], "Bob") # Create expenses with different scenarios create_test_expense( trip["id"], participant1["id"], [ {"participant_id": participant1["id"], "percentage": 50.0}, {"participant_id": participant2["id"], "percentage": 50.0} ] ) create_test_expense( trip["id"], participant2["id"], [ {"participant_id": participant1["id"], "percentage": 100.0} ] ) # Export summary response = client.get(f"/api/export/{trip['id']}/summary/json") assert response.status_code == 200 assert response.headers["content-type"] == "application/json" summary = response.json() # Check trip information assert "trip_info" in summary trip_info = summary["trip_info"] assert trip_info["id"] == trip["id"] assert trip_info["name"] == trip["name"] assert trip_info["share_code"] == trip["share_code"] assert trip_info["currency_code"] == "USD" # Check participants summary assert "participants_summary" in summary participants = summary["participants_summary"] assert len(participants) == 3 # John Doe + Alice + Bob # Check expenses summary assert "expenses_summary" in summary expenses = summary["expenses_summary"] assert len(expenses) == 2 # Check balances assert "balances" in summary balances = summary["balances"] assert len(balances) == 3 def test_export_all_data_csv(self): """Test exporting all trip data as combined CSV""" # Create test data trip = create_test_trip() participant1 = create_test_participant(trip["id"], "Alice") participant2 = create_test_participant(trip["id"], "Bob") # Create expense create_test_expense( trip["id"], participant1["id"], [ {"participant_id": participant1["id"], "percentage": 50.0}, {"participant_id": participant2["id"], "percentage": 50.0} ] ) # Export all data response = client.get(f"/api/export/{trip['id']}/all/csv") assert response.status_code == 200 assert response.headers["content-type"] == "text/csv; charset=utf-8" # Parse CSV content csv_content = response.content.decode('utf-8') # Check that it contains different sections assert "=== TRIP INFORMATION ===" in csv_content assert "=== PARTICIPANTS ===" in csv_content assert "=== EXPENSES ===" in csv_content # Check trip information assert f"Name,{trip['name']}" in csv_content assert f"Share Code,{trip['share_code']}" in csv_content assert f"Currency,{trip['currency_code']}" in csv_content def test_export_with_nonexistent_trip(self): """Test export endpoints with non-existent trip""" # Test expenses export response = client.get("/api/export/99999/expenses/csv") assert response.status_code == 404 # Test participants export response = client.get("/api/export/99999/participants/csv") assert response.status_code == 404 # Test summary export response = client.get("/api/export/99999/summary/json") assert response.status_code == 404 # Test all data export response = client.get("/api/export/99999/all/csv") assert response.status_code == 404 def test_export_with_currency_conversion(self): """Test export with currency conversion in expenses""" # Create trip with EUR trip_data = { "name": "Europe Trip", "creator_name": "Alice", "currency_code": "EUR" } trip_response = client.post("/api/trips/", json=trip_data) trip = trip_response.json() participant = create_test_participant(trip["id"], "Bob") # Create expense in USD expense_data = { "description": "Hotel Booking", "amount": 120.00, "currency_code": "USD", "exchange_rate": 0.85, # 1 USD = 0.85 EUR "category": "Accommodation", "paid_by_id": participant["id"], "splits": [ {"participant_id": participant["id"], "percentage": 100.0} ] } client.post(f"/api/expenses/{trip['id']}/expenses", json=expense_data) # Export expenses response = client.get(f"/api/export/{trip['id']}/expenses/csv") assert response.status_code == 200 # Parse CSV and check currency conversion csv_content = response.content.decode('utf-8') csv_reader = csv.reader(io.StringIO(csv_content)) rows = list(csv_reader) # Find the amount and amount in trip currency columns header_row = rows[0] amount_idx = header_row.index("Amount") amount_trip_idx = header_row.index("Amount in Trip Currency") currency_idx = header_row.index("Currency") expense_row = rows[1] assert expense_row[amount_idx] == "120.0" # Original USD amount assert expense_row[amount_trip_idx] == "102.0" # Converted EUR amount assert expense_row[currency_idx] == "USD" def test_export_settlement_calculations(self): """Test that export includes correct settlement calculations""" # Create test data with complex splits trip = create_test_trip() participant1 = create_test_participant(trip["id"], "Alice") participant2 = create_test_participant(trip["id"], "Bob") # Alice pays $100, split 50/50 create_test_expense( trip["id"], participant1["id"], [ {"participant_id": participant1["id"], "percentage": 50.0}, {"participant_id": participant2["id"], "percentage": 50.0} ] ) # Bob pays $60, used by Alice only create_test_expense( trip["id"], participant2["id"], [ {"participant_id": participant1["id"], "percentage": 100.0} ] ) # Export summary response = client.get(f"/api/export/{trip['id']}/summary/json") summary = response.json() # Check balances calculation balances = summary["balances"] # Alice paid $100, owes $50 + $60 = $110, balance = $100 - $110 = -$10 (owes money) alice_balance = next(b for b in balances if b["participant_name"] == "Alice") assert alice_balance["total_paid"] == 100.0 assert alice_balance["total_owed"] == 110.0 assert alice_balance["balance"] == -10.0 # Bob paid $60, owes $50, balance = $60 - $50 = $10 (is owed money) bob_balance = next(b for b in balances if b["participant_name"] == "Bob") assert bob_balance["total_paid"] == 60.0 assert bob_balance["total_owed"] == 50.0 assert bob_balance["balance"] == 10.0 # Check settlement suggestions assert "settlements" in summary settlements = summary["settlements"] assert len(settlements) == 1 # Alice should pay Bob $10 settlement = settlements[0] assert settlement["from"] == "Alice" assert settlement["to"] == "Bob" assert settlement["amount"] == 10.0 if __name__ == "__main__": pytest.main([__file__, "-v"])