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:

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

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);
+	});
+});