feat: add vendor invoice management UI and comprehensive tests
UI Components: - Add vendor invoices page route in vendor_pages.py - Create invoices.html template with stats cards, invoice table, settings tab, and create invoice modal - Add invoices.js Alpine.js component for CRUD operations, PDF download, and settings management - Add Invoices link to vendor sidebar in Sales section Unit Tests (35 tests): - VAT calculation (EU rates, regimes, labels) - Invoice settings CRUD and number generation - Invoice retrieval, listing, and pagination - Status management and validation - Statistics calculation Integration Tests (34 tests): - Settings API endpoints (GET/POST/PUT) - Stats API endpoint - Invoice list with filtering and pagination - Invoice detail retrieval - Invoice creation from orders - Status update transitions - PDF generation endpoints - Authentication/authorization checks 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
867
tests/integration/api/v1/vendor/test_invoices.py
vendored
Normal file
867
tests/integration/api/v1/vendor/test_invoices.py
vendored
Normal file
@@ -0,0 +1,867 @@
|
||||
# tests/integration/api/v1/vendor/test_invoices.py
|
||||
"""Integration tests for vendor invoice management endpoints.
|
||||
|
||||
Tests the /api/v1/vendor/invoices/* endpoints.
|
||||
All endpoints require vendor JWT authentication.
|
||||
"""
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from models.database.customer import Customer
|
||||
from models.database.invoice import Invoice, InvoiceStatus, VendorInvoiceSettings
|
||||
from models.database.order import Order
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
class TestVendorInvoiceSettingsAPI:
|
||||
"""Test vendor invoice settings endpoints at /api/v1/vendor/invoices/settings."""
|
||||
|
||||
def test_get_settings_not_configured(self, client, vendor_user_headers):
|
||||
"""Test getting settings when not configured returns null."""
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices/settings",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() is None
|
||||
|
||||
def test_create_settings_success(self, client, vendor_user_headers):
|
||||
"""Test creating invoice settings successfully."""
|
||||
settings_data = {
|
||||
"company_name": "Test Company S.A.",
|
||||
"company_address": "123 Test Street",
|
||||
"company_city": "Luxembourg",
|
||||
"company_postal_code": "L-1234",
|
||||
"company_country": "LU",
|
||||
"vat_number": "LU12345678",
|
||||
"invoice_prefix": "INV",
|
||||
"default_vat_rate": 17.0,
|
||||
"bank_name": "BCEE",
|
||||
"bank_iban": "LU123456789012345678",
|
||||
"bank_bic": "BCEELULL",
|
||||
"payment_terms": "Net 30 days",
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices/settings",
|
||||
headers=vendor_user_headers,
|
||||
json=settings_data,
|
||||
)
|
||||
|
||||
assert response.status_code == 201, f"Failed: {response.json()}"
|
||||
data = response.json()
|
||||
assert data["company_name"] == "Test Company S.A."
|
||||
assert data["company_country"] == "LU"
|
||||
assert data["invoice_prefix"] == "INV"
|
||||
|
||||
def test_create_settings_minimal(self, client, vendor_user_headers):
|
||||
"""Test creating settings with minimal required data."""
|
||||
settings_data = {
|
||||
"company_name": "Minimal Company",
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices/settings",
|
||||
headers=vendor_user_headers,
|
||||
json=settings_data,
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["company_name"] == "Minimal Company"
|
||||
# Defaults should be applied
|
||||
assert data["invoice_prefix"] == "INV"
|
||||
|
||||
def test_create_settings_duplicate_fails(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test creating duplicate settings fails."""
|
||||
# Create settings directly in DB
|
||||
settings = VendorInvoiceSettings(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
company_name="Existing Company",
|
||||
company_country="LU",
|
||||
)
|
||||
db.add(settings)
|
||||
db.commit()
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices/settings",
|
||||
headers=vendor_user_headers,
|
||||
json={"company_name": "New Company"},
|
||||
)
|
||||
|
||||
# ValidationException returns 422
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_get_settings_success(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test getting existing settings."""
|
||||
# Create settings
|
||||
settings = VendorInvoiceSettings(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
company_name="Get Settings Company",
|
||||
company_country="LU",
|
||||
invoice_prefix="FAC",
|
||||
)
|
||||
db.add(settings)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices/settings",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["company_name"] == "Get Settings Company"
|
||||
assert data["invoice_prefix"] == "FAC"
|
||||
|
||||
def test_update_settings_success(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test updating invoice settings."""
|
||||
# Create settings
|
||||
settings = VendorInvoiceSettings(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
company_name="Original Company",
|
||||
company_country="LU",
|
||||
)
|
||||
db.add(settings)
|
||||
db.commit()
|
||||
|
||||
update_data = {
|
||||
"company_name": "Updated Company",
|
||||
"bank_iban": "LU999888777666555444",
|
||||
}
|
||||
|
||||
response = client.put(
|
||||
"/api/v1/vendor/invoices/settings",
|
||||
headers=vendor_user_headers,
|
||||
json=update_data,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["company_name"] == "Updated Company"
|
||||
assert data["bank_iban"] == "LU999888777666555444"
|
||||
|
||||
def test_update_settings_not_found(self, client, vendor_user_headers):
|
||||
"""Test updating non-existent settings returns error."""
|
||||
response = client.put(
|
||||
"/api/v1/vendor/invoices/settings",
|
||||
headers=vendor_user_headers,
|
||||
json={"company_name": "Updated"},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_settings_without_auth_returns_unauthorized(self, client):
|
||||
"""Test accessing settings without auth returns 401."""
|
||||
response = client.get("/api/v1/vendor/invoices/settings")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
class TestVendorInvoiceStatsAPI:
|
||||
"""Test vendor invoice statistics endpoint at /api/v1/vendor/invoices/stats."""
|
||||
|
||||
def test_get_stats_empty(self, client, vendor_user_headers):
|
||||
"""Test getting stats when no invoices exist."""
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices/stats",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["total_invoices"] == 0
|
||||
assert data["total_revenue_cents"] == 0
|
||||
assert data["draft_count"] == 0
|
||||
assert data["paid_count"] == 0
|
||||
|
||||
def test_get_stats_with_invoices(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test getting stats with existing invoices."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
# Create invoices
|
||||
for i, status in enumerate([InvoiceStatus.DRAFT, InvoiceStatus.PAID]):
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number=f"INV0000{i+1}",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=status.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000 * (i + 1),
|
||||
vat_amount_cents=1700 * (i + 1),
|
||||
total_cents=11700 * (i + 1),
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices/stats",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["total_invoices"] == 2
|
||||
assert data["draft_count"] == 1
|
||||
assert data["paid_count"] == 1
|
||||
|
||||
def test_stats_without_auth_returns_unauthorized(self, client):
|
||||
"""Test accessing stats without auth returns 401."""
|
||||
response = client.get("/api/v1/vendor/invoices/stats")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
class TestVendorInvoiceListAPI:
|
||||
"""Test vendor invoice list endpoint at /api/v1/vendor/invoices."""
|
||||
|
||||
def test_list_invoices_empty(self, client, vendor_user_headers):
|
||||
"""Test listing invoices when none exist."""
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["items"] == []
|
||||
assert data["total"] == 0
|
||||
assert data["page"] == 1
|
||||
|
||||
def test_list_invoices_success(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test listing invoices successfully."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
# Create invoices
|
||||
for i in range(3):
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number=f"INV0000{i+1}",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": f"Buyer {i+1}"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data["items"]) == 3
|
||||
assert data["total"] == 3
|
||||
|
||||
def test_list_invoices_with_status_filter(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test filtering invoices by status."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
# Create invoices with different statuses
|
||||
for status in [InvoiceStatus.DRAFT, InvoiceStatus.ISSUED, InvoiceStatus.PAID]:
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number=f"INV-{status.value.upper()}",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=status.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
# Filter by paid status
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices?status=paid",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["total"] == 1
|
||||
assert data["items"][0]["status"] == "paid"
|
||||
|
||||
def test_list_invoices_pagination(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test invoice list pagination."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
# Create 5 invoices
|
||||
for i in range(5):
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number=f"INV0000{i+1}",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
# Get first page with 2 items
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices?page=1&per_page=2",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data["items"]) == 2
|
||||
assert data["total"] == 5
|
||||
assert data["page"] == 1
|
||||
assert data["pages"] == 3
|
||||
|
||||
def test_list_invoices_without_auth_returns_unauthorized(self, client):
|
||||
"""Test listing invoices without auth returns 401."""
|
||||
response = client.get("/api/v1/vendor/invoices")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
class TestVendorInvoiceDetailAPI:
|
||||
"""Test vendor invoice detail endpoint at /api/v1/vendor/invoices/{id}."""
|
||||
|
||||
def test_get_invoice_success(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test getting invoice by ID."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Seller Co"},
|
||||
buyer_details={"name": "John Doe"},
|
||||
line_items=[
|
||||
{
|
||||
"description": "Product A",
|
||||
"quantity": 2,
|
||||
"unit_price_cents": 5000,
|
||||
"total_cents": 10000,
|
||||
}
|
||||
],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
|
||||
response = client.get(
|
||||
f"/api/v1/vendor/invoices/{invoice.id}",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["invoice_number"] == "INV00001"
|
||||
assert data["total_cents"] == 11700
|
||||
assert len(data["line_items"]) == 1
|
||||
|
||||
def test_get_invoice_not_found(self, client, vendor_user_headers):
|
||||
"""Test getting non-existent invoice returns 404."""
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices/99999",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_get_invoice_without_auth_returns_unauthorized(self, client):
|
||||
"""Test getting invoice without auth returns 401."""
|
||||
response = client.get("/api/v1/vendor/invoices/1")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
class TestVendorInvoiceCreateAPI:
|
||||
"""Test vendor invoice creation endpoint at /api/v1/vendor/invoices."""
|
||||
|
||||
def test_create_invoice_success(
|
||||
self,
|
||||
client,
|
||||
vendor_user_headers,
|
||||
db,
|
||||
test_vendor_with_vendor_user,
|
||||
):
|
||||
"""Test creating invoice from order."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
now = datetime.now(UTC)
|
||||
|
||||
# Create invoice settings first
|
||||
settings = VendorInvoiceSettings(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
company_name="Test Company",
|
||||
company_country="LU",
|
||||
invoice_prefix="INV",
|
||||
invoice_next_number=1,
|
||||
)
|
||||
db.add(settings)
|
||||
db.commit()
|
||||
|
||||
# Create a customer first
|
||||
customer = Customer(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
email="test@example.com",
|
||||
hashed_password="$2b$12$test_hashed_password",
|
||||
first_name="John",
|
||||
last_name="Doe",
|
||||
customer_number="CUST-001",
|
||||
)
|
||||
db.add(customer)
|
||||
db.commit()
|
||||
db.refresh(customer)
|
||||
|
||||
# Create an order
|
||||
order = Order(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
customer_id=customer.id,
|
||||
order_number="ORD-001",
|
||||
channel="direct",
|
||||
order_date=now,
|
||||
customer_first_name="John",
|
||||
customer_last_name="Doe",
|
||||
customer_email="test@example.com",
|
||||
ship_first_name="John",
|
||||
ship_last_name="Doe",
|
||||
ship_address_line_1="123 Test St",
|
||||
ship_city="Luxembourg",
|
||||
ship_postal_code="L-1234",
|
||||
ship_country_iso="LU",
|
||||
bill_first_name="John",
|
||||
bill_last_name="Doe",
|
||||
bill_address_line_1="123 Test St",
|
||||
bill_city="Luxembourg",
|
||||
bill_postal_code="L-1234",
|
||||
bill_country_iso="LU",
|
||||
currency="EUR",
|
||||
status="completed",
|
||||
subtotal_cents=10000,
|
||||
total_amount_cents=11700,
|
||||
)
|
||||
db.add(order)
|
||||
db.commit()
|
||||
db.refresh(order)
|
||||
|
||||
# Create invoice (without order items - service handles empty items)
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices",
|
||||
headers=vendor_user_headers,
|
||||
json={"order_id": order.id, "notes": "Test invoice"},
|
||||
)
|
||||
|
||||
assert response.status_code == 201, f"Failed: {response.json()}"
|
||||
data = response.json()
|
||||
assert data["order_id"] == order.id
|
||||
assert data["invoice_number"] == "INV00001"
|
||||
assert data["status"] == "draft"
|
||||
assert data["notes"] == "Test invoice"
|
||||
|
||||
def test_create_invoice_without_settings_fails(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test creating invoice without settings configured fails."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
now = datetime.now(UTC)
|
||||
|
||||
# Create a customer first
|
||||
customer = Customer(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
email="jane@example.com",
|
||||
hashed_password="$2b$12$test_hashed_password",
|
||||
first_name="Jane",
|
||||
last_name="Doe",
|
||||
customer_number="CUST-002",
|
||||
)
|
||||
db.add(customer)
|
||||
db.commit()
|
||||
db.refresh(customer)
|
||||
|
||||
# Create an order without settings
|
||||
order = Order(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
customer_id=customer.id,
|
||||
order_number="ORD-002",
|
||||
channel="direct",
|
||||
order_date=now,
|
||||
customer_first_name="Jane",
|
||||
customer_last_name="Doe",
|
||||
customer_email="jane@example.com",
|
||||
ship_first_name="Jane",
|
||||
ship_last_name="Doe",
|
||||
ship_address_line_1="456 Test Ave",
|
||||
ship_city="Luxembourg",
|
||||
ship_postal_code="L-5678",
|
||||
ship_country_iso="LU",
|
||||
bill_first_name="Jane",
|
||||
bill_last_name="Doe",
|
||||
bill_address_line_1="456 Test Ave",
|
||||
bill_city="Luxembourg",
|
||||
bill_postal_code="L-5678",
|
||||
bill_country_iso="LU",
|
||||
currency="EUR",
|
||||
status="completed",
|
||||
subtotal_cents=10000,
|
||||
total_amount_cents=11700,
|
||||
)
|
||||
db.add(order)
|
||||
db.commit()
|
||||
db.refresh(order)
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices",
|
||||
headers=vendor_user_headers,
|
||||
json={"order_id": order.id},
|
||||
)
|
||||
|
||||
assert response.status_code == 404 # Settings not found
|
||||
|
||||
def test_create_invoice_order_not_found(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test creating invoice for non-existent order fails."""
|
||||
# Create settings
|
||||
settings = VendorInvoiceSettings(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
company_name="Test Company",
|
||||
company_country="LU",
|
||||
)
|
||||
db.add(settings)
|
||||
db.commit()
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices",
|
||||
headers=vendor_user_headers,
|
||||
json={"order_id": 99999},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_create_invoice_without_auth_returns_unauthorized(self, client):
|
||||
"""Test creating invoice without auth returns 401."""
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices",
|
||||
json={"order_id": 1},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
class TestVendorInvoiceStatusAPI:
|
||||
"""Test vendor invoice status update endpoint."""
|
||||
|
||||
def test_update_status_to_issued(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test updating invoice status to issued."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
|
||||
response = client.put(
|
||||
f"/api/v1/vendor/invoices/{invoice.id}/status",
|
||||
headers=vendor_user_headers,
|
||||
json={"status": "issued"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "issued"
|
||||
|
||||
def test_update_status_to_paid(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test updating invoice status to paid."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.ISSUED.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
|
||||
response = client.put(
|
||||
f"/api/v1/vendor/invoices/{invoice.id}/status",
|
||||
headers=vendor_user_headers,
|
||||
json={"status": "paid"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "paid"
|
||||
|
||||
def test_update_status_to_cancelled(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test cancelling an invoice."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
|
||||
response = client.put(
|
||||
f"/api/v1/vendor/invoices/{invoice.id}/status",
|
||||
headers=vendor_user_headers,
|
||||
json={"status": "cancelled"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "cancelled"
|
||||
|
||||
def test_update_cancelled_invoice_fails(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test updating cancelled invoice fails."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.CANCELLED.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
|
||||
response = client.put(
|
||||
f"/api/v1/vendor/invoices/{invoice.id}/status",
|
||||
headers=vendor_user_headers,
|
||||
json={"status": "issued"},
|
||||
)
|
||||
|
||||
# ValidationException returns 422
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_update_status_invalid_status(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test updating with invalid status fails."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
|
||||
response = client.put(
|
||||
f"/api/v1/vendor/invoices/{invoice.id}/status",
|
||||
headers=vendor_user_headers,
|
||||
json={"status": "invalid_status"},
|
||||
)
|
||||
|
||||
# ValidationException returns 422
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_update_status_not_found(self, client, vendor_user_headers):
|
||||
"""Test updating non-existent invoice fails."""
|
||||
response = client.put(
|
||||
"/api/v1/vendor/invoices/99999/status",
|
||||
headers=vendor_user_headers,
|
||||
json={"status": "issued"},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_update_status_without_auth_returns_unauthorized(self, client):
|
||||
"""Test updating status without auth returns 401."""
|
||||
response = client.put(
|
||||
"/api/v1/vendor/invoices/1/status",
|
||||
json={"status": "issued"},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
class TestVendorInvoicePDFAPI:
|
||||
"""Test vendor invoice PDF endpoints."""
|
||||
|
||||
@pytest.mark.skip(reason="WeasyPrint not installed in test environment")
|
||||
def test_generate_pdf_success(
|
||||
self, client, vendor_user_headers, db, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test generating PDF for an invoice."""
|
||||
from datetime import datetime, UTC
|
||||
|
||||
# Create settings
|
||||
settings = VendorInvoiceSettings(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
company_name="PDF Test Company",
|
||||
company_country="LU",
|
||||
)
|
||||
db.add(settings)
|
||||
db.commit()
|
||||
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=datetime.now(UTC),
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "PDF Test Company"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[
|
||||
{
|
||||
"description": "Test Item",
|
||||
"quantity": 1,
|
||||
"unit_price_cents": 10000,
|
||||
"total_cents": 10000,
|
||||
}
|
||||
],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
|
||||
response = client.post(
|
||||
f"/api/v1/vendor/invoices/{invoice.id}/pdf",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_generate_pdf_not_found(self, client, vendor_user_headers):
|
||||
"""Test generating PDF for non-existent invoice fails."""
|
||||
response = client.post(
|
||||
"/api/v1/vendor/invoices/99999/pdf",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_download_pdf_not_found(self, client, vendor_user_headers):
|
||||
"""Test downloading PDF for non-existent invoice fails."""
|
||||
response = client.get(
|
||||
"/api/v1/vendor/invoices/99999/pdf",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_pdf_endpoints_without_auth_returns_unauthorized(self, client):
|
||||
"""Test PDF endpoints without auth return 401."""
|
||||
response = client.post("/api/v1/vendor/invoices/1/pdf")
|
||||
assert response.status_code == 401
|
||||
|
||||
response = client.get("/api/v1/vendor/invoices/1/pdf")
|
||||
assert response.status_code == 401
|
||||
622
tests/unit/services/test_invoice_service.py
Normal file
622
tests/unit/services/test_invoice_service.py
Normal file
@@ -0,0 +1,622 @@
|
||||
# tests/unit/services/test_invoice_service.py
|
||||
"""Unit tests for InvoiceService."""
|
||||
|
||||
import uuid
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from app.exceptions import ValidationException
|
||||
from app.exceptions.invoice import (
|
||||
InvoiceNotFoundException,
|
||||
InvoiceSettingsNotFoundException,
|
||||
)
|
||||
from app.services.invoice_service import (
|
||||
EU_VAT_RATES,
|
||||
InvoiceService,
|
||||
LU_VAT_RATES,
|
||||
)
|
||||
from models.database.invoice import (
|
||||
Invoice,
|
||||
InvoiceStatus,
|
||||
VATRegime,
|
||||
VendorInvoiceSettings,
|
||||
)
|
||||
from models.schema.invoice import (
|
||||
VendorInvoiceSettingsCreate,
|
||||
VendorInvoiceSettingsUpdate,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.invoice
|
||||
class TestInvoiceServiceVATCalculation:
|
||||
"""Test suite for InvoiceService VAT calculation methods."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = InvoiceService()
|
||||
|
||||
# ==================== VAT Rate Lookup Tests ====================
|
||||
|
||||
def test_get_vat_rate_for_luxembourg(self):
|
||||
"""Test Luxembourg VAT rate is 17%."""
|
||||
rate = self.service.get_vat_rate_for_country("LU")
|
||||
assert rate == Decimal("17.00")
|
||||
|
||||
def test_get_vat_rate_for_germany(self):
|
||||
"""Test Germany VAT rate is 19%."""
|
||||
rate = self.service.get_vat_rate_for_country("DE")
|
||||
assert rate == Decimal("19.00")
|
||||
|
||||
def test_get_vat_rate_for_france(self):
|
||||
"""Test France VAT rate is 20%."""
|
||||
rate = self.service.get_vat_rate_for_country("FR")
|
||||
assert rate == Decimal("20.00")
|
||||
|
||||
def test_get_vat_rate_for_non_eu_country(self):
|
||||
"""Test non-EU country returns 0% VAT."""
|
||||
rate = self.service.get_vat_rate_for_country("US")
|
||||
assert rate == Decimal("0.00")
|
||||
|
||||
def test_get_vat_rate_lowercase_country(self):
|
||||
"""Test VAT rate lookup works with lowercase country codes."""
|
||||
rate = self.service.get_vat_rate_for_country("de")
|
||||
assert rate == Decimal("19.00")
|
||||
|
||||
# ==================== VAT Rate Label Tests ====================
|
||||
|
||||
def test_get_vat_rate_label_luxembourg(self):
|
||||
"""Test VAT rate label for Luxembourg."""
|
||||
label = self.service.get_vat_rate_label("LU", Decimal("17.00"))
|
||||
assert "Luxembourg" in label
|
||||
assert "17" in label
|
||||
|
||||
def test_get_vat_rate_label_germany(self):
|
||||
"""Test VAT rate label for Germany."""
|
||||
label = self.service.get_vat_rate_label("DE", Decimal("19.00"))
|
||||
assert "Germany" in label
|
||||
assert "19" in label
|
||||
|
||||
# ==================== VAT Regime Determination Tests ====================
|
||||
|
||||
def test_determine_vat_regime_domestic(self):
|
||||
"""Test domestic sales (same country) use domestic VAT."""
|
||||
regime, rate, dest = self.service.determine_vat_regime(
|
||||
seller_country="LU",
|
||||
buyer_country="LU",
|
||||
buyer_vat_number=None,
|
||||
seller_oss_registered=False,
|
||||
)
|
||||
assert regime == VATRegime.DOMESTIC
|
||||
assert rate == Decimal("17.00")
|
||||
assert dest is None
|
||||
|
||||
def test_determine_vat_regime_reverse_charge(self):
|
||||
"""Test B2B with valid VAT number uses reverse charge."""
|
||||
regime, rate, dest = self.service.determine_vat_regime(
|
||||
seller_country="LU",
|
||||
buyer_country="DE",
|
||||
buyer_vat_number="DE123456789",
|
||||
seller_oss_registered=False,
|
||||
)
|
||||
assert regime == VATRegime.REVERSE_CHARGE
|
||||
assert rate == Decimal("0.00")
|
||||
assert dest == "DE"
|
||||
|
||||
def test_determine_vat_regime_oss_registered(self):
|
||||
"""Test B2C cross-border with OSS uses destination VAT."""
|
||||
regime, rate, dest = self.service.determine_vat_regime(
|
||||
seller_country="LU",
|
||||
buyer_country="DE",
|
||||
buyer_vat_number=None,
|
||||
seller_oss_registered=True,
|
||||
)
|
||||
assert regime == VATRegime.OSS
|
||||
assert rate == Decimal("19.00") # German VAT
|
||||
assert dest == "DE"
|
||||
|
||||
def test_determine_vat_regime_no_oss(self):
|
||||
"""Test B2C cross-border without OSS uses origin VAT."""
|
||||
regime, rate, dest = self.service.determine_vat_regime(
|
||||
seller_country="LU",
|
||||
buyer_country="DE",
|
||||
buyer_vat_number=None,
|
||||
seller_oss_registered=False,
|
||||
)
|
||||
assert regime == VATRegime.ORIGIN
|
||||
assert rate == Decimal("17.00") # Luxembourg VAT
|
||||
assert dest == "DE"
|
||||
|
||||
def test_determine_vat_regime_non_eu_exempt(self):
|
||||
"""Test non-EU sales are VAT exempt."""
|
||||
regime, rate, dest = self.service.determine_vat_regime(
|
||||
seller_country="LU",
|
||||
buyer_country="US",
|
||||
buyer_vat_number=None,
|
||||
seller_oss_registered=True,
|
||||
)
|
||||
assert regime == VATRegime.EXEMPT
|
||||
assert rate == Decimal("0.00")
|
||||
assert dest == "US"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.invoice
|
||||
class TestInvoiceServiceSettings:
|
||||
"""Test suite for InvoiceService settings management."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = InvoiceService()
|
||||
|
||||
# ==================== Get Settings Tests ====================
|
||||
|
||||
def test_get_settings_not_found(self, db, test_vendor):
|
||||
"""Test getting settings for vendor without settings returns None."""
|
||||
settings = self.service.get_settings(db, test_vendor.id)
|
||||
assert settings is None
|
||||
|
||||
def test_get_settings_or_raise_not_found(self, db, test_vendor):
|
||||
"""Test get_settings_or_raise raises when settings don't exist."""
|
||||
with pytest.raises(InvoiceSettingsNotFoundException):
|
||||
self.service.get_settings_or_raise(db, test_vendor.id)
|
||||
|
||||
# ==================== Create Settings Tests ====================
|
||||
|
||||
def test_create_settings_success(self, db, test_vendor):
|
||||
"""Test creating invoice settings successfully."""
|
||||
data = VendorInvoiceSettingsCreate(
|
||||
company_name="Test Company S.A.",
|
||||
company_address="123 Test Street",
|
||||
company_city="Luxembourg",
|
||||
company_postal_code="L-1234",
|
||||
company_country="LU",
|
||||
vat_number="LU12345678",
|
||||
)
|
||||
|
||||
settings = self.service.create_settings(db, test_vendor.id, data)
|
||||
|
||||
assert settings.vendor_id == test_vendor.id
|
||||
assert settings.company_name == "Test Company S.A."
|
||||
assert settings.company_country == "LU"
|
||||
assert settings.vat_number == "LU12345678"
|
||||
assert settings.invoice_prefix == "INV"
|
||||
assert settings.invoice_next_number == 1
|
||||
|
||||
def test_create_settings_with_custom_prefix(self, db, test_vendor):
|
||||
"""Test creating settings with custom invoice prefix."""
|
||||
data = VendorInvoiceSettingsCreate(
|
||||
company_name="Custom Prefix Company",
|
||||
invoice_prefix="FAC",
|
||||
invoice_number_padding=6,
|
||||
)
|
||||
|
||||
settings = self.service.create_settings(db, test_vendor.id, data)
|
||||
|
||||
assert settings.invoice_prefix == "FAC"
|
||||
assert settings.invoice_number_padding == 6
|
||||
|
||||
def test_create_settings_duplicate_raises(self, db, test_vendor):
|
||||
"""Test creating duplicate settings raises ValidationException."""
|
||||
data = VendorInvoiceSettingsCreate(company_name="First Settings")
|
||||
self.service.create_settings(db, test_vendor.id, data)
|
||||
|
||||
with pytest.raises(ValidationException) as exc_info:
|
||||
self.service.create_settings(db, test_vendor.id, data)
|
||||
|
||||
assert "already exist" in str(exc_info.value)
|
||||
|
||||
# ==================== Update Settings Tests ====================
|
||||
|
||||
def test_update_settings_success(self, db, test_vendor):
|
||||
"""Test updating invoice settings."""
|
||||
# Create initial settings
|
||||
create_data = VendorInvoiceSettingsCreate(
|
||||
company_name="Original Company"
|
||||
)
|
||||
self.service.create_settings(db, test_vendor.id, create_data)
|
||||
|
||||
# Update settings
|
||||
update_data = VendorInvoiceSettingsUpdate(
|
||||
company_name="Updated Company",
|
||||
bank_iban="LU123456789012345678",
|
||||
)
|
||||
settings = self.service.update_settings(db, test_vendor.id, update_data)
|
||||
|
||||
assert settings.company_name == "Updated Company"
|
||||
assert settings.bank_iban == "LU123456789012345678"
|
||||
|
||||
def test_update_settings_not_found(self, db, test_vendor):
|
||||
"""Test updating non-existent settings raises exception."""
|
||||
update_data = VendorInvoiceSettingsUpdate(company_name="Updated")
|
||||
|
||||
with pytest.raises(InvoiceSettingsNotFoundException):
|
||||
self.service.update_settings(db, test_vendor.id, update_data)
|
||||
|
||||
# ==================== Invoice Number Generation Tests ====================
|
||||
|
||||
def test_get_next_invoice_number(self, db, test_vendor):
|
||||
"""Test invoice number generation and increment."""
|
||||
create_data = VendorInvoiceSettingsCreate(
|
||||
company_name="Test Company",
|
||||
invoice_prefix="INV",
|
||||
invoice_number_padding=5,
|
||||
)
|
||||
settings = self.service.create_settings(db, test_vendor.id, create_data)
|
||||
|
||||
# Generate first invoice number
|
||||
num1 = self.service._get_next_invoice_number(db, settings)
|
||||
assert num1 == "INV00001"
|
||||
assert settings.invoice_next_number == 2
|
||||
|
||||
# Generate second invoice number
|
||||
num2 = self.service._get_next_invoice_number(db, settings)
|
||||
assert num2 == "INV00002"
|
||||
assert settings.invoice_next_number == 3
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.invoice
|
||||
class TestInvoiceServiceCRUD:
|
||||
"""Test suite for InvoiceService CRUD operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = InvoiceService()
|
||||
|
||||
# ==================== Get Invoice Tests ====================
|
||||
|
||||
def test_get_invoice_not_found(self, db, test_vendor):
|
||||
"""Test getting non-existent invoice returns None."""
|
||||
invoice = self.service.get_invoice(db, test_vendor.id, 99999)
|
||||
assert invoice is None
|
||||
|
||||
def test_get_invoice_or_raise_not_found(self, db, test_vendor):
|
||||
"""Test get_invoice_or_raise raises for non-existent invoice."""
|
||||
with pytest.raises(InvoiceNotFoundException):
|
||||
self.service.get_invoice_or_raise(db, test_vendor.id, 99999)
|
||||
|
||||
def test_get_invoice_wrong_vendor(self, db, test_vendor, test_invoice_settings):
|
||||
"""Test cannot get invoice from different vendor."""
|
||||
# Create an invoice
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
# Try to get with different vendor ID
|
||||
result = self.service.get_invoice(db, 99999, invoice.id)
|
||||
assert result is None
|
||||
|
||||
# ==================== List Invoices Tests ====================
|
||||
|
||||
def test_list_invoices_empty(self, db, test_vendor):
|
||||
"""Test listing invoices when none exist."""
|
||||
invoices, total = self.service.list_invoices(db, test_vendor.id)
|
||||
assert invoices == []
|
||||
assert total == 0
|
||||
|
||||
def test_list_invoices_with_status_filter(
|
||||
self, db, test_vendor, test_invoice_settings
|
||||
):
|
||||
"""Test listing invoices filtered by status."""
|
||||
# Create invoices with different statuses
|
||||
for status in [InvoiceStatus.DRAFT, InvoiceStatus.ISSUED, InvoiceStatus.PAID]:
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number=f"INV-{status.value}",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=status.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
# Filter by draft
|
||||
drafts, total = self.service.list_invoices(
|
||||
db, test_vendor.id, status="draft"
|
||||
)
|
||||
assert total == 1
|
||||
assert all(inv.status == "draft" for inv in drafts)
|
||||
|
||||
def test_list_invoices_pagination(self, db, test_vendor, test_invoice_settings):
|
||||
"""Test invoice listing pagination."""
|
||||
# Create 5 invoices
|
||||
for i in range(5):
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number=f"INV0000{i+1}",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
# Get first page
|
||||
page1, total = self.service.list_invoices(
|
||||
db, test_vendor.id, page=1, per_page=2
|
||||
)
|
||||
assert len(page1) == 2
|
||||
assert total == 5
|
||||
|
||||
# Get second page
|
||||
page2, _ = self.service.list_invoices(
|
||||
db, test_vendor.id, page=2, per_page=2
|
||||
)
|
||||
assert len(page2) == 2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.invoice
|
||||
class TestInvoiceServiceStatusManagement:
|
||||
"""Test suite for InvoiceService status management."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = InvoiceService()
|
||||
|
||||
def test_update_status_draft_to_issued(
|
||||
self, db, test_vendor, test_invoice_settings
|
||||
):
|
||||
"""Test updating invoice status from draft to issued."""
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
updated = self.service.update_status(
|
||||
db, test_vendor.id, invoice.id, "issued"
|
||||
)
|
||||
|
||||
assert updated.status == "issued"
|
||||
|
||||
def test_update_status_issued_to_paid(
|
||||
self, db, test_vendor, test_invoice_settings
|
||||
):
|
||||
"""Test updating invoice status from issued to paid."""
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=InvoiceStatus.ISSUED.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
updated = self.service.update_status(
|
||||
db, test_vendor.id, invoice.id, "paid"
|
||||
)
|
||||
|
||||
assert updated.status == "paid"
|
||||
|
||||
def test_update_status_cancelled_cannot_change(
|
||||
self, db, test_vendor, test_invoice_settings
|
||||
):
|
||||
"""Test that cancelled invoices cannot have status changed."""
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=InvoiceStatus.CANCELLED.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
with pytest.raises(ValidationException) as exc_info:
|
||||
self.service.update_status(db, test_vendor.id, invoice.id, "issued")
|
||||
|
||||
assert "cancelled" in str(exc_info.value).lower()
|
||||
|
||||
def test_update_status_invalid_status(
|
||||
self, db, test_vendor, test_invoice_settings
|
||||
):
|
||||
"""Test updating with invalid status raises ValidationException."""
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
with pytest.raises(ValidationException) as exc_info:
|
||||
self.service.update_status(
|
||||
db, test_vendor.id, invoice.id, "invalid_status"
|
||||
)
|
||||
|
||||
assert "Invalid status" in str(exc_info.value)
|
||||
|
||||
def test_mark_as_issued(self, db, test_vendor, test_invoice_settings):
|
||||
"""Test mark_as_issued helper method."""
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
updated = self.service.mark_as_issued(db, test_vendor.id, invoice.id)
|
||||
assert updated.status == InvoiceStatus.ISSUED.value
|
||||
|
||||
def test_mark_as_paid(self, db, test_vendor, test_invoice_settings):
|
||||
"""Test mark_as_paid helper method."""
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=InvoiceStatus.ISSUED.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
updated = self.service.mark_as_paid(db, test_vendor.id, invoice.id)
|
||||
assert updated.status == InvoiceStatus.PAID.value
|
||||
|
||||
def test_cancel_invoice(self, db, test_vendor, test_invoice_settings):
|
||||
"""Test cancel_invoice helper method."""
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number="INV00001",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=InvoiceStatus.DRAFT.value,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=10000,
|
||||
vat_amount_cents=1700,
|
||||
total_cents=11700,
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
updated = self.service.cancel_invoice(db, test_vendor.id, invoice.id)
|
||||
assert updated.status == InvoiceStatus.CANCELLED.value
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.invoice
|
||||
class TestInvoiceServiceStatistics:
|
||||
"""Test suite for InvoiceService statistics."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = InvoiceService()
|
||||
|
||||
def test_get_invoice_stats_empty(self, db, test_vendor):
|
||||
"""Test stats when no invoices exist."""
|
||||
stats = self.service.get_invoice_stats(db, test_vendor.id)
|
||||
|
||||
assert stats["total_invoices"] == 0
|
||||
assert stats["total_revenue_cents"] == 0
|
||||
assert stats["draft_count"] == 0
|
||||
assert stats["paid_count"] == 0
|
||||
|
||||
def test_get_invoice_stats_with_invoices(
|
||||
self, db, test_vendor, test_invoice_settings
|
||||
):
|
||||
"""Test stats calculation with multiple invoices."""
|
||||
# Create invoices
|
||||
statuses = [
|
||||
(InvoiceStatus.DRAFT.value, 10000),
|
||||
(InvoiceStatus.ISSUED.value, 20000),
|
||||
(InvoiceStatus.PAID.value, 30000),
|
||||
(InvoiceStatus.CANCELLED.value, 5000),
|
||||
]
|
||||
|
||||
for i, (status, total) in enumerate(statuses):
|
||||
invoice = Invoice(
|
||||
vendor_id=test_vendor.id,
|
||||
invoice_number=f"INV0000{i+1}",
|
||||
invoice_date=test_invoice_settings.created_at,
|
||||
status=status,
|
||||
seller_details={"company_name": "Test"},
|
||||
buyer_details={"name": "Buyer"},
|
||||
line_items=[],
|
||||
vat_rate=Decimal("17.00"),
|
||||
subtotal_cents=total,
|
||||
vat_amount_cents=int(total * 0.17),
|
||||
total_cents=total + int(total * 0.17),
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
stats = self.service.get_invoice_stats(db, test_vendor.id)
|
||||
|
||||
assert stats["total_invoices"] == 4
|
||||
# Revenue only counts issued and paid
|
||||
expected_revenue = 20000 + int(20000 * 0.17) + 30000 + int(30000 * 0.17)
|
||||
assert stats["total_revenue_cents"] == expected_revenue
|
||||
assert stats["draft_count"] == 1
|
||||
assert stats["paid_count"] == 1
|
||||
|
||||
|
||||
# ==================== Fixtures ====================
|
||||
|
||||
@pytest.fixture
|
||||
def test_invoice_settings(db, test_vendor):
|
||||
"""Create test invoice settings."""
|
||||
settings = VendorInvoiceSettings(
|
||||
vendor_id=test_vendor.id,
|
||||
company_name="Test Invoice Company",
|
||||
company_country="LU",
|
||||
invoice_prefix="INV",
|
||||
invoice_next_number=1,
|
||||
invoice_number_padding=5,
|
||||
)
|
||||
db.add(settings)
|
||||
db.commit()
|
||||
db.refresh(settings)
|
||||
return settings
|
||||
Reference in New Issue
Block a user