diff --git a/backend/api/v1/modules/a76/invoices/exports/process/main_process.py b/backend/api/v1/modules/a76/invoices/exports/process/main_process.py index 91c96821..2ee2991d 100644 --- a/backend/api/v1/modules/a76/invoices/exports/process/main_process.py +++ b/backend/api/v1/modules/a76/invoices/exports/process/main_process.py @@ -11,6 +11,7 @@ from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRat from core.exceptions import ErrorCollector from .pre_validators import pre_validators from .sub_process.review_exchange_rate import review_exchange_rate +from .sub_process.review_class import review_class from .sub_process.assign_values import assign_values from .sub_process.assign_no_discharges import assign_no_discharges_items, assign_no_discharges_series @@ -38,6 +39,9 @@ def main_process(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id lines = pre_process(db, invoice, tenant_id, company_id, errors) errors.raise_if_errors() + review_class(db, lines, errors) + errors.raise_if_errors() + settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) q_settings = settings.get("qsisgen", {}) s_settings = settings.get("ssisgen", {}) diff --git a/backend/tests/unit/invoices/test_actseguridad_exports.py b/backend/tests/unit/invoices/test_actseguridad_exports.py deleted file mode 100644 index 15b17b1a..00000000 --- a/backend/tests/unit/invoices/test_actseguridad_exports.py +++ /dev/null @@ -1,211 +0,0 @@ -""" -Tests para la resolución de 'act_seguridad' en el proceso de exportación. - -Verifica que: - 1. _assign_invoice_totals llama a AuditService.create_audit_log cuando act_seguridad=1. - 2. _assign_invoice_totals NO llama a AuditService cuando act_seguridad=0. - 3. _assign_invoice_totals NO llama a AuditService cuando actseguridad está ausente. - 4. exports/main_process emite log AUDIT_DEBUG con el valor resuelto (paridad con imports). - 5. Fallback qsisgen: si ssisgen no tiene actseguridad, se lee de qsisgen. -""" -import logging -from decimal import Decimal -from types import SimpleNamespace -from unittest.mock import MagicMock, patch - -from core.exceptions import ErrorCollector - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - -def _mk_invoice_expo(invoice_number: str = "EXP-TEST-01", invoice_type: str = "DONAC"): - fin = SimpleNamespace( - exchange_rate=Decimal("17.25"), - exchange_rate_mm=None, - currency="ME", - value_mn=None, - value_me=None, - total_quantity=None, - total_packages=None, - net_weight=None, - gross_weight=None, - ) - cmx = SimpleNamespace(pedimento=None, which_exchange_rate=None) - return SimpleNamespace( - id=1, - invoice_number=invoice_number, - invoice_type=invoice_type, - operation_type="exp", - financials=fin, - compliance_mx=cmx, - status=None, - party_count=None, - updated_date=None, - process_log=None, - updated_by=None, - ) - - -def _mk_lines_expo(qty: Decimal = Decimal("4")): - """Línea de exportación con todos los atributos que _assign_invoice_totals necesita.""" - line = SimpleNamespace( - line_number=1, - quantity=SimpleNamespace( - quantity=qty, - net_weight=Decimal("10"), - gross_weight=Decimal("11"), - package_quantity=1, - ), - financial=SimpleNamespace(value_mxn=Decimal("500"), value_usd=Decimal("30")), - description=SimpleNamespace(has_serial=False), - # fa_data.is_subitem es necesario en _assign_invoice_totals para party_count - fa_data=SimpleNamespace(is_subitem=False, discharge=False), - ) - return [line] - - -def _settings_actseguridad(value: int) -> dict: - return { - "ssisgen": {"actseguridad": value}, - "qsisgen": {}, - "invoices": {}, - } - - -# --------------------------------------------------------------------------- -# Tests: _assign_invoice_totals → AuditService -# --------------------------------------------------------------------------- - -class TestAssignInvoiceTotalsAudit: - """ - _assign_invoice_totals (en finalize_invoice.py) resuelve act_seguridad desde - los settings y llama condicionalmente a AuditService.create_audit_log. - """ - - # El AuditService se importa localmente dentro del `if act_seguridad == 1:` - # por eso el patch va al módulo fuente, no al de uso. - _AUDIT_PATCH = "api.v1.modules.a76.audit_log.services.service.AuditService" - _SETTINGS_PATCH = ( - "api.v1.modules.a76.invoices.exports.process" - ".sub_process.finalize_invoice.AppSettingsService.get_resolved_settings" - ) - # ExchangeRate select (usado internamente cuando cal_val_base_tc=1, irrelevante aquí) - _SELECT_PATCH = ( - "api.v1.modules.a76.invoices.exports.process" - ".sub_process.finalize_invoice.db.execute" - ) - - def _call_assign(self, invoice, lines, settings: dict, username: str = "TESTUSER"): - from api.v1.modules.a76.invoices.exports.process.sub_process.finalize_invoice import ( - _assign_invoice_totals, - ) - db = MagicMock() - with ( - patch(self._SETTINGS_PATCH, return_value=settings), - patch(self._AUDIT_PATCH) as mock_audit, - ): - _assign_invoice_totals(db, invoice, lines, "1", "1", username=username) - return mock_audit - - def test_actseguridad_1_llama_audit_service(self): - """act_seguridad=1 → AuditService.create_audit_log invocado con datos de exportación.""" - invoice = _mk_invoice_expo() - lines = _mk_lines_expo() - mock_audit = self._call_assign(invoice, lines, _settings_actseguridad(1), "JOSE") - - mock_audit.create_audit_log.assert_called_once() - kw = mock_audit.create_audit_log.call_args.kwargs - assert kw["movement"] == "EXPORTACION" - assert kw["username"] == "JOSE" - assert kw["procedure"] == "ACTUALIZAR FACTURA" - assert kw["operation_type"] == "UPDATE" - - def test_actseguridad_0_no_llama_audit_service(self): - """act_seguridad=0 → AuditService.create_audit_log NO se llama.""" - invoice = _mk_invoice_expo() - lines = _mk_lines_expo() - mock_audit = self._call_assign(invoice, lines, _settings_actseguridad(0)) - mock_audit.create_audit_log.assert_not_called() - - def test_actseguridad_ausente_no_llama_audit_service(self): - """ssisgen sin 'actseguridad' (default 0) → AuditService NO se llama.""" - settings = {"ssisgen": {}, "qsisgen": {}, "invoices": {}} - invoice = _mk_invoice_expo() - lines = _mk_lines_expo() - mock_audit = self._call_assign(invoice, lines, settings) - mock_audit.create_audit_log.assert_not_called() - - -# --------------------------------------------------------------------------- -# Tests: main_process exports → log AUDIT_DEBUG y fallback a qsisgen -# --------------------------------------------------------------------------- - -class TestExportsMainProcessActSeguridadLog: - """ - exports/process/main_process resuelve act_seguridad desde los settings y - emite el log AUDIT_DEBUG (paridad con imports/process/main_process). - """ - - def _run_main_process(self, settings: dict, caplog, invoice_number: str = "EXP-TEST-01"): - from api.v1.modules.a76.invoices.exports.process import main_process as expo_mp - - invoice = _mk_invoice_expo(invoice_number=invoice_number) - lines = _mk_lines_expo() - - db = MagicMock() - # begin_nested() debe retornar un context manager funcional - db.begin_nested.return_value.__enter__ = MagicMock(return_value=MagicMock()) - db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) - - with ( - patch.object(expo_mp, "pre_process", return_value=lines), - patch.object(expo_mp, "review_exchange_rate"), - patch.object(expo_mp, "assign_values"), - patch.object(expo_mp, "assign_no_discharges_items"), - patch.object(expo_mp, "assign_no_discharges_series"), - patch.object(expo_mp, "finalize_invoice_no_discharge"), - patch.object(expo_mp, "finalize_invoice_with_discharge"), - patch( - "api.v1.modules.a76.invoices.exports.process.main_process" - ".AppSettingsService.get_resolved_settings", - return_value=settings, - ), - caplog.at_level( - logging.INFO, - logger="api.v1.modules.a76.invoices.exports.process.main_process", - ), - ): - expo_mp.main_process(db, invoice, "1", "1", username="TESTUSER") - - def test_audit_debug_log_emitido_con_actseguridad_1(self, caplog): - """main_process emite AUDIT_DEBUG con el valor resuelto de act_seguridad.""" - self._run_main_process(_settings_actseguridad(1), caplog) - - audit_logs = [r.message for r in caplog.records if "AUDIT_DEBUG" in r.message] - assert audit_logs, "Debe emitir al menos un log con 'AUDIT_DEBUG'" - assert "act_seguridad resolve result = 1" in audit_logs[0] - assert "EXP-TEST-01" in audit_logs[0] - - def test_audit_debug_log_emitido_con_actseguridad_0(self, caplog): - """main_process emite AUDIT_DEBUG incluso cuando act_seguridad=0.""" - self._run_main_process(_settings_actseguridad(0), caplog, "EXP-TEST-02") - - audit_logs = [r.message for r in caplog.records if "AUDIT_DEBUG" in r.message] - assert audit_logs, "Debe emitir AUDIT_DEBUG aunque act_seguridad=0" - assert "act_seguridad resolve result = 0" in audit_logs[0] - - def test_fallback_a_qsisgen_cuando_ssisgen_esta_vacio(self, caplog): - """Si ssisgen no tiene actseguridad, debe tomarlo desde qsisgen.""" - settings = { - "ssisgen": {}, # sin actseguridad - "qsisgen": {"actseguridad": 1}, # qsisgen lo tiene - "invoices": {}, - } - self._run_main_process(settings, caplog, "EXP-TEST-03") - - audit_logs = [r.message for r in caplog.records if "AUDIT_DEBUG" in r.message] - assert audit_logs - # El valor resuelto debe ser 1 (tomado de qsisgen) - assert "act_seguridad resolve result = 1" in audit_logs[0] diff --git a/backend/tests/unit/invoices/test_validarseries_switch.py b/backend/tests/unit/invoices/test_validarseries_switch.py deleted file mode 100644 index 03be69c2..00000000 --- a/backend/tests/unit/invoices/test_validarseries_switch.py +++ /dev/null @@ -1,229 +0,0 @@ -""" -Tests para el switch maestro 'validarseries' y la validación 'cantvscantseries'. - -Comportamiento esperado del código: - - La comprobación SERIES_VACIAS (sin series cuando has_serial=True) se emite - ANTES de leer el parámetro 'validarseries'; es una validación incondicional. - - El switch maestro 'validarseries' sólo controla la validación de MISMATCH - (cantidad de series vs cantidad de la partida), que es el check 'cantvscantseries'. - -Casos cubiertos: - Importaciones (review_series): - 1. validarseries=0 + series existen + mismatch → SIN error SERIES_VS_CANT (switch off) - 2. validarseries=1 + cantvscantseries=0 + mismatch → SIN error SERIES_VS_CANT - 3. validarseries=1 + cantvscantseries=1 + cantidad correcta → sin error - 4. validarseries=1 + cantvscantseries=1 + mismatch → error SERIES_VS_CANT - 5. has_serial=True + sin series → error SERIES_VACIAS (incondicional) - 6. has_serial=False → sin validación (return inmediato) - - Exportaciones (review_qty_series): - 7. validarseries=0 + series existen + mismatch → SIN error SERIES_VS_CANT - 8. validarseries=1 + cantvscantseries=1 + mismatch → error SERIES_VS_CANT - 9. validarseries=1 + cantvscantseries=1 + match → sin error - 10. has_serial=False → sin validación -""" -from decimal import Decimal -from types import SimpleNamespace -from unittest.mock import patch - -from core.exceptions import ErrorCollector - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - -def _mk_line(qty: Decimal, has_serial: bool = True, line_id: int = 9999): - return SimpleNamespace( - id=line_id, - line_number=1, - description=SimpleNamespace(has_serial=has_serial), - quantity=SimpleNamespace(quantity=qty), - unit_of_measure="PZA", - unit_of_measure_info=SimpleNamespace(code="PZA"), - ) - - -def _mk_invoice_expo(is_regime_change: bool = False): - return SimpleNamespace( - compliance_mx=SimpleNamespace(is_regime_change=is_regime_change) - ) - - -def _settings(*, validarseries: int = 0, cantvscantseries: int = 0) -> dict: - return { - "ssisgen": { - "validarseries": validarseries, - "cantvscantseries": cantvscantseries, - }, - "qsisgen": {}, - } - - -def _db_with_series_count(count: int): - """Mock de DB que devuelve `count` al llamar .query(...).filter(...).count().""" - mock_query = SimpleNamespace( - filter=lambda *a, **kw: SimpleNamespace(count=lambda: count) - ) - return SimpleNamespace( - query=lambda *a: mock_query, - get=lambda *a, **kw: SimpleNamespace(rfc="TST010101AAA"), - ) - - -# --------------------------------------------------------------------------- -# Importaciones: review_series -# --------------------------------------------------------------------------- - -class TestReviewSeriesImports: - - def _call(self, db, line, settings: dict) -> ErrorCollector: - from api.v1.modules.a76.invoices.imports.process.sub_process.review_series import review_series - errors = ErrorCollector() - with patch( - "api.v1.modules.a76.invoices.imports.process.sub_process" - ".review_series.AppSettingsService.get_resolved_settings", - return_value=settings, - ): - review_series(db, line, "TST010101AAA", "1", "1", errors) - return errors - - # -- Switch maestro apagado -------------------------------------------------- - - def test_switch_off_no_emite_series_vs_cant_aunque_haya_mismatch(self): - """ - validarseries=0: aunque hay 3 series para una qty=5, no se emite SERIES_VS_CANT. - SERIES_VACIAS no aplica porque sí existen series (count=3). - """ - db = _db_with_series_count(3) - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, line, _settings(validarseries=0)) - codes = [e["code"] for e in errors.get_errors()] - assert "SERIES_VS_CANT" not in codes - assert not errors.has_errors() - - # -- SERIES_VACIAS es incondicional (antes del switch) ---------------------- - - def test_sin_series_emite_series_vacias_independiente_de_validarseries(self): - """ - Cuando has_serial=True pero no hay ninguna serie registrada, se emite - SERIES_VACIAS siempre, incluso con validarseries=0. - (Esta validación ocurre antes de leer el parámetro.) - """ - db = _db_with_series_count(0) - line = _mk_line(qty=Decimal("5"), has_serial=True) - # Incluso con switch apagado, SERIES_VACIAS se emite - errors = self._call(db, line, _settings(validarseries=0)) - codes = [e["code"] for e in errors.get_errors()] - assert "SERIES_VACIAS" in codes - - # -- cantvscantseries=0 (switch on, pero sub-flag off) ---------------------- - - def test_cantvscantseries_off_no_valida_mismatch(self): - """ - validarseries=1 + cantvscantseries=0: aunque series(3) != qty(5), sin error. - """ - db = _db_with_series_count(3) - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, line, _settings(validarseries=1, cantvscantseries=0)) - codes = [e["code"] for e in errors.get_errors()] - assert "SERIES_VS_CANT" not in codes - - # -- Validación activa y sin mismatch --------------------------------------- - - def test_validacion_activa_cantidad_correcta_sin_error(self): - """validarseries=1 + cantvscantseries=1 + series==qty → sin error.""" - db = _db_with_series_count(5) - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, line, _settings(validarseries=1, cantvscantseries=1)) - assert not errors.has_errors() - - # -- Validación activa con mismatch ----------------------------------------- - - def test_validacion_activa_mismatch_emite_error(self): - """validarseries=1 + cantvscantseries=1 + series(3) != qty(5) → SERIES_VS_CANT.""" - db = _db_with_series_count(3) - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, line, _settings(validarseries=1, cantvscantseries=1)) - codes = [e["code"] for e in errors.get_errors()] - assert "SERIES_VS_CANT" in codes - - # -- has_serial=False ------------------------------------------------------- - - def test_sin_lleva_serie_no_valida_nada(self): - """Líneas sin has_serial hacen return inmediato; sin errores.""" - db = _db_with_series_count(0) - line = _mk_line(qty=Decimal("5"), has_serial=False) - errors = self._call(db, line, _settings(validarseries=1, cantvscantseries=1)) - assert not errors.has_errors() - - -# --------------------------------------------------------------------------- -# Exportaciones: review_qty_series -# --------------------------------------------------------------------------- - -class TestReviewQtySeriesExports: - - def _call(self, db, invoice, lines, settings: dict) -> ErrorCollector: - from api.v1.modules.a76.invoices.exports.process.sub_process.series.review_qty_series import ( - review_qty_series, - ) - errors = ErrorCollector() - with ( - patch( - "api.v1.modules.a76.invoices.exports.process.sub_process" - ".series.review_qty_series.AppSettingsService.get_resolved_settings", - return_value=settings, - ), - patch( - "api.v1.modules.a76.invoices.exports.process.sub_process" - ".series.review_qty_series.db.get", - return_value=SimpleNamespace(rfc="TST010101AAA"), - ) if False else __import__("contextlib").nullcontext(), # get() ya en mock db - ): - review_qty_series(db, invoice, lines, "1", "1", errors) - return errors - - def test_switch_off_no_emite_series_vs_cant_aunque_haya_mismatch(self): - """validarseries=0: series(2) != qty(5) pero sin error SERIES_VS_CANT.""" - db = _db_with_series_count(2) - invoice = _mk_invoice_expo() - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, invoice, [line], _settings(validarseries=0)) - codes = [e["code"] for e in errors.get_errors()] - assert "SERIES_VS_CANT" not in codes - assert not errors.has_errors() - - def test_sin_series_emite_series_vacias_con_switch_on(self): - """validarseries=1 + sin series registradas → SERIES_VACIAS.""" - db = _db_with_series_count(0) - invoice = _mk_invoice_expo() - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, invoice, [line], _settings(validarseries=1, cantvscantseries=1)) - codes = [e["code"] for e in errors.get_errors()] - assert "SERIES_VACIAS" in codes - - def test_validacion_activa_mismatch_emite_error(self): - """validarseries=1 + cantvscantseries=1 + series(2) != qty(5) → SERIES_VS_CANT.""" - db = _db_with_series_count(2) - invoice = _mk_invoice_expo() - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, invoice, [line], _settings(validarseries=1, cantvscantseries=1)) - codes = [e["code"] for e in errors.get_errors()] - assert "SERIES_VS_CANT" in codes - - def test_validacion_activa_cantidad_correcta_sin_error(self): - """validarseries=1 + cantvscantseries=1 + series==qty → sin error.""" - db = _db_with_series_count(5) - invoice = _mk_invoice_expo() - line = _mk_line(qty=Decimal("5"), has_serial=True) - errors = self._call(db, invoice, [line], _settings(validarseries=1, cantvscantseries=1)) - assert not errors.has_errors() - - def test_sin_lleva_serie_no_valida_nada(self): - """Líneas sin has_serial se ignoran.""" - db = _db_with_series_count(0) - invoice = _mk_invoice_expo() - line = _mk_line(qty=Decimal("5"), has_serial=False) - errors = self._call(db, invoice, [line], _settings(validarseries=1, cantvscantseries=1)) - assert not errors.has_errors()