fin-hub/tests/test_accounting_adapters.py

103 lines
3.4 KiB
Python

from dataclasses import replace
from decimal import Decimal
import pytest
from fin_hub.accounting import (
AccountingCapability,
DatevDuoAdapter,
SchemaMigrations,
default_registry,
)
from fin_hub.services.billing import BillingBasisExport, BillingBasisRecord
def _export(schema_version: str = "0.1") -> BillingBasisExport:
record = BillingBasisRecord(
billing_basis_id="billing-basis:stable",
cost_attribution_key="client:acme|app:portal|instance:prod-01",
client_id="acme",
application_id="portal",
app_instance_id="prod-01",
period_month="2026-07",
currency="EUR",
price_id="price:agreement-v1",
price_revision_of=None,
price_source="agreement-v1",
revenue=Decimal("200.00"),
direct_cost=Decimal("20.00"),
allocated_cost=Decimal("80.00"),
total_cost=Decimal("100.00"),
margin=Decimal("100.00"),
financial_fact_ids=("fact:1",),
financial_fact_corrections=(),
allocation_ids=("allocation:1",),
allocation_revisions=(),
provenance=("agreement-v1",),
)
return BillingBasisExport(
schema_version=schema_version,
artifact_type="billing_basis_report",
records=(record,),
exceptions=(),
disclaimer="Reporting basis only; not an invoice.",
)
def test_datev_duo_is_default_and_prepares_only_an_incomplete_draft():
transfer = default_registry().prepare(_export())
assert transfer.adapter_id == "datev-duo-v1"
assert transfer.provider == "DATEV Unternehmen online"
assert transfer.records[0]["external_reference"] == "billing-basis:stable"
assert transfer.records[0]["net_billing_basis"] == Decimal("200.00")
assert transfer.records[0]["requires_legal_and_tax_completion"] is True
assert "Provider draft only" in transfer.disclaimer
assert not any(
key in transfer.records[0]
for key in ("invoice_number", "issue_date", "payment_status")
)
def test_qonto_requires_explicit_selection_and_keeps_datev_as_destination():
transfer = default_registry().prepare(
_export(),
adapter_id="qonto-invoice-draft-v1",
required_capabilities=frozenset(
{AccountingCapability.INVOICE_DRAFT_WORKFLOW}
),
)
assert transfer.provider == "Qonto"
assert transfer.records[0]["bookkeeping_destination"] == (
"DATEV Unternehmen online"
)
def test_capability_mismatch_never_silently_falls_back_to_qonto():
with pytest.raises(ValueError, match="lacks capabilities: invoice_draft_workflow"):
default_registry().prepare(
_export(),
required_capabilities=frozenset(
{AccountingCapability.INVOICE_DRAFT_WORKFLOW}
),
)
def test_breaking_schema_requires_an_explicit_migration():
with pytest.raises(ValueError, match="no explicit billing-basis schema migration"):
default_registry().prepare(_export("1.0"))
def test_registered_schema_migration_must_produce_declared_version():
migrations = SchemaMigrations()
migrations.register("1.0", "0.1", lambda export: replace(export, schema_version="0.1"))
from fin_hub.accounting.adapters import AdapterRegistry
registry = AdapterRegistry(
(DatevDuoAdapter(),),
default_adapter_id="datev-duo-v1",
migrations=migrations,
)
assert registry.prepare(_export("1.0")).source_schema_version == "0.1"