306 lines
10 KiB
Python
306 lines
10 KiB
Python
"""Tests for DATEV export module: export creation, CSV format, date range validation."""
|
|
|
|
import csv
|
|
import io
|
|
import uuid
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.sale import Sale
|
|
from app.models.vehicle import Vehicle
|
|
from app.models.contact import Contact
|
|
from app.utils.datev import generate_datev_csv, validate_datev_csv, DATEV_HEADERS
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_vehicle_for_datev(db_session: AsyncSession) -> Vehicle:
|
|
"""Create a test vehicle for DATEV tests."""
|
|
vehicle = Vehicle(
|
|
make="MAN",
|
|
model="TGX",
|
|
fin="WMA12345678901234",
|
|
year=2021,
|
|
vehicle_type="lkw",
|
|
price=Decimal("55000.00"),
|
|
availability="available",
|
|
condition="used",
|
|
)
|
|
db_session.add(vehicle)
|
|
await db_session.commit()
|
|
await db_session.refresh(vehicle)
|
|
return vehicle
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_buyer_for_datev(db_session: AsyncSession) -> Contact:
|
|
"""Create a test buyer contact for DATEV tests."""
|
|
contact = Contact(
|
|
company_name="DATEV Test Buyer GmbH",
|
|
role="kaeufer",
|
|
address_country="DE",
|
|
vat_id="DE111222333",
|
|
address_street="Testweg 3",
|
|
address_zip="67890",
|
|
address_city="Hamburg",
|
|
)
|
|
db_session.add(contact)
|
|
await db_session.commit()
|
|
await db_session.refresh(contact)
|
|
return contact
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_completed_sales(
|
|
db_session: AsyncSession, test_vehicle_for_datev, test_buyer_for_datev
|
|
) -> list[Sale]:
|
|
"""Create test completed sales within a date range."""
|
|
sales = []
|
|
for i, (day, price) in enumerate(
|
|
[(5, Decimal("30000.00")), (10, Decimal("25000.00")), (15, Decimal("40000.00"))]
|
|
):
|
|
sale = Sale(
|
|
vehicle_id=test_vehicle_for_datev.id,
|
|
buyer_contact_id=test_buyer_for_datev.id,
|
|
sale_price=price,
|
|
sale_date=date(2025, 1, day),
|
|
status="completed",
|
|
is_gwg=False,
|
|
)
|
|
db_session.add(sale)
|
|
sales.append(sale)
|
|
await db_session.commit()
|
|
for s in sales:
|
|
await db_session.refresh(s)
|
|
return sales
|
|
|
|
|
|
class TestDATEVExportAPI:
|
|
"""Test DATEV export API endpoints."""
|
|
|
|
async def test_create_export(self, admin_client: AsyncClient, test_completed_sales):
|
|
"""POST /datev/export with valid date range returns 201."""
|
|
response = await admin_client.post(
|
|
"/api/v1/datev/export",
|
|
json={
|
|
"start_date": "2025-01-01",
|
|
"end_date": "2025-01-31",
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["start_date"] == "2025-01-01"
|
|
assert data["end_date"] == "2025-01-31"
|
|
assert data["file_path"] is not None
|
|
assert float(data["total_amount"]) == 95000.00
|
|
|
|
async def test_create_export_invalid_date_range(self, admin_client: AsyncClient):
|
|
"""POST /datev/export with start > end returns 422."""
|
|
response = await admin_client.post(
|
|
"/api/v1/datev/export",
|
|
json={
|
|
"start_date": "2025-12-31",
|
|
"end_date": "2025-01-01",
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
async def test_list_exports(self, admin_client: AsyncClient, test_completed_sales):
|
|
"""GET /datev/exports returns list of exports."""
|
|
# First create an export
|
|
await admin_client.post(
|
|
"/api/v1/datev/export",
|
|
json={
|
|
"start_date": "2025-01-01",
|
|
"end_date": "2025-01-31",
|
|
},
|
|
)
|
|
|
|
response = await admin_client.get("/api/v1/datev/exports")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 1
|
|
assert len(data["items"]) >= 1
|
|
assert "start_date" in data["items"][0]
|
|
assert "end_date" in data["items"][0]
|
|
|
|
async def test_download_export_csv(
|
|
self, admin_client: AsyncClient, test_completed_sales
|
|
):
|
|
"""GET /datev/exports/:id/download returns CSV content."""
|
|
# Create export
|
|
create_response = await admin_client.post(
|
|
"/api/v1/datev/export",
|
|
json={
|
|
"start_date": "2025-01-01",
|
|
"end_date": "2025-01-31",
|
|
},
|
|
)
|
|
assert create_response.status_code == 201
|
|
export_id = create_response.json()["id"]
|
|
|
|
# Download
|
|
response = await admin_client.get(f"/api/v1/datev/exports/{export_id}/download")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "text/csv; charset=utf-8"
|
|
|
|
# Parse CSV and verify headers
|
|
csv_content = response.text
|
|
reader = csv.reader(io.StringIO(csv_content), delimiter=";")
|
|
header = next(reader)
|
|
assert header == DATEV_HEADERS
|
|
|
|
# Verify at least one data row
|
|
rows = list(reader)
|
|
assert len(rows) >= 1
|
|
for row in rows:
|
|
assert len(row) == 6
|
|
|
|
async def test_download_export_not_found(self, admin_client: AsyncClient):
|
|
"""GET /datev/exports/:nonexistent/download returns 404."""
|
|
response = await admin_client.get(
|
|
f"/api/v1/datev/exports/{uuid.uuid4()}/download"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestDATEVCSVFormat:
|
|
"""Test DATEV CSV format helper functions."""
|
|
|
|
def test_datev_csv_headers(self):
|
|
"""DATEV CSV has correct headers."""
|
|
assert DATEV_HEADERS == [
|
|
"Datum",
|
|
"Konto",
|
|
"Gegenkonto",
|
|
"Betrag",
|
|
"Belegfeld",
|
|
"Buchungstext",
|
|
]
|
|
|
|
def test_generate_datev_csv_empty(self):
|
|
"""Generate DATEV CSV with no sales returns only headers."""
|
|
csv_content = generate_datev_csv([])
|
|
assert validate_datev_csv(csv_content)
|
|
|
|
reader = csv.reader(io.StringIO(csv_content), delimiter=";")
|
|
header = next(reader)
|
|
assert header == DATEV_HEADERS
|
|
# No data rows
|
|
rows = list(reader)
|
|
assert len(rows) == 0
|
|
|
|
def test_generate_datev_csv_with_sales(self, test_completed_sales):
|
|
"""Generate DATEV CSV with sales produces correct rows."""
|
|
|
|
# test_completed_sales is a fixture but we need to call it differently for sync test
|
|
# Instead, create mock objects
|
|
class MockVehicle:
|
|
make = "MAN"
|
|
model = "TGX"
|
|
fin = "WMA12345678901234"
|
|
|
|
class MockSale:
|
|
def __init__(self, sale_id, sale_date, sale_price):
|
|
self.id = sale_id
|
|
self.sale_date = sale_date
|
|
self.sale_price = sale_price
|
|
self.vehicle = MockVehicle()
|
|
|
|
sales = [
|
|
MockSale(uuid.uuid4(), date(2025, 1, 5), Decimal("30000.00")),
|
|
MockSale(uuid.uuid4(), date(2025, 1, 10), Decimal("25000.00")),
|
|
]
|
|
|
|
csv_content = generate_datev_csv(sales)
|
|
assert validate_datev_csv(csv_content)
|
|
|
|
reader = csv.reader(io.StringIO(csv_content), delimiter=";")
|
|
header = next(reader)
|
|
assert header == DATEV_HEADERS
|
|
|
|
rows = list(reader)
|
|
assert len(rows) == 2
|
|
|
|
# Check first row format
|
|
row1 = rows[0]
|
|
assert row1[0] == "05.01.2025" # Datum in DD.MM.YYYY
|
|
assert row1[1] == "1200" # Konto
|
|
assert row1[2] == "8400" # Gegenkonto
|
|
assert "," in row1[3] # Betrag with comma separator
|
|
assert row1[4] # Belegfeld (short UUID)
|
|
assert "MAN" in row1[5] # Buchungstext contains vehicle make
|
|
|
|
def test_validate_datev_csv_invalid_headers(self):
|
|
"""Validate DATEV CSV with wrong headers returns False."""
|
|
csv_content = "Wrong;Headers;Here\nval1;val2;val3"
|
|
assert not validate_datev_csv(csv_content)
|
|
|
|
def test_validate_datev_csv_empty(self):
|
|
"""Validate empty CSV returns False."""
|
|
assert not validate_datev_csv("")
|
|
assert not validate_datev_csv(" ")
|
|
|
|
def test_validate_datev_csv_valid(self):
|
|
"""Validate correct DATEV CSV returns True."""
|
|
csv_content = "Datum;Konto;Gegenkonto;Betrag;Belegfeld;Buchungstext\n05.01.2025;1200;8400;30000,00;ABC12345;MAN TGX"
|
|
assert validate_datev_csv(csv_content)
|
|
|
|
def test_datev_csv_amount_format(self):
|
|
"""DATEV CSV amount uses comma as decimal separator."""
|
|
|
|
class MockVehicle:
|
|
make = "VW"
|
|
model = "Crafter"
|
|
fin = "WVW12345678901234"
|
|
|
|
class MockSale:
|
|
def __init__(self):
|
|
self.id = uuid.uuid4()
|
|
self.sale_date = date(2025, 3, 15)
|
|
self.sale_price = Decimal("12345.67")
|
|
self.vehicle = MockVehicle()
|
|
|
|
csv_content = generate_datev_csv([MockSale()])
|
|
reader = csv.reader(io.StringIO(csv_content), delimiter=";")
|
|
next(reader) # skip header
|
|
row = next(reader)
|
|
assert row[3] == "12345,67" # Comma as decimal separator
|
|
|
|
|
|
class TestDATEVExportService:
|
|
"""Test DATEV export service directly."""
|
|
|
|
async def test_create_export_no_sales_in_range(self, db_session: AsyncSession):
|
|
"""Creating export with no sales in range produces empty CSV with 0 total."""
|
|
from app.services import datev_service
|
|
|
|
export = await datev_service.create_export(
|
|
db_session,
|
|
start_date=date(2025, 6, 1),
|
|
end_date=date(2025, 6, 30),
|
|
)
|
|
assert export.total_amount == Decimal("0")
|
|
assert export.file_path is not None
|
|
|
|
async def test_list_exports_pagination(self, db_session: AsyncSession):
|
|
"""List exports with pagination."""
|
|
from app.services import datev_service
|
|
|
|
# Create a few exports
|
|
for _ in range(3):
|
|
await datev_service.create_export(
|
|
db_session,
|
|
start_date=date(2025, 1, 1),
|
|
end_date=date(2025, 1, 31),
|
|
)
|
|
|
|
exports, total = await datev_service.list_exports(
|
|
db_session, page=1, page_size=2
|
|
)
|
|
assert total >= 3
|
|
assert len(exports) <= 2
|