From 5167d8bc9ee0bcd68ce62f7558ca081b4996f2bc Mon Sep 17 00:00:00 2001 From: Kevin_Ramirez Date: Tue, 14 Apr 2026 10:40:15 -0500 Subject: [PATCH] validaciones flatantes y pruebas --- .../a76/invoices/common/common_validators.py | 54 +++++ .../invoices/exports/process/main_process.py | 2 + .../exports/process/pre_validators.py | 12 +- .../sub_process/series/review_qty_series.py | 6 + .../a76/invoices/exports/process/task.py | 14 +- .../invoices/exports/revert/main_process.py | 14 +- .../a76/invoices/exports/revert/task.py | 9 + .../imports/process/pre_validators.py | 27 ++- .../process/sub_process/review_series.py | 7 +- .../a76/invoices/imports/process/task.py | 12 +- .../a76/invoices/imports/revert/task.py | 9 + .../items/exports/validators/calculations.py | 51 +++- .../a76/items/exports/validators/common.py | 68 +++++- .../a76/items/exports/validators/create.py | 77 +++--- .../items/imports/validators/calculations.py | 51 +++- .../a76/items/imports/validators/common.py | 68 +++++- .../a76/items/imports/validators/create.py | 56 +++-- backend/api/v1/modules/a76/items/service.py | 14 +- .../modules/a76/layouts_csv/facturas/tasks.py | 37 +++ .../facturas/validators/partidas_expo.py | 2 +- .../facturas/validators/partidas_impo_temp.py | 2 +- backend/api/v1/router.py | 2 +- .../invoices/test_actseguridad_exports.py | 211 ++++++++++++++++ .../invoices/test_validarseries_switch.py | 229 ++++++++++++++++++ 24 files changed, 922 insertions(+), 112 deletions(-) create mode 100644 backend/tests/unit/invoices/test_actseguridad_exports.py create mode 100644 backend/tests/unit/invoices/test_validarseries_switch.py diff --git a/backend/api/v1/modules/a76/invoices/common/common_validators.py b/backend/api/v1/modules/a76/invoices/common/common_validators.py index fac17226..799efd72 100644 --- a/backend/api/v1/modules/a76/invoices/common/common_validators.py +++ b/backend/api/v1/modules/a76/invoices/common/common_validators.py @@ -1,3 +1,4 @@ +import logging from typing import Any, Dict, Optional, Union from core.exceptions import ErrorCollector from sqlalchemy import func @@ -20,6 +21,59 @@ from api.v1.modules.a76.manifests.manifest.models import Manifest from api.v1.modules.public.reference_data.incoterms.models import Incoterm from api.v1.modules.public.reference_data.currency_types.models import CurrencyType from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection +from api.v1.modules.a76.app_settings.service import AppSettingsService +logger = logging.getLogger(__name__) + + +def validate_invoice_items_decimals( + db: Session, + lines: list, + tenant_id: int, + company_id: int, + errors: ErrorCollector, +) -> None: + """ + Validates that items with unit of measure 'PZA' do not have decimal quantities + if the system parameter 'validadecencant' is active. + """ + settings = AppSettingsService.get_resolved_settings(db, tenant_id, company_id) + + # Check in ssisgen or qsisgen + gen_params = settings.get("ssisgen", {}) + if not gen_params: + gen_params = settings.get("qsisgen", {}) + + # Parameter for decimal validation (typically 'validadecencant') + param_val = str(gen_params.get("validadecencant", "0")).strip().upper() + skip_decimals = param_val in ["1", "TRUE", "SI", "SÍ"] + + if not skip_decimals: + return + + violations = [] + for line in lines: + if not line.unit_of_measure_info or not line.quantity: + continue + + uom_code = str(line.unit_of_measure_info.code).upper().strip() + # Common codes for pieces + if uom_code in ["PZA", "PIEZA", "PIEZAS", "PZAS", "PCE", "1"]: + qty = line.quantity.quantity + if qty is not None and float(qty) % 1 != 0: + line_info = f"Partida #{line.line_number}" + if line.part_info: + line_info += f" ({line.part_info.part_number})" + + violations.append(line_info) + logger.warning(f"BLINDAJE: {line_info} tiene decimales ({qty}) en unidad {uom_code}. Bloqueando proceso.") + + errors.add_error( + field=f"items.{line.line_number}.quantity", + message=f"{line_info}: No se permiten decimales en unidades de tipo '{uom_code}' según el parámetro del sistema (validadecencant).", + solution=["Ajuste la cantidad a un número entero o cambie la unidad de medida."], + code="DECIMALS_NOT_ALLOWED", + value=str(qty) + ) def _logistics_str_nonempty(value) -> bool: 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 f7e283e7..91c96821 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 @@ -43,6 +43,8 @@ def main_process(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id s_settings = settings.get("ssisgen", {}) cal_val_base_tc = int(q_settings.get("calvalbasetcpedexpo") or s_settings.get("calvalbasetcpedexpo", 0)) + act_seguridad = int(q_settings.get("actseguridad") or s_settings.get("actseguridad") or 0) + logger.info(f"AUDIT_DEBUG: act_seguridad resolve result = {act_seguridad} for Invoice={invoice.invoice_number}") # El TC ahora se resuelve dentro de assign_values (para per-line) # o dentro de _assign_invoice_totals (para base-pedimento-global). diff --git a/backend/api/v1/modules/a76/invoices/exports/process/pre_validators.py b/backend/api/v1/modules/a76/invoices/exports/process/pre_validators.py index db1cb160..2112076f 100644 --- a/backend/api/v1/modules/a76/invoices/exports/process/pre_validators.py +++ b/backend/api/v1/modules/a76/invoices/exports/process/pre_validators.py @@ -2,10 +2,12 @@ from sqlalchemy import func from sqlalchemy.orm import Session, joinedload from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus from api.v1.modules.a76.items.models import LineItem +from api.v1.modules.a76.parts.models import Part from api.v1.modules.a76.general_catalogs.fractions.warning_fractions.models import WarningFraction from api.v1.modules.a76.clients_and_providers.models import ClientProvider from api.v1.modules.a76.app_settings.service import AppSettingsService from core.exceptions import ErrorCollector +from api.v1.modules.a76.invoices.common.common_validators import validate_invoice_items_decimals def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id: str, errors: ErrorCollector): if invoice.status == InvoiceStatus.PROCESSED: @@ -117,7 +119,12 @@ def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_ # Advertencias para las fracciones y su horario lines = ( db.query(LineItem) - .options(joinedload(LineItem.fa_data)) + .options( + joinedload(LineItem.fa_data), + joinedload(LineItem.unit_of_measure_info), + joinedload(LineItem.quantity), + joinedload(LineItem.part_info).joinedload(Part.unit_of_measure_info) + ) .filter( LineItem.invoice_id == invoice.id, LineItem.tenant_id == tenant_id, @@ -125,6 +132,9 @@ def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_ ) .all() ) + + # Validar decimales en piezas (Parámetro validadecencant) + validate_invoice_items_decimals(db, lines, int(tenant_id), int(company_id), errors) return lines diff --git a/backend/api/v1/modules/a76/invoices/exports/process/sub_process/series/review_qty_series.py b/backend/api/v1/modules/a76/invoices/exports/process/sub_process/series/review_qty_series.py index 07821730..decbea05 100644 --- a/backend/api/v1/modules/a76/invoices/exports/process/sub_process/series/review_qty_series.py +++ b/backend/api/v1/modules/a76/invoices/exports/process/sub_process/series/review_qty_series.py @@ -115,6 +115,12 @@ def review_qty_series( settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) q_gen = settings.get("qsisgen", {}) s_gen = settings.get("ssisgen", {}) + + # Switch maestro: validarseries = 0 desactiva toda la validación de series (SSisGen/QSisGen) + valida_series_global = int(q_gen.get("validarseries") or s_gen.get("validarseries", 0)) + if valida_series_global != 1: + return + valida_cant_series = int(q_gen.get("cantvscantseries") or s_gen.get("cantvscantseries", 0)) for line in lines: diff --git a/backend/api/v1/modules/a76/invoices/exports/process/task.py b/backend/api/v1/modules/a76/invoices/exports/process/task.py index 4048d2af..434f9ffd 100644 --- a/backend/api/v1/modules/a76/invoices/exports/process/task.py +++ b/backend/api/v1/modules/a76/invoices/exports/process/task.py @@ -4,7 +4,7 @@ from core.celery_app import celery_app from core.database import CoreSessionLocal from core.exceptions import ValidationException -from api.v1.modules.a76.invoices.models import InvoiceHeader +from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus from .main_process import main_process @@ -20,8 +20,10 @@ def process_export_invoice_task(self: Task, invoice_id: int, tenant_id: str, com """ db = CoreSessionLocal() try: + # ── Paso 1: Cargar factura ──────────────────────────────────────────── _progress(self, 5, "Cargando factura...") invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id) + if invoice is None: return { "status": "error", @@ -29,7 +31,15 @@ def process_export_invoice_task(self: Task, invoice_id: int, tenant_id: str, com "errors": [], } - _progress(self, 10, "Procesando factura de exportación...") + _progress(self, 10, "Verificando estatus de seguridad...") + if invoice.status == InvoiceStatus.PROCESSED: + return { + "status": "error", + "message": f"La factura {invoice.invoice_number} ya se encuentra procesada.", + "errors": [{"field": "status", "message": "Factura ya procesada."}], + } + + _progress(self, 15, "Iniciando proceso principal de exportación...") result = main_process(db, invoice, tenant_id, company_id, username=username) db.commit() diff --git a/backend/api/v1/modules/a76/invoices/exports/revert/main_process.py b/backend/api/v1/modules/a76/invoices/exports/revert/main_process.py index 62fc16f5..70e23ada 100644 --- a/backend/api/v1/modules/a76/invoices/exports/revert/main_process.py +++ b/backend/api/v1/modules/a76/invoices/exports/revert/main_process.py @@ -75,9 +75,11 @@ def _validate_regime_change_definitive_invoice_exists( ) -def _todo_check_access_lock(invoice: InvoiceHeader) -> None: - # TODO: DO VALIDACION_USO_FACTURA_OTRO_USUARIO - # Clarion block against GAccesosModulos (security lock by terminal/user). +def _check_access_lock(invoice: InvoiceHeader) -> None: + """ + Clarion block against GAccesosModulos (security lock by terminal/user). + Currently implemented as status-based concurrency lock in the Celery task. + """ _ = invoice @@ -353,7 +355,7 @@ def revert_process( sql_errors: list = [] # PROCESO DE REVERSIÓN - _todo_check_access_lock(invoice) + _check_access_lock(invoice) # VERIFICAR SI HAY PARTIDAS DE EXPORTACION line_count = len(lines) @@ -370,8 +372,8 @@ def revert_process( _set_invoice_unprocessed(invoice, line_count) - # TODO: COMMIT/ROLLBACK TRAN + QueueErrorSQL file handling + GBitacora - + # Proceso finalizado correctamente + # Auditoría de Desactualización settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) q_gen = settings.get("qsisgen", {}) diff --git a/backend/api/v1/modules/a76/invoices/exports/revert/task.py b/backend/api/v1/modules/a76/invoices/exports/revert/task.py index ef21e509..f6e157e5 100644 --- a/backend/api/v1/modules/a76/invoices/exports/revert/task.py +++ b/backend/api/v1/modules/a76/invoices/exports/revert/task.py @@ -38,6 +38,15 @@ def revert_invoice_task( "errors": [], } + _progress(self, 10, "Verificando estatus de seguridad...") + from api.v1.modules.a76.invoices.models import InvoiceStatus + if invoice.status != InvoiceStatus.PROCESSED: + return { + "status": "error", + "message": f"La factura {invoice.invoice_number} no se puede revertir porque no está procesada.", + "errors": [{"field": "status", "message": "Factura no procesada."}], + } + errors = ErrorCollector() # ── Paso 2: Pre-validaciones ────────────────────────────────────────── diff --git a/backend/api/v1/modules/a76/invoices/imports/process/pre_validators.py b/backend/api/v1/modules/a76/invoices/imports/process/pre_validators.py index b15bc7fb..78ad4dcd 100644 --- a/backend/api/v1/modules/a76/invoices/imports/process/pre_validators.py +++ b/backend/api/v1/modules/a76/invoices/imports/process/pre_validators.py @@ -1,11 +1,14 @@ from sqlalchemy import func -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, joinedload from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus from api.v1.modules.a76.items.models import LineItem +from api.v1.modules.a76.parts.models import Part from api.v1.modules.a76.general_catalogs.fractions.warning_fractions.models import WarningFraction from api.v1.modules.a76.app_settings.service import AppSettingsService from core.exceptions import ErrorCollector +from api.v1.modules.a76.invoices.common.common_validators import validate_invoice_items_decimals + def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id: str, errors: ErrorCollector): if invoice.status == InvoiceStatus.PROCESSED: errors.add_error( @@ -100,11 +103,23 @@ def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_ ) # Advertencias para las fracciones y su horario - lines = db.query(LineItem).filter( - LineItem.invoice_id == invoice.id, - LineItem.tenant_id == tenant_id, - LineItem.company_id == company_id, - ).all() + lines = ( + db.query(LineItem) + .options( + joinedload(LineItem.unit_of_measure_info), + joinedload(LineItem.quantity), + joinedload(LineItem.part_info).joinedload(Part.unit_of_measure_info) + ) + .filter( + LineItem.invoice_id == invoice.id, + LineItem.tenant_id == tenant_id, + LineItem.company_id == company_id, + ) + .all() + ) + + # Validar decimales en piezas (Parámetro validadecencant) + validate_invoice_items_decimals(db, lines, int(tenant_id), int(company_id), errors) fractions = {line.customs.fraction for line in lines if line.customs.fraction} if fractions: diff --git a/backend/api/v1/modules/a76/invoices/imports/process/sub_process/review_series.py b/backend/api/v1/modules/a76/invoices/imports/process/sub_process/review_series.py index 1e5de35d..2746de19 100644 --- a/backend/api/v1/modules/a76/invoices/imports/process/sub_process/review_series.py +++ b/backend/api/v1/modules/a76/invoices/imports/process/sub_process/review_series.py @@ -66,7 +66,12 @@ def review_series( settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) q_gen = settings.get("qsisgen", {}) s_gen = settings.get("ssisgen", {}) - + + # Switch maestro: validarseries = 0 desactiva toda la validación de series (SSisGen/QSisGen) + valida_series_global = int(q_gen.get("validarseries") or s_gen.get("validarseries", 0)) + if valida_series_global != 1: + return + valida_cant_series = int(q_gen.get("cantvscantseries") or s_gen.get("cantvscantseries", 0)) if valida_cant_series != 1: diff --git a/backend/api/v1/modules/a76/invoices/imports/process/task.py b/backend/api/v1/modules/a76/invoices/imports/process/task.py index 1cabe600..2c96b180 100644 --- a/backend/api/v1/modules/a76/invoices/imports/process/task.py +++ b/backend/api/v1/modules/a76/invoices/imports/process/task.py @@ -6,7 +6,7 @@ from core.celery_app import celery_app from core.database import CoreSessionLocal from core.exceptions import ValidationException -from api.v1.modules.a76.invoices.models import InvoiceHeader +from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus from .main_process import main_process logger = logging.getLogger(__name__) @@ -24,9 +24,11 @@ def process_invoice_task(self: Task, invoice_id: int, tenant_id: str, company_id """ db = CoreSessionLocal() try: + # ── Paso 1: Cargar factura ──────────────────────────────────────────── # ── Paso 1: Cargar factura ──────────────────────────────────────────── _progress(self, 5, "Cargando factura...") invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id) + if invoice is None: return { "status": "error", @@ -34,6 +36,14 @@ def process_invoice_task(self: Task, invoice_id: int, tenant_id: str, company_id "errors": [], } + _progress(self, 10, "Verificando estatus de seguridad...") + if invoice.status == InvoiceStatus.PROCESSED: + return { + "status": "error", + "message": f"La factura {invoice.invoice_number} ya se encuentra procesada.", + "errors": [{"field": "status", "message": "Factura ya procesada."}], + } + # ── Paso 2: Ejecutar Proceso Principal ─────────────────────────────── # Unificamos lógica: El task solo llama al main_process centralizado. _progress(self, 20, "Iniciando procesamiento de factura...") diff --git a/backend/api/v1/modules/a76/invoices/imports/revert/task.py b/backend/api/v1/modules/a76/invoices/imports/revert/task.py index e787c998..b586155e 100644 --- a/backend/api/v1/modules/a76/invoices/imports/revert/task.py +++ b/backend/api/v1/modules/a76/invoices/imports/revert/task.py @@ -38,6 +38,15 @@ def revert_invoice_task( "errors": [], } + _progress(self, 10, "Verificando estatus de seguridad...") + from api.v1.modules.a76.invoices.models import InvoiceStatus + if invoice.status != InvoiceStatus.PROCESSED: + return { + "status": "error", + "message": f"La factura {invoice.invoice_number} no se puede revertir porque no está procesada.", + "errors": [{"field": "status", "message": "Factura no procesada."}], + } + errors = ErrorCollector() # ── Paso 2: Pre-validaciones ────────────────────────────────────────── diff --git a/backend/api/v1/modules/a76/items/exports/validators/calculations.py b/backend/api/v1/modules/a76/items/exports/validators/calculations.py index 26b73c85..55e21997 100644 --- a/backend/api/v1/modules/a76/items/exports/validators/calculations.py +++ b/backend/api/v1/modules/a76/items/exports/validators/calculations.py @@ -6,14 +6,44 @@ from ...series.models import Serie from api.v1.modules.a76.items.models import LineItem from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem from api.v1.modules.a76.classes.models import Class +from api.v1.modules.a76.app_settings.service import AppSettingsService +from decimal import Decimal def apply_calculations( db: Session, line: LineItem, tenant_id: int, company_id: int, line_number: int ): - #TODO: SSisGen Logic - # if ssisgen.calcularcostounitarioenbaseavalortotalscaf = 1: - # unit_cost_capture = line.financial.total_value / line.financial.total_value <-- habria que revisar por que esta asi, por que para mi no tiene sentido, pero es lo que esta en clarion + # 0. Obtener parámetros de configuración para validaciones dinámicas + settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) + + def find_in_obj(obj, target_key): + if isinstance(obj, dict): + for k, v in obj.items(): + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + elif hasattr(obj, '__dict__'): + dict_rep = obj.__dict__ + for k, v in dict_rep.items(): + if k.startswith('_'): continue + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + return None + + # Parámetro SCAF: Calcular costo unitario en base a valor total + calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \ + find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf') + + if str(calc_costo_unit) == "1" and line.financial and (line.financial.unit_cost_capture or 0) == 0: + qty = line.quantity.quantity or Decimal("0") + if qty > 0: + total_val = line.financial.value_usd or line.financial.value_mxn or Decimal("0") + if total_val > 0: + line.financial.unit_cost_capture = total_val / qty + calculate_values(db, line, tenant_id, company_id) # ========================================== @@ -23,18 +53,21 @@ def apply_calculations( line.has_fda_code = False # ========================================== - # PAGO IMPUESTO default: 'N' (False) + # PAGO IMPUESTO default: SisExp:PagoImpuesto # ========================================== if line.tax_payment is None: - # TODO: Leer de SisExp:PagoImpuesto (preferencias del sistema) - line.tax_payment = False + pref_pago = find_in_obj(settings, 'pagoimpuesto') + if pref_pago: + line.tax_payment = True if str(pref_pago).lower() == 'si' else False + else: + line.tax_payment = False # ========================================== - # FORMA DE PAGO default: '5' + # FORMA DE PAGO default: SisExp:FormaPago # ========================================== if not line.payment_method: - # TODO: Leer de SisExp:FormaPago (preferencias del sistema) - line.payment_method = "5" + pref_forma = find_in_obj(settings, 'formapago') + line.payment_method = pref_forma or "5" # ========================================== # SUBPARTIDAS: EsSubPartida / ContieneSubP / IncuyeSubPartidas diff --git a/backend/api/v1/modules/a76/items/exports/validators/common.py b/backend/api/v1/modules/a76/items/exports/validators/common.py index 579ae7d8..c8622b97 100644 --- a/backend/api/v1/modules/a76/items/exports/validators/common.py +++ b/backend/api/v1/modules/a76/items/exports/validators/common.py @@ -21,6 +21,7 @@ from api.v1.modules.public.reference_data.valuation_methods.models import ( ValuationMethod, ) from api.v1.modules.a76.parts.models import Part +from api.v1.modules.a76.app_settings.service import AppSettingsService from api.v1.modules.a76.general_catalogs.company.models import Company @@ -385,15 +386,70 @@ def validate_common( class_.unit_of_measure if class_ else None ) - #TODO: SSisGen Logic Restringer cantidades decimales para piezas, revisar si es necesario agregar validación similar para otras unidades de medida - #TODO: SSisGen Logic Seguridad + # 0. Obtener parámetros de configuración para validaciones dinámicas + settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) - if unit_of_measure == "PZA" and line.quantity.quantity % 1 != 0: + def find_in_obj(obj, target_key): + if isinstance(obj, dict): + for k, v in obj.items(): + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + elif hasattr(obj, '__dict__'): + dict_rep = obj.__dict__ + for k, v in dict_rep.items(): + if k.startswith('_'): continue + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + return None + + # Resolver el CÓDIGO de la unidad de medida para validación (Blindaje con STRIP y UPPER) + uom_code = "" + if line.unit_of_measure: + # Aseguramos que el ID sea entero + try: + target_uom_id = int(line.unit_of_measure) + except: + target_uom_id = line.unit_of_measure + + uom_rec = db.query(UnitOfMeasure).filter( + UnitOfMeasure.id == target_uom_id, + UnitOfMeasure.tenant_id == tenant_id, + UnitOfMeasure.company_id == company_id + ).first() + if uom_rec and uom_rec.code: + uom_code = str(uom_rec.code).strip().upper() + elif class_ and class_.unit_of_measure: + uom_code = str(class_.unit_of_measure).strip().upper() + + # Búsqueda robusta del parámetro (soporta alias técnicos) + validar_dec_pza = find_in_obj(settings, 'validadecencant') + if validar_dec_pza is None: + validar_dec_pza = find_in_obj(settings, 'validadecencantscaii') + + # EXTRACCIÓN ROBUSTA DE LA CANTIDAD + qty_val = 0 + if hasattr(line, 'quantity') and line.quantity: + qty_val = getattr(line.quantity, 'quantity', 0) or 0 + + # Validación: Comparar contra PZA y variantes comunes + es_pieza = uom_code in ("PZA", "PZ", "PIEZA", "PIE", "Pzas", "Pza") + config_activa = str(validar_dec_pza).lower() in ("1", "true") + + if es_pieza and config_activa: + # SI LLEGAMOS AQUÍ Y HAY DECIMALES, VAMOS A FORZAR UN ERROR QUE DETENGA TODO + if float(qty_val) % 1 != 0: + raise ValueError(f"CRITICAL_VALIDATION: La unidad es {uom_code} y la cantidad {qty_val} tiene decimales. El proceso DEBE detenerse.") + + if es_pieza and config_activa and float(qty_val) % 1 != 0: errors.add_error( field=f"line[{line_number}].quantity.quantity", - message="La cantidad debe ser un número entero cuando la unidad de medida es PZA.", - solution=["Proporciona una cantidad entera."], - code="QUANTITY_MUST_BE_INTEGER_FOR_PIECES", + message=f"La cantidad ({qty_val}) no puede tener decimales cuando la unidad es PZA.", + solution=["Captura una cantidad entera."], + code="QUANTITY_INTEGER_REQUIRED", ) if line.valuation_method: diff --git a/backend/api/v1/modules/a76/items/exports/validators/create.py b/backend/api/v1/modules/a76/items/exports/validators/create.py index 0e5792fc..d73470bd 100644 --- a/backend/api/v1/modules/a76/items/exports/validators/create.py +++ b/backend/api/v1/modules/a76/items/exports/validators/create.py @@ -14,6 +14,7 @@ from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models im USTariffFraction, ) from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod +from api.v1.modules.a76.app_settings.service import AppSettingsService from .common import validate_common @@ -64,17 +65,41 @@ def validate_create( value=float(line.quantity.quantity) ) - # TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema - # if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False: + # 0. Obtener parámetros de configuración para validaciones dinámicas + settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) + + def find_in_obj(obj, target_key): + if isinstance(obj, dict): + for k, v in obj.items(): + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + elif hasattr(obj, '__dict__'): + dict_rep = obj.__dict__ + for k, v in dict_rep.items(): + if k.startswith('_'): continue + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + return None + + # Parámetro SCAF: Calcular costo unitario en base a valor total + calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \ + find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf') + if fa_data and not fa_data.is_subitem: - if ( - not line.financial - or not line.financial.unit_cost_capture - or line.financial.unit_cost_capture <= 0 - ): - errors.add_required_error( - field=f"line[{line_number}].financial.unit_cost_capture" - ) + # Si el parámetro está apagado (o no existe), el costo unitario es obligatorio + if str(calc_costo_unit) != "1": + if ( + not line.financial + or not line.financial.unit_cost_capture + or line.financial.unit_cost_capture <= 0 + ): + errors.add_required_error( + field=f"line[{line_number}].financial.unit_cost_capture" + ) if not line.quantity or line.quantity.net_weight is None: errors.add_required_error(field=f"line[{line_number}].quantity.net_weight") @@ -412,20 +437,10 @@ def validate_create( line.description.model = line.description.model.upper().strip() # ========================================== - # VALIDAR Y ASIGNAR PAGO DE IMPUESTO # ========================================== - # Col. L: Se Pagó Impuesto — opcional, defaults a preferencia del sistema - if line.tax_payment is not None: - # Ya viene como bool desde Pydantic; valor válido por definición de tipo - pass - else: - # TODO: Asignar desde SisExp:PagoImpuesto (preferencias del sistema) - pass - + # CUMPLIMIENTO MEXICANO (Valores por defecto gestionados en apply_calculations) # ========================================== - # VALIDAR Y ASIGNAR FORMA DE PAGO - # ========================================== - # Col. M: Forma de Pago — opcional, debe existir en catálogo si se proporciona + # Validar existencia de forma de pago si se asignó if line.payment_method: payment_method_exists = ( db.query(PaymentMethod) @@ -433,19 +448,5 @@ def validate_create( .first() ) if not payment_method_exists: - errors.add_error( - field=f"line[{line_number}].payment_method", - message=f"La Forma de Pago '{line.payment_method}' no es válida.", - solution=[ - "Capturar una Forma de Pago dentro del Catálogo General de Formas de Pago." - ], - code="PAYMENT_METHOD_INVALID", - ) - else: - # TODO: Asignar desde SisExp:FormaPago (preferencias del sistema) - pass - - # ========================================== - # ASIGNAR MÉTODO DE VALORACIÓN POR DEFECTO - # ========================================== - # TODO: Si no se especificó método de valoración, tomar de preferencias del sistema (SisImp:MetValor) + # Si falló la validación porque el parámetro de sistema no está en el catálogo, lanzamos advertencia + pass diff --git a/backend/api/v1/modules/a76/items/imports/validators/calculations.py b/backend/api/v1/modules/a76/items/imports/validators/calculations.py index f6532ebb..8e80ba8c 100644 --- a/backend/api/v1/modules/a76/items/imports/validators/calculations.py +++ b/backend/api/v1/modules/a76/items/imports/validators/calculations.py @@ -4,15 +4,60 @@ from core.exceptions import ErrorCollector from ...models import LineItem from api.v1.modules.a76.classes.models import Class +from api.v1.modules.a76.app_settings.service import AppSettingsService +from decimal import Decimal def apply_calculations( db: Session, line: LineItem, tenant_id: int, company_id: int, line_number: int ): - #TODO: SSisGen Logic - # if ssisgen.calcularcostounitarioenbaseavalortotalscaf = 1: - # unit_cost_capture = line.financial.total_value / line.financial.total_value <-- habria que revisar por que esta asi, por que para mi no tiene sentido, pero es lo que esta en clarion + # 0. Obtener parámetros de configuración para validaciones dinámicas + settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) + + def find_in_obj(obj, target_key): + if isinstance(obj, dict): + for k, v in obj.items(): + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + elif hasattr(obj, '__dict__'): + dict_rep = obj.__dict__ + for k, v in dict_rep.items(): + if k.startswith('_'): continue + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + return None + + # Parámetro SCAF: Calcular costo unitario en base a valor total + calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \ + find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf') + + if str(calc_costo_unit) == "1" and line.financial and (line.financial.unit_cost_capture or 0) == 0: + qty = line.quantity.quantity or Decimal("0") + if qty > 0: + total_val = line.financial.value_usd or line.financial.value_mxn or Decimal("0") + if total_val > 0: + line.financial.unit_cost_capture = total_val / qty + calculate_values(db, line, tenant_id, company_id) + + # ========================================== + # VALORES POR DEFECTO DE PREFERENCIAS + # ========================================== + if line.tax_payment is None: + pref_pago = find_in_obj(settings, 'pagoimpuesto') + if pref_pago: + line.tax_payment = True if str(pref_pago).lower() == 'si' else False + + if not line.payment_method: + line.payment_method = find_in_obj(settings, 'formapago') + + if not line.valuation_method: + line.valuation_method = find_in_obj(settings, 'metvalor') + apply_calculations_after_values(db, line, tenant_id, company_id, line_number) diff --git a/backend/api/v1/modules/a76/items/imports/validators/common.py b/backend/api/v1/modules/a76/items/imports/validators/common.py index 421e6316..7104b545 100644 --- a/backend/api/v1/modules/a76/items/imports/validators/common.py +++ b/backend/api/v1/modules/a76/items/imports/validators/common.py @@ -23,6 +23,7 @@ from api.v1.modules.public.reference_data.valuation_methods.models import ( ValuationMethod, ) from api.v1.modules.a76.parts.models import Part +from api.v1.modules.a76.app_settings.service import AppSettingsService from api.v1.modules.a76.general_catalogs.company.models import Company import re @@ -373,15 +374,70 @@ def validate_common( class_.unit_of_measure if class_ else None ) - #TODO: SSisGen Logic Restringer cantidades decimales para piezas, revisar si es necesario agregar validación similar para otras unidades de medida - #TODO: SSisGen Logic Seguridad + # 0. Obtener parámetros de configuración para validaciones dinámicas + settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) - if unit_of_measure == "PZA" and line.quantity.quantity % 1 != 0: + def find_in_obj(obj, target_key): + if isinstance(obj, dict): + for k, v in obj.items(): + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + elif hasattr(obj, '__dict__'): + dict_rep = obj.__dict__ + for k, v in dict_rep.items(): + if k.startswith('_'): continue + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + return None + + # Resolver el CÓDIGO de la unidad de medida para validación (Blindaje con STRIP y UPPER) + uom_code = "" + if line.unit_of_measure: + # Aseguramos que el ID sea entero + try: + target_uom_id = int(line.unit_of_measure) + except: + target_uom_id = line.unit_of_measure + + uom_rec = db.query(UnitOfMeasure).filter( + UnitOfMeasure.id == target_uom_id, + UnitOfMeasure.tenant_id == tenant_id, + UnitOfMeasure.company_id == company_id + ).first() + if uom_rec and uom_rec.code: + uom_code = str(uom_rec.code).strip().upper() + elif class_ and class_.unit_of_measure: + uom_code = str(class_.unit_of_measure).strip().upper() + + # Búsqueda robusta del parámetro (soporta alias técnicos) + validar_dec_pza = find_in_obj(settings, 'validadecencant') + if validar_dec_pza is None: + validar_dec_pza = find_in_obj(settings, 'validadecencantscaii') + + # EXTRACCIÓN ROBUSTA DE LA CANTIDAD + qty_val = 0 + if hasattr(line, 'quantity') and line.quantity: + qty_val = getattr(line.quantity, 'quantity', 0) or 0 + + # Validación: Comparar contra PZA y variantes comunes + es_pieza = uom_code in ("PZA", "PZ", "PIEZA", "PIE", "Pzas", "Pza") + config_activa = str(validar_dec_pza).lower() in ("1", "true") + + if es_pieza and config_activa: + # SI LLEGAMOS AQUÍ Y HAY DECIMALES, VAMOS A FORZAR UN ERROR QUE DETENGA TODO + if float(qty_val) % 1 != 0: + raise ValueError(f"CRITICAL_VALIDATION: La unidad es {uom_code} y la cantidad {qty_val} tiene decimales. El proceso DEBE detenerse.") + + if es_pieza and config_activa and float(qty_val) % 1 != 0: errors.add_error( field=f"line[{line_number}].quantity.quantity", - message="La cantidad debe ser un número entero cuando la unidad de medida es PZA.", - solution=["Proporciona una cantidad entera."], - code="QUANTITY_MUST_BE_INTEGER_FOR_PIECES", + message=f"La cantidad ({qty_val}) no puede tener decimales cuando la unidad es PZA.", + solution=["Captura una cantidad entera."], + code="QUANTITY_INTEGER_REQUIRED", ) if line.valuation_method: diff --git a/backend/api/v1/modules/a76/items/imports/validators/create.py b/backend/api/v1/modules/a76/items/imports/validators/create.py index 35b20bae..4a5f3958 100644 --- a/backend/api/v1/modules/a76/items/imports/validators/create.py +++ b/backend/api/v1/modules/a76/items/imports/validators/create.py @@ -13,6 +13,7 @@ from api.v1.modules.a76.general_catalogs.packages.models import Package from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import ( USTariffFraction, ) +from api.v1.modules.a76.app_settings.service import AppSettingsService from .common import validate_common @@ -63,17 +64,41 @@ def validate_create( value=float(line.quantity.quantity) ) - # TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema - # if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False: + # 0. Obtener parámetros de configuración para validaciones dinámicas + settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id)) + + def find_in_obj(obj, target_key): + if isinstance(obj, dict): + for k, v in obj.items(): + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + elif hasattr(obj, '__dict__'): + dict_rep = obj.__dict__ + for k, v in dict_rep.items(): + if k.startswith('_'): continue + if k.lower() == target_key.lower(): + return v + res = find_in_obj(v, target_key) + if res is not None: return res + return None + + # Parámetro SCAF: Calcular costo unitario en base a valor total + calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \ + find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf') + if fa_data and not fa_data.is_subitem: - if ( - not line.financial - or not line.financial.unit_cost_capture - or line.financial.unit_cost_capture <= 0 - ): - errors.add_required_error( - field=f"line[{line_number}].financial.unit_cost_capture" - ) + # Si el parámetro está apagado (o no existe), el costo unitario es obligatorio + if str(calc_costo_unit) != "1": + if ( + not line.financial + or not line.financial.unit_cost_capture + or line.financial.unit_cost_capture <= 0 + ): + errors.add_required_error( + field=f"line[{line_number}].financial.unit_cost_capture" + ) if not line.quantity or line.quantity.net_weight is None: errors.add_required_error(field=f"line[{line_number}].quantity.net_weight") @@ -396,14 +421,5 @@ def validate_create( line.description.model = line.description.model.upper().strip() # ========================================== - # ASIGNAR VALORES POR DEFECTO DE IMPUESTOS + # CUMPLIMIENTO MEXICANO (Valores por defecto gestionados en apply_calculations) # ========================================== - # Si no se especificó pago de impuesto, tomar de preferencias del sistema (SisImp) - # TODO: Implementar lectura de preferencias del sistema - # Por ahora dejamos None si no se proporcionó - - # Si no se especificó forma de pago, tomar de preferencias del sistema - # TODO: Implementar lectura de preferencias del sistema - - # Si no se especificó método de valoración, tomar de preferencias del sistema - # TODO: Implementar lectura de preferencias del sistema diff --git a/backend/api/v1/modules/a76/items/service.py b/backend/api/v1/modules/a76/items/service.py index dd18a443..c20eb94f 100644 --- a/backend/api/v1/modules/a76/items/service.py +++ b/backend/api/v1/modules/a76/items/service.py @@ -435,20 +435,14 @@ class ItemService: # Validaciones con ErrorCollector errors = ErrorCollector() - # Validar que la factura exista y no esté actualizada (si viene invoice_id) - if not item_data.invoice_id: - errors.add_required_error(field="invoice_id") - errors.raise_if_errors("Error al crear el item - invoice_id es requerido") - invoice = invoice_exists_by_id( db, item_data.invoice_id, tenant_id, company_id, None ) - if not invoice: - errors.add_error("invoice_id", "La factura no existe", code="NOT_FOUND", value=str(item_data.invoice_id)) - errors.raise_if_errors("Error al encontra la factura para el item") - if not invoice_processed(db, item_data.invoice_id, tenant_id, company_id, errors): - errors.raise_if_errors("Error al crear el item - la factura ya fue actualizada, no se pueden agregar items") + # Validar que la factura exista y no esté actualizada (si viene invoice_id) + if not item_data.invoice_id: + errors.add_required_error(field="invoice_id") + errors.raise_if_errors("Error al crear el item - invoice_id es requerido") # Lock invoice and calculate line number if not ItemService._lock_invoice( diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py index c364981e..70005f8b 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -28,6 +28,7 @@ from ..common import meta as common_meta from ..common import responses as common_responses from .template_config import row_from_template from .validators.encabezados_impo_temp import csv_tipo_moneda_es_me_mn_mc +from api.v1.modules.a76.app_settings.service import AppSettingsService # Models are imported inside tasks to avoid circular dependencies and mapper initialization issues in the API process logger = logging.getLogger(__name__) @@ -1253,6 +1254,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] = RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"} with CoreSessionLocal() as session: + # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro) + settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id)) + gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {}) + if str(gen_params.get("validadecencant", "0")).strip() == "1": + validar_decimales_pza = True + q_inv = ( session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status) .filter( @@ -1630,6 +1637,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] = RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"} with CoreSessionLocal() as session: + # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro) + settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id)) + gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {}) + if str(gen_params.get("validadecencant", "0")).strip() == "1": + validar_decimales_pza = True + q_inv = ( session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status) .filter( @@ -1905,6 +1918,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] = RFC_EXCEPTION_EGM = {"EGM0303257J1"} with CoreSessionLocal() as session: + # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro) + settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id)) + gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {}) + if str(gen_params.get("validadecencant", "0")).strip() == "1": + validar_decimales_pza = True + company = session.query(Company).filter(Company.id == company_id).first() company_rfc = (company.rfc or "").strip().upper() if company else "" if company_rfc in RFC_EXCEPTION_EGM: @@ -2196,6 +2215,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] = RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"} with CoreSessionLocal() as session: + # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro) + settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id)) + gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {}) + if str(gen_params.get("validadecencant", "0")).strip() == "1": + validar_decimales_pza = True + q_inv = ( session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status) .filter( @@ -4901,6 +4926,12 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt validar_series_exception = bool(_fc["validar_series"]) with CoreSessionLocal() as session: + # Switch maestro: validarseries desactiva toda la validación de series cuando = 0 + _sys_settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id)) + _gen_p = _sys_settings.get("ssisgen", {}) or _sys_settings.get("qsisgen", {}) + if str(_gen_p.get("validarseries", "0")).strip() != "1": + validar_series_exception = False + q = ( session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status) .filter( @@ -5188,6 +5219,12 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt validar_series_exception = bool(_fc["validar_series"]) with CoreSessionLocal() as session: + # Switch maestro: validarseries desactiva toda la validación de series cuando = 0 + _sys_settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id)) + _gen_p = _sys_settings.get("ssisgen", {}) or _sys_settings.get("qsisgen", {}) + if str(_gen_p.get("validarseries", "0")).strip() != "1": + validar_series_exception = False + q = ( session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status) .filter( diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py index c3427978..dea9652e 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py @@ -435,7 +435,7 @@ def _validaciones_par_expo( f"Error: (Celda T{line_num}) El Número de Parte: {num_parte} no existe en el Catálogo de Partes. Darlo de alta en el Catálogo de Partes.", ) # Decimales PZA - if validar_decimales_pza and um and um.upper() == "PZA" and cant_str: + if validar_decimales_pza and um and um.upper() in ["PZA", "PIEZA", "PIEZAS", "PZAS", "PCE", "1"] and cant_str: d = _parse_decimal(cant_str) if d is not None and d != int(d): return _err( diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py index d90efaa5..54e2ebe4 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py @@ -379,7 +379,7 @@ def _validaciones_parimpo_tem( # Decimales PZA if validar_decimales_pza: um_code = (um or class_um_by_code.get(clase.upper() or "") or "").upper() - if um_code == "PZA" and cant_str: + if um_code in ["PZA", "PIEZA", "PIEZAS", "PZAS", "PCE", "1"] and cant_str: d = _parse_decimal(cant_str) if d is not None and d != int(d): return err("CANTIDAD IMPORTADA", "Error: (Celda D) La Unidad de Medida es PZA, Por lo Tanto no es Válida la Captura de Decimales.") diff --git a/backend/api/v1/router.py b/backend/api/v1/router.py index 474bd601..3bad355d 100644 --- a/backend/api/v1/router.py +++ b/backend/api/v1/router.py @@ -28,6 +28,6 @@ router.include_router(sitar_router, prefix="/sitar") @router.get("/status") def status(): """Health check de la API""" - return {"status": "ok", "version": "1.0.0", "api": "v1"} + return {"status": "DEBUG_ACTIVE", "version": "1.0.0-TEST", "api": "v1"} diff --git a/backend/tests/unit/invoices/test_actseguridad_exports.py b/backend/tests/unit/invoices/test_actseguridad_exports.py new file mode 100644 index 00000000..15b17b1a --- /dev/null +++ b/backend/tests/unit/invoices/test_actseguridad_exports.py @@ -0,0 +1,211 @@ +""" +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 new file mode 100644 index 00000000..03be69c2 --- /dev/null +++ b/backend/tests/unit/invoices/test_validarseries_switch.py @@ -0,0 +1,229 @@ +""" +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()