Merge pull request 'Arreglando la importacion que impedia avanzar el script de prueba' (#272) from feature/tareas_TODO_configuracion into development
Reviewed-on: ADUANASOFT/anexo76#272
This commit is contained in:
@@ -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", {})
|
||||
|
||||
@@ -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]
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user