T2026-08-047 feat(fin): catálogos SAT, conceptos de facturación y datos fiscales del emisor #5
@@ -332,12 +332,21 @@ def _get_item(db, item_id, tenant_id, company_id) -> InvoiceItem:
|
||||
return obj
|
||||
|
||||
|
||||
def _resolve_item_concept(db, data: dict, tenant_id, company_id) -> None:
|
||||
"""Completa ``concept`` a partir del concepto del catálogo cuando no se envió.
|
||||
# Claves del SAT que la partida hereda del concepto del catálogo cuando no se envían.
|
||||
_CONCEPT_INHERITED_FIELDS = ("product_service_id", "unit_of_measure_id", "tax_object_id")
|
||||
|
||||
El PDF de la factura sigue leyendo la columna de texto libre ``concept``, así que
|
||||
al capturar por catálogo se hereda ahí la descripción del concepto (recortada al
|
||||
largo de la columna).
|
||||
|
||||
def _resolve_item_concept(db, data: dict, tenant_id, company_id) -> None:
|
||||
"""Completa la partida a partir del concepto del catálogo.
|
||||
|
||||
Hereda dos cosas cuando el cliente no las manda:
|
||||
|
||||
- ``concept``: el PDF de la factura sigue leyendo esa columna de texto libre, así
|
||||
que ahí va la descripción del concepto (recortada al largo de la columna).
|
||||
- Las claves fiscales (``product_service_id``, ``unit_of_measure_id``,
|
||||
``tax_object_id``): sin ellas la partida capturada por catálogo quedaría
|
||||
incompleta para el CFDI. Lo que el cliente sí envía manda sobre el catálogo,
|
||||
para poder facturar una partida con una unidad distinta a la del concepto.
|
||||
"""
|
||||
concept_id = data.get("concept_id")
|
||||
if concept_id is not None:
|
||||
@@ -352,6 +361,9 @@ def _resolve_item_concept(db, data: dict, tenant_id, company_id) -> None:
|
||||
)
|
||||
if not data.get("concept"):
|
||||
data["concept"] = catalog_concept.description[:60]
|
||||
for field in _CONCEPT_INHERITED_FIELDS:
|
||||
if data.get(field) is None:
|
||||
data[field] = getattr(catalog_concept, field)
|
||||
if not data.get("concept"):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
@@ -374,7 +386,12 @@ def create_item(db, payload: InvoiceItemCreate, tenant_id, company_id) -> Invoic
|
||||
|
||||
def update_item(db, item_id, payload: InvoiceItemUpdate, tenant_id, company_id) -> InvoiceItem:
|
||||
item = _get_item(db, item_id, tenant_id, company_id)
|
||||
for f, v in payload.model_dump(exclude_unset=True).items():
|
||||
data = payload.model_dump(exclude_unset=True)
|
||||
# Cambiar el concepto del catálogo revalida la referencia y vuelve a heredar
|
||||
# descripción y claves fiscales del concepto nuevo.
|
||||
if data.get("concept_id") is not None:
|
||||
_resolve_item_concept(db, data, tenant_id, company_id)
|
||||
for f, v in data.items():
|
||||
setattr(item, f, v)
|
||||
db.flush()
|
||||
_recompute(db, get_invoice(db, item.invoice_id, tenant_id, company_id))
|
||||
|
||||
@@ -16,14 +16,19 @@ from fastapi.testclient import TestClient
|
||||
from pydantic import ValidationError
|
||||
|
||||
from api.v1.modules.crm.accounts import service as accounts_service
|
||||
from api.v1.modules.crm.accounts.dto import AccountCreate
|
||||
from api.v1.modules.fin.catalogs.models import ProductService, TaxRegime
|
||||
from api.v1.modules.crm.accounts.dto import AccountCreate, AccountUpdate
|
||||
from api.v1.modules.fin.catalogs.models import CfdiUse, ProductService, TaxObject, TaxRegime, UnitOfMeasure
|
||||
from api.v1.modules.fin.catalogs.routes import router as catalogs_router
|
||||
from api.v1.modules.fin.catalogs.seed_data import CATALOGS, sync_catalogs
|
||||
from api.v1.modules.fin.concepts import service as concepts_service
|
||||
from api.v1.modules.fin.concepts.dto import ConceptCreate, ConceptUpdate
|
||||
from api.v1.modules.fin.invoices import service as invoices_service
|
||||
from api.v1.modules.fin.invoices.dto import InvoiceCreate, InvoiceItemCreate, InvoiceItemResponse
|
||||
from api.v1.modules.fin.invoices.dto import (
|
||||
InvoiceCreate,
|
||||
InvoiceItemCreate,
|
||||
InvoiceItemResponse,
|
||||
InvoiceItemUpdate,
|
||||
)
|
||||
from api.v1.modules.fin.issuer import service as issuer_service
|
||||
from api.v1.modules.fin.issuer.dto import IssuerSettingsInput
|
||||
from api.v1.modules.fin.issuer.models import IssuerSettings
|
||||
@@ -54,6 +59,8 @@ def _concept_payload(db, code: str = "FLETE-MAR", ps_code: str = "78101600") ->
|
||||
code=code,
|
||||
description="Flete marítimo internacional",
|
||||
product_service_id=_product_service(db, ps_code).id,
|
||||
unit_of_measure_id=db.query(UnitOfMeasure).filter(UnitOfMeasure.code == "E48").one().id,
|
||||
tax_object_id=db.query(TaxObject).filter(TaxObject.code == "02").one().id,
|
||||
unit_price=Decimal("1500.00"),
|
||||
)
|
||||
|
||||
@@ -69,6 +76,7 @@ CATALOG_EXPECTATIONS = [
|
||||
("voucher-types", 5, "I"),
|
||||
("payment-methods", 2, "PUE"),
|
||||
("tax-objects", 4, "02"),
|
||||
("cfdi-uses", 24, "G03"),
|
||||
]
|
||||
|
||||
|
||||
@@ -314,6 +322,105 @@ def test_invoice_item_rejects_concept_from_another_company(db):
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
|
||||
def test_invoice_item_inherits_sat_keys_from_concept(db):
|
||||
"""La partida hereda las claves fiscales del concepto para quedar completa (CFDI)."""
|
||||
concept = concepts_service.create_concept(db, _concept_payload(db), T, C)
|
||||
invoice = invoices_service.create_invoice(db, InvoiceCreate(reference="F-SAT-4"), T, C)
|
||||
|
||||
item = invoices_service.create_item(
|
||||
db, InvoiceItemCreate(invoice_id=invoice.id, concept_id=concept.id, unit_amount=1500), T, C
|
||||
)
|
||||
assert item.product_service_id == concept.product_service_id
|
||||
assert item.unit_of_measure_id == concept.unit_of_measure_id
|
||||
assert item.tax_object_id == concept.tax_object_id
|
||||
|
||||
|
||||
def test_invoice_item_sat_keys_sent_by_client_win_over_concept(db):
|
||||
"""Lo que el cliente envía manda: permite facturar con otra unidad de medida."""
|
||||
concept = concepts_service.create_concept(db, _concept_payload(db), T, C)
|
||||
invoice = invoices_service.create_invoice(db, InvoiceCreate(reference="F-SAT-5"), T, C)
|
||||
other_unit = db.query(UnitOfMeasure).filter(UnitOfMeasure.code == "KGM").one()
|
||||
|
||||
item = invoices_service.create_item(
|
||||
db,
|
||||
InvoiceItemCreate(
|
||||
invoice_id=invoice.id, concept_id=concept.id, unit_of_measure_id=other_unit.id, unit_amount=10
|
||||
),
|
||||
T,
|
||||
C,
|
||||
)
|
||||
assert item.unit_of_measure_id == other_unit.id
|
||||
assert item.product_service_id == concept.product_service_id # el resto sí se hereda
|
||||
|
||||
|
||||
def test_changing_item_concept_reinherits_keys(db):
|
||||
"""Cambiar el concepto de una partida revalida y vuelve a heredar del nuevo."""
|
||||
first = concepts_service.create_concept(db, _concept_payload(db), T, C)
|
||||
second = concepts_service.create_concept(
|
||||
db, _concept_payload(db, code="DESPACHO", ps_code="78141600"), T, C
|
||||
)
|
||||
invoice = invoices_service.create_invoice(db, InvoiceCreate(reference="F-SAT-6"), T, C)
|
||||
item = invoices_service.create_item(
|
||||
db, InvoiceItemCreate(invoice_id=invoice.id, concept_id=first.id, unit_amount=100), T, C
|
||||
)
|
||||
|
||||
updated = invoices_service.update_item(
|
||||
db, item.id, InvoiceItemUpdate(concept_id=second.id), T, C
|
||||
)
|
||||
assert updated.concept_id == second.id
|
||||
assert updated.product_service_id == second.product_service_id
|
||||
assert updated.concept == second.description
|
||||
|
||||
|
||||
def test_updating_item_rejects_concept_from_another_tenant(db):
|
||||
"""El PATCH valida la referencia igual que el alta: no cruza tenants."""
|
||||
mine = concepts_service.create_concept(db, _concept_payload(db), T, C)
|
||||
alien = concepts_service.create_concept(db, _concept_payload(db), OTHER_TENANT, C)
|
||||
invoice = invoices_service.create_invoice(db, InvoiceCreate(reference="F-SAT-7"), T, C)
|
||||
item = invoices_service.create_item(
|
||||
db, InvoiceItemCreate(invoice_id=invoice.id, concept_id=mine.id, unit_amount=100), T, C
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
invoices_service.update_item(db, item.id, InvoiceItemUpdate(concept_id=alien.id), T, C)
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
|
||||
# ---------- Claves fiscales del receptor (crm.accounts) ----------
|
||||
|
||||
def test_account_accepts_sat_fiscal_keys(db):
|
||||
regime = db.query(TaxRegime).filter(TaxRegime.code == "601").one()
|
||||
cfdi_use = db.query(CfdiUse).filter(CfdiUse.code == "G03").one()
|
||||
|
||||
account = accounts_service.create_account(
|
||||
db,
|
||||
AccountCreate(name="Cliente fiscal", tax_regime_id=regime.id, cfdi_use_id=cfdi_use.id),
|
||||
T,
|
||||
C,
|
||||
)
|
||||
assert account.tax_regime_id == regime.id and account.cfdi_use_id == cfdi_use.id
|
||||
|
||||
|
||||
def test_account_rejects_unknown_sat_fiscal_keys(db):
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
accounts_service.create_account(db, AccountCreate(name="Cliente malo", cfdi_use_id=999999), T, C)
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
account = accounts_service.create_account(db, AccountCreate(name="Cliente ok"), T, C)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
accounts_service.update_account(db, account.id, AccountUpdate(tax_regime_id=999999), T, C)
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
|
||||
def test_account_free_text_fiscal_fields_are_preserved(db):
|
||||
"""El texto libre previo se conserva: las FK lo complementan, no lo sustituyen."""
|
||||
account = accounts_service.create_account(
|
||||
db, AccountCreate(name="Cliente heredado", tax_regime="601", cfdi_use="G03"), T, C
|
||||
)
|
||||
assert account.tax_regime == "601" and account.cfdi_use == "G03"
|
||||
assert account.tax_regime_id is None and account.cfdi_use_id is None
|
||||
|
||||
|
||||
def test_legacy_invoices_keep_working_without_sat_fields(db, monkeypatch):
|
||||
"""Las facturas previas, sin claves del SAT, siguen listándose y generando PDF."""
|
||||
stored = {}
|
||||
|
||||
Reference in New Issue
Block a user