From cb2acb11fc14368155c14246fc87726ba7dfaac9 Mon Sep 17 00:00:00 2001 From: Jair Cedillo Date: Fri, 7 Aug 2026 16:58:09 -0500 Subject: [PATCH] =?UTF-8?q?test(fin):=20cobertura=20de=20cat=C3=A1logos,?= =?UTF-8?q?=20conceptos=20y=20emisor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend: los 8 endpoints de catálogo responden 200 con las semillas exactas y filtran por búsqueda; tax-regimes acota por tipo de persona; ninguna ruta de catálogo acepta escritura (405); sync_catalogs es idempotente. CRUD de conceptos, conflicto 409 por clave ProdServ repetida en la misma empresa, la misma clave permitida en otra empresa, la baja lógica liberándola, aislamiento multi-tenant, upsert del emisor sin duplicar filas y RFC inválido rechazado. También que una partida con concept_id hereda la descripción y que las facturas sin claves del SAT siguen listándose y generando PDF. El fixture de pruebas siembra los catálogos con la misma función que usa la migración, sobre el schema sat mapeado a SQLite. Frontend: prueba del cacheo del cliente de catálogos. RFC dummy XAXX010101000 en todas las pruebas: sin datos reales. Co-Authored-By: Claude Opus 5 (1M context) --- backend/tests/conftest.py | 10 +- backend/tests/test_fin_sat_catalogs.py | 331 ++++++++++++++++++++++ frontend/src/lib/api/fin/catalogs.test.ts | 62 ++++ 3 files changed, 402 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_fin_sat_catalogs.py create mode 100644 frontend/src/lib/api/fin/catalogs.test.ts diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 70d2eaa..7881476 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -37,9 +37,13 @@ import api.v1.modules.crm.quotes.models # noqa: E402,F401 import api.v1.modules.crm.service_requests.models # noqa: E402,F401 import api.v1.modules.crm.suppliers.models # noqa: E402,F401 import api.v1.modules.ops.shipments.models # noqa: E402,F401 +import api.v1.modules.fin.catalogs.models # noqa: E402,F401 +import api.v1.modules.fin.concepts.models # noqa: E402,F401 +import api.v1.modules.fin.issuer.models # noqa: E402,F401 import api.v1.modules.fin.invoices.models # noqa: E402,F401 +from api.v1.modules.fin.catalogs.seed_data import sync_catalogs # noqa: E402 -_SCHEMA_MAP = {"crm": None, "core": None, "ops": None, "fin": None} +_SCHEMA_MAP = {"crm": None, "core": None, "ops": None, "fin": None, "sat": None} # Tabla mínima core.tenants para resolver la FK tenant_id de las tablas crm. # En CI (PostgreSQL) la tabla real la crea la migración inicial del core. @@ -75,6 +79,10 @@ def db(): Base.metadata.create_all(engine) session_factory = sessionmaker(bind=engine, future=True) session = session_factory() + # Los catálogos del SAT los siembra la migración en PostgreSQL; aquí se replica + # con la misma función para que conceptos y emisor tengan claves que referenciar. + sync_catalogs(session.connection()) + session.commit() try: yield session finally: diff --git a/backend/tests/test_fin_sat_catalogs.py b/backend/tests/test_fin_sat_catalogs.py new file mode 100644 index 0000000..8c3e9d9 --- /dev/null +++ b/backend/tests/test_fin_sat_catalogs.py @@ -0,0 +1,331 @@ +"""Pruebas de los catálogos del SAT, el catálogo de conceptos y los datos fiscales +del emisor (módulo fin). + +Cubren: lectura de los 8 catálogos y su filtrado, que no acepten escritura, el CRUD de +conceptos con la relación 1:1 contra c_ClaveProdServ, el aislamiento multi-tenant, el +upsert del emisor y el amarre de las partidas de factura al catálogo de conceptos. + +Los RFC de las pruebas son dummies (XAXX010101000): nunca datos reales. +""" +from decimal import Decimal + +import pytest +import sqlalchemy as sa +from fastapi import FastAPI, HTTPException +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.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 +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 +from core.database import get_core_db +from core.security import get_current_user + +T, C = 1, 1 +OTHER_TENANT, OTHER_COMPANY = 2, 2 +RFC_DUMMY = "XAXX010101000" + + +@pytest.fixture() +def client(db): + """App mínima con solo el router de catálogos: evita levantar auth y permisos.""" + app = FastAPI() + app.include_router(catalogs_router, prefix="/fin") + app.dependency_overrides[get_core_db] = lambda: db + app.dependency_overrides[get_current_user] = lambda: {"sub": "tester", "tenant_id": T} + return TestClient(app) + + +def _product_service(db, code: str = "78101600") -> ProductService: + return db.query(ProductService).filter(ProductService.code == code).one() + + +def _concept_payload(db, code: str = "FLETE-MAR", ps_code: str = "78101600") -> ConceptCreate: + return ConceptCreate( + code=code, + description="Flete marítimo internacional", + product_service_id=_product_service(db, ps_code).id, + unit_price=Decimal("1500.00"), + ) + + +# ---------- Catálogos del SAT: lectura ---------- + +CATALOG_EXPECTATIONS = [ + ("tax-regimes", 19, "601"), + ("taxes", 3, "002"), + ("payment-forms", 22, "03"), + ("units-of-measure", 21, "H87"), + ("products-services", 11, "78101500"), + ("voucher-types", 5, "I"), + ("payment-methods", 2, "PUE"), + ("tax-objects", 4, "02"), +] + + +@pytest.mark.parametrize("path,expected_count,sample_code", CATALOG_EXPECTATIONS) +def test_catalog_endpoints_return_seeded_rows(client, path, expected_count, sample_code): + res = client.get(f"/fin/catalogs/{path}") + assert res.status_code == 200 + rows = res.json() + assert len(rows) == expected_count + assert sample_code in [r["code"] for r in rows] + + +def test_catalog_search_filters_by_code_or_description(client): + by_code = client.get("/fin/catalogs/payment-forms", params={"search": "03"}).json() + assert [r["code"] for r in by_code] == ["03"] + + by_description = client.get("/fin/catalogs/payment-forms", params={"search": "transferencia"}).json() + assert [r["code"] for r in by_description] == ["03"] + + prodserv = client.get("/fin/catalogs/products-services", params={"search": "marítimo"}).json() + assert [r["code"] for r in prodserv] == ["78101600"] + + +def test_tax_regimes_person_type_excludes_individual_only(client): + moral = client.get("/fin/catalogs/tax-regimes", params={"person_type": "moral"}).json() + codes = [r["code"] for r in moral] + assert "601" in codes # General de Ley Personas Morales + assert "605" not in codes # Sueldos y Salarios: solo persona física + assert all(r["applies_to_legal_entity"] for r in moral) + + fisica = client.get("/fin/catalogs/tax-regimes", params={"person_type": "fisica"}).json() + fisica_codes = [r["code"] for r in fisica] + assert "605" in fisica_codes and "601" not in fisica_codes + + +def test_products_services_limit_caps_results(client): + assert len(client.get("/fin/catalogs/products-services", params={"limit": 3}).json()) == 3 + assert client.get("/fin/catalogs/products-services", params={"limit": 500}).status_code == 422 + + +def test_catalogs_are_read_only(client): + """Los catálogos del SAT no exponen métodos de escritura.""" + for method, path in [ + ("post", "/fin/catalogs/payment-forms"), + ("put", "/fin/catalogs/tax-regimes"), + ("patch", "/fin/catalogs/units-of-measure"), + ("delete", "/fin/catalogs/products-services"), + ]: + res = client.request(method.upper(), path, json={"code": "XX", "description": "Inventado"}) + assert res.status_code == 405, f"{method.upper()} {path} no debería aceptarse" + + +def _catalog_counts(db) -> dict[str, int]: + return { + table.name: db.execute(sa.select(sa.func.count()).select_from(table)).scalar() + for table, _ in CATALOGS + } + + +def test_sync_catalogs_is_idempotent(db): + """Volver a correrla no duplica ni borra filas.""" + before = _catalog_counts(db) + inserted = sync_catalogs(db.connection()) # el fixture ya sembró los catálogos + db.commit() + assert sum(inserted.values()) == 0 + assert _catalog_counts(db) == before + + +# ---------- Conceptos ---------- + +def test_concept_crud(db): + created = concepts_service.create_concept(db, _concept_payload(db), T, C, "tester") + assert created.code == "FLETE-MAR" and created.currency == "MXN" and created.is_active + + fetched = concepts_service.get_concept(db, created.id, T, C) + assert fetched.product_service.code == "78101600" # catálogo resuelto sin N+1 + + updated = concepts_service.update_concept( + db, created.id, ConceptUpdate(description="Flete marítimo FCL", is_active=False), T, C, "tester" + ) + assert updated.description == "Flete marítimo FCL" and updated.is_active is False + + assert concepts_service.get_concepts(db, T, C, active_only=False) == [updated] + assert concepts_service.get_concepts(db, T, C, active_only=True) == [] + + concepts_service.delete_concept(db, created.id, T, C) + assert concepts_service.get_concepts(db, T, C) == [] + with pytest.raises(HTTPException) as exc: + concepts_service.get_concept(db, created.id, T, C) + assert exc.value.status_code == 404 + + +def test_duplicate_product_service_in_same_company_conflicts(db): + concepts_service.create_concept(db, _concept_payload(db), T, C) + with pytest.raises(HTTPException) as exc: + concepts_service.create_concept(db, _concept_payload(db, code="OTRO-CODIGO"), T, C) + assert exc.value.status_code == 409 + assert "producto/servicio" in exc.value.detail + + +def test_duplicate_concept_code_in_same_company_conflicts(db): + concepts_service.create_concept(db, _concept_payload(db), T, C) + with pytest.raises(HTTPException) as exc: + concepts_service.create_concept(db, _concept_payload(db, ps_code="78101500"), T, C) + assert exc.value.status_code == 409 + assert "clave 'FLETE-MAR'" in exc.value.detail + + +def test_same_product_service_allowed_in_another_company(db): + concepts_service.create_concept(db, _concept_payload(db), T, C) + other = concepts_service.create_concept(db, _concept_payload(db), T, OTHER_COMPANY) + assert other.company_id == OTHER_COMPANY + assert other.product_service_id == _product_service(db).id + + +def test_soft_deleted_concept_frees_its_product_service(db): + first = concepts_service.create_concept(db, _concept_payload(db), T, C) + concepts_service.delete_concept(db, first.id, T, C) + reused = concepts_service.create_concept(db, _concept_payload(db), T, C) + assert reused.id != first.id + assert reused.product_service_id == first.product_service_id + + +def test_concept_is_isolated_by_tenant(db): + other_tenant_concept = concepts_service.create_concept(db, _concept_payload(db), OTHER_TENANT, C) + assert concepts_service.get_concepts(db, T, C) == [] + with pytest.raises(HTTPException) as exc: + concepts_service.get_concept(db, other_tenant_concept.id, T, C) + assert exc.value.status_code == 404 + with pytest.raises(HTTPException) as exc: + concepts_service.update_concept( + db, other_tenant_concept.id, ConceptUpdate(description="Ajeno"), T, C + ) + assert exc.value.status_code == 404 + + +def test_concept_rejects_unknown_sat_key(db): + payload = _concept_payload(db) + payload.product_service_id = 999999 + with pytest.raises(HTTPException) as exc: + concepts_service.create_concept(db, payload, T, C) + assert exc.value.status_code == 422 + + +# ---------- Datos fiscales del emisor ---------- + +def _issuer_payload(db, legal_name: str = "Empresa Demo SA de CV") -> IssuerSettingsInput: + regime = db.query(TaxRegime).filter(TaxRegime.code == "601").one() + return IssuerSettingsInput( + legal_name=legal_name, rfc=RFC_DUMMY, tax_regime_id=regime.id, zip_code="64000" + ) + + +def test_issuer_settings_upsert_keeps_one_row_per_company(db): + created = issuer_service.save_issuer_settings(db, _issuer_payload(db), T, C, "tester") + assert created.rfc == RFC_DUMMY + + updated = issuer_service.save_issuer_settings( + db, _issuer_payload(db, legal_name="Empresa Demo Renombrada SA de CV"), T, C, "tester" + ) + assert updated.id == created.id + assert updated.legal_name == "Empresa Demo Renombrada SA de CV" + + rows = db.query(IssuerSettings).filter( + IssuerSettings.tenant_id == T, IssuerSettings.company_id == C, IssuerSettings.deleted_at.is_(None) + ).all() + assert len(rows) == 1 + + +def test_issuer_settings_missing_returns_404(db): + with pytest.raises(HTTPException) as exc: + issuer_service.get_issuer_settings(db, T, C) + assert exc.value.status_code == 404 + + +def test_issuer_rfc_is_validated_and_normalized(db): + regime = db.query(TaxRegime).filter(TaxRegime.code == "601").one() + with pytest.raises(ValidationError): + IssuerSettingsInput(legal_name="Demo", rfc="RFC-INVALIDO", tax_regime_id=regime.id) + with pytest.raises(ValidationError): + IssuerSettingsInput(legal_name="Demo", rfc=RFC_DUMMY, tax_regime_id=regime.id, zip_code="123") + + normalized = IssuerSettingsInput( + legal_name="Demo", rfc=" xaxx010101000 ", tax_regime_id=regime.id + ) + assert normalized.rfc == RFC_DUMMY + + +def test_issuer_rejects_unknown_tax_regime(db): + payload = _issuer_payload(db) + payload.tax_regime_id = 999999 + with pytest.raises(HTTPException) as exc: + issuer_service.save_issuer_settings(db, payload, T, C) + assert exc.value.status_code == 422 + + +# ---------- Amarre con las facturas ---------- + +def test_invoice_item_inherits_concept_description(db): + concept = concepts_service.create_concept(db, _concept_payload(db), T, C) + invoice = invoices_service.create_invoice(db, InvoiceCreate(reference="F-SAT-1"), T, C) + item = invoices_service.create_item( + db, + InvoiceItemCreate(invoice_id=invoice.id, concept_id=concept.id, quantity=1, unit_amount=1500), + T, + C, + ) + assert item.concept == concept.description # copiada del catálogo para el PDF + assert item.concept_id == concept.id + + # Si el cliente sí manda el texto, se respeta tal cual. + explicit = invoices_service.create_item( + db, + InvoiceItemCreate( + invoice_id=invoice.id, concept_id=concept.id, concept="Flete a la medida", unit_amount=100 + ), + T, + C, + ) + assert explicit.concept == "Flete a la medida" + + +def test_invoice_item_without_concept_or_catalog_is_rejected(db): + invoice = invoices_service.create_invoice(db, InvoiceCreate(reference="F-SAT-2"), T, C) + with pytest.raises(HTTPException) as exc: + invoices_service.create_item(db, InvoiceItemCreate(invoice_id=invoice.id, unit_amount=10), T, C) + assert exc.value.status_code == 422 + + +def test_invoice_item_rejects_concept_from_another_company(db): + concept = concepts_service.create_concept(db, _concept_payload(db), T, OTHER_COMPANY) + invoice = invoices_service.create_invoice(db, InvoiceCreate(reference="F-SAT-3"), T, C) + with pytest.raises(HTTPException) as exc: + invoices_service.create_item( + db, InvoiceItemCreate(invoice_id=invoice.id, concept_id=concept.id, unit_amount=10), T, C + ) + assert exc.value.status_code == 422 + + +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 = {} + monkeypatch.setattr( + "core.storage_s3.put_object_bytes", + lambda key, body, content_type="": stored.update({"key": key, "len": len(body)}), + ) + account = accounts_service.create_account(db, AccountCreate(name="Cliente heredado"), T, C) + invoice = invoices_service.create_invoice( + db, InvoiceCreate(reference="F-LEGACY", account_id=account.id, tax_rate=Decimal("16")), T, C + ) + invoices_service.create_item( + db, InvoiceItemCreate(invoice_id=invoice.id, concept="flete_internacional", unit_amount=1000), T, C + ) + assert invoice.voucher_type_id is None and invoice.payment_form_id is None + + listed = invoices_service.get_invoices(db, T, C) + assert invoice.id in [i.id for i in listed] + + sent = invoices_service.send_invoice(db, invoice.id, T, C) + assert sent.status == "enviada" and stored["len"] > 0 diff --git a/frontend/src/lib/api/fin/catalogs.test.ts b/frontend/src/lib/api/fin/catalogs.test.ts new file mode 100644 index 0000000..29ab94f --- /dev/null +++ b/frontend/src/lib/api/fin/catalogs.test.ts @@ -0,0 +1,62 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +const get = vi.fn(); + +// El cliente de catálogos solo usa `api.get`; se sustituye para contar peticiones. +vi.mock('$lib/api', () => ({ api: { get } })); + +const { satCatalogsAPI, clearCatalogCache } = await import('./catalogs'); + +const COMPANY_ID = 1; +const PAYMENT_FORMS = [ + { id: 1, code: '01', description: 'Efectivo', is_active: true }, + { id: 2, code: '03', description: 'Transferencia electrónica de fondos', is_active: true } +]; + +describe('satCatalogsAPI — cacheo en memoria', () => { + beforeEach(() => { + clearCatalogCache(); + get.mockReset(); + get.mockResolvedValue({ data: PAYMENT_FORMS, status: 200 }); + }); + + it('consulta el backend la primera vez y reusa el cache después', async () => { + const first = await satCatalogsAPI.paymentForms(COMPANY_ID); + const second = await satCatalogsAPI.paymentForms(COMPANY_ID); + + expect(first).toEqual(PAYMENT_FORMS); + expect(second).toBe(first); // misma referencia: vino del cache + expect(get).toHaveBeenCalledTimes(1); + }); + + it('cachea por separado cada combinación de parámetros', async () => { + await satCatalogsAPI.paymentForms(COMPANY_ID); + await satCatalogsAPI.paymentForms(COMPANY_ID, { search: 'transferencia' }); + await satCatalogsAPI.paymentForms(COMPANY_ID, { search: 'transferencia' }); + + expect(get).toHaveBeenCalledTimes(2); + }); + + it('no comparte cache entre compañías', async () => { + await satCatalogsAPI.paymentForms(COMPANY_ID); + await satCatalogsAPI.paymentForms(2); + + expect(get).toHaveBeenCalledTimes(2); + }); + + it('clearCatalogCache obliga a volver a consultar', async () => { + await satCatalogsAPI.paymentForms(COMPANY_ID); + clearCatalogCache(); + await satCatalogsAPI.paymentForms(COMPANY_ID); + + expect(get).toHaveBeenCalledTimes(2); + }); + + it('propaga el error del backend y no lo cachea', async () => { + get.mockResolvedValueOnce({ error: 'Falla del servidor', status: 500 }); + await expect(satCatalogsAPI.taxRegimes(COMPANY_ID)).rejects.toThrow('Falla del servidor'); + + await satCatalogsAPI.taxRegimes(COMPANY_ID); + expect(get).toHaveBeenCalledTimes(2); + }); +});