import shutil import tempfile from datetime import date from decimal import Decimal from io import BytesIO from unittest.mock import patch from django.contrib.auth.models import Group, Permission from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.test import override_settings from django.utils import timezone from rest_framework.test import APIClient, APITestCase from apps.boletas.models import Boleta from apps.boletas.utils import build_folio, build_qr_text from apps.catalogos.models import Bascula, Bodega, Campo, Castigo, Contrato, ContratoVariedad, Linea, Oficina, Parametro, Productor, Variedad, Zafra from apps.muestreo_laboratorio.models import ParametroResultado, ResultadoLaboratorio from apps.recepciones.models import Bonificacion, Deduccion, Flete, Recepcion, RecepcionLog User = get_user_model() TEMP_MEDIA_ROOT = tempfile.mkdtemp() @override_settings(MEDIA_ROOT=TEMP_MEDIA_ROOT) class RecepcionesFlowTests(APITestCase): @classmethod def tearDownClass(cls): super().tearDownClass() shutil.rmtree(TEMP_MEDIA_ROOT, ignore_errors=True) def setUp(self): self.admin = User.objects.create_superuser( username="admin", email="admin@example.com", password="admin12345", ) self.operador = User.objects.create_user( username="operador", email="operador@example.com", password="operador12345", ) self.zafra = Zafra.objects.create(nombre="Z 2026", fecha_inicio=date(2026, 1, 1), fecha_fin=date(2026, 12, 31)) self.oficina = Oficina.objects.create( codigo="OF-1", nombre="Oficina 1", activa=True, created_by=self.admin, updated_by=self.admin, ) self.bodega = Bodega.objects.create(codigo="B-1", nombre="Bodega 1", oficina=self.oficina, activa=True) self.campo = Campo.objects.create(nombre="Campo 1", oficina=self.oficina) self.operador.profile.oficinas.add(self.oficina) self.variedad = Variedad.objects.create(nombre="V1") self.productor = Productor.objects.create(nombre="Prod 1", rfc="ABC123456T12") self.contrato = Contrato.objects.create( numero_contrato="C-1", productor=self.productor, zafra=self.zafra, activo=True, ) ContratoVariedad.objects.create(contrato=self.contrato, variedad=self.variedad) self.boleta_folio = build_folio(self.oficina.codigo, self.productor.id, self.zafra.fecha_inicio.year, 1) self.boleta = Boleta.objects.create( zafra=self.zafra, oficina=self.oficina, productor=self.productor, contrato=self.contrato, folio=self.boleta_folio, qr_texto=build_qr_text(self.zafra.id, self.boleta_folio), peso_estimado=100, campo=self.campo, origen_trigo="Campo 1", placas_camion="ABC-123", marca_unidad="Kenworth", modelo_unidad="T680", chofer_nombre="Chofer 1", chofer_telefono="555-0001", linea_nombre="Línea 1", created_by=self.admin, updated_by=self.admin, ) resp = self.client.post( "/auth/token/", {"username": "admin", "password": "admin12345"}, format="json", ) self.assertEqual(resp.status_code, 200) self.token = resp.data["access"] self.client.credentials( HTTP_AUTHORIZATION=f"Bearer {self.token}", HTTP_X_OFFICE_ID=str(self.oficina.id), ) def _login_operador(self, with_process_perm: bool = False): if with_process_perm: perm = Permission.objects.get(codename="process", content_type__app_label="recepciones") self.operador.user_permissions.add(perm) resp = self.client.post( "/auth/token/", {"username": "operador", "password": "operador12345"}, format="json", ) self.assertEqual(resp.status_code, 200) token = resp.data["access"] self.client.credentials( HTTP_AUTHORIZATION=f"Bearer {token}", HTTP_X_OFFICE_ID=str(self.oficina.id), ) def _identificar(self): return self.client.post( "/recepciones/identificar/", {"qr": self.boleta.qr_texto, "bodega_destino": self.bodega.id}, format="json", ) def test_identificar_boleta_ok_crea_recepcion(self): resp = self._identificar() self.assertEqual(resp.status_code, 200) self.assertTrue(resp.data.get("created")) self.assertEqual(Recepcion.objects.count(), 1) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_ASIGNADA) resp2 = self._identificar() self.assertEqual(resp2.status_code, 200) self.assertFalse(resp2.data.get("created")) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_ASIGNADA) def test_identificar_boleta_not_found(self): resp = self.client.post( "/recepciones/identificar/", {"qr": "zafra:999|folio:999"}, format="json", ) self.assertEqual(resp.status_code, 404) def test_identificar_por_folio_y_zafra_ok(self): resp = self.client.post( "/recepciones/identificar/", {"folio": self.boleta_folio, "zafra_id": self.zafra.id}, format="json", ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["boleta"]["folio"], self.boleta_folio) def test_permissions_identificar_requires_process_perm(self): self._login_operador(with_process_perm=False) resp = self._identificar() self.assertEqual(resp.status_code, 403) self._login_operador(with_process_perm=True) resp2 = self._identificar() self.assertEqual(resp2.status_code, 200) def test_patch_camion_placas_validation(self): resp = self._identificar() recepcion_id = resp.data["id"] ok = self.client.patch( f"/recepciones/{recepcion_id}/", {"placas_camion": "ABC-123"}, format="json", ) self.assertEqual(ok.status_code, 200) self.assertEqual(ok.data["placas_camion"], "ABC-123") def test_patch_camion_acepta_linea(self): linea = Linea.objects.create(codigo="L-1", nombre="Línea 1", oficina=self.oficina) recepcion_id = self._identificar().data["id"] resp = self.client.patch( f"/recepciones/{recepcion_id}/", {"linea": linea.id}, format="json", ) self.assertEqual(resp.status_code, 200) self.boleta.refresh_from_db() self.assertEqual(self.boleta.linea_id, linea.id) self.assertEqual(self.boleta.linea_nombre, linea.nombre) def test_registrar_pesos_muestreo_y_cerrar_con_alerta(self): resp = self._identificar() recepcion_id = resp.data["id"] resp_lleno = self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_lleno.status_code, 201) resp_muestreo = self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) self.assertEqual(resp_muestreo.status_code, 201) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impureza", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="3.000", descontar_kg_por_tonelada="0.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="2.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() resp_vacio = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "800.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_vacio.status_code, 201) self.assertIn("peso_bruto", resp_vacio.data) resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 200) self.assertEqual(resp_calc.data.get("detail"), "Sin deducciones aplicables.") resp_cerrar = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {}, format="json", ) self.assertEqual(resp_cerrar.status_code, 409) self.assertIn("alerta", resp_cerrar.data) resp_cerrar2 = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {"decision": "aceptar_con_comentario", "comentario": "OK"}, format="json", ) self.assertEqual(resp_cerrar2.status_code, 200) self.assertEqual(resp_cerrar2.data["estado"], Recepcion.ESTADO_CERRADA) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_CERRADA) def test_calcular_deducciones_impurezas_3_00(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "11000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impurezas", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="3.100", descontar_kg_por_tonelada="1.16", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="3.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() resp_vacio = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_vacio.status_code, 201) resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 200) self.assertFalse(resp_calc.data.get("rechazo")) recepcion.refresh_from_db() self.assertFalse(recepcion.rechazo_calidad) self.assertEqual(recepcion.rechazos_calidad, []) ded = Deduccion.objects.filter(recepcion_id=recepcion_id, parametro=p_imp).first() self.assertIsNotNone(ded) self.assertEqual(str(ded.kg_por_tonelada), "1.16") self.assertEqual(str(ded.valor_deducido), "11.60") self.assertEqual(Bonificacion.objects.filter(recepcion_id=recepcion_id).count(), 0) def test_calcular_deducciones_impurezas_4_20_rechazo_calidad(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "11000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impurezas", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="4.100", descontar_kg_por_tonelada="10.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="4.200", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() resp_vacio = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_vacio.status_code, 201) resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 200) self.assertTrue(resp_calc.data.get("rechazo")) recepcion.refresh_from_db() self.assertTrue(recepcion.rechazo_calidad) self.assertTrue(bool(recepcion.rechazos_calidad)) self.assertEqual(Deduccion.objects.filter(recepcion_id=recepcion_id).count(), 0) self.assertEqual(Bonificacion.objects.filter(recepcion_id=recepcion_id).count(), 0) def test_calcular_deducciones_y_bonificaciones_bonificacion_sobre_neto(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "11000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) # Impurezas: deducción 100 kg/t p_imp = Parametro.objects.create( nombre="Impurezas", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="10.000", descontar_kg_por_tonelada="100.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="1.000", created_by=self.admin, updated_by=self.admin, ) # Proteína: bonificación 50 $/t p_prot = Parametro.objects.create( nombre="Proteína", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_prot, limite_superior="20.000", descontar_kg_por_tonelada="0.00", bonificar_por_tonelada="50.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_prot, valor="12.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() resp_vacio = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_vacio.status_code, 201) resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 200) self.assertFalse(resp_calc.data.get("rechazo")) # bruto = 10000kg -> 10.0000 t # deducción: 100 kg/t * 10 t = 1000 kg => neto 9000kg # bonificación: 50 $/t * 9 t = 450 self.assertEqual(str(resp_calc.data.get("peso_bruto")), "10000.00") self.assertEqual(str(resp_calc.data.get("peso_neto")), "9000.00") self.assertEqual(str(resp_calc.data.get("total_bonificado")), "450.00") recepcion.refresh_from_db() self.assertEqual(Deduccion.objects.filter(recepcion_id=recepcion_id).count(), 1) self.assertEqual(Bonificacion.objects.filter(recepcion_id=recepcion_id).count(), 1) bon = Bonificacion.objects.filter(recepcion_id=recepcion_id, parametro=p_prot).first() self.assertIsNotNone(bon) self.assertEqual(str(bon.monto_por_tonelada), "50.00") self.assertEqual(str(bon.monto_total), "450.00") def test_calcular_deducciones_sin_parametros_retornar_422(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "11000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) # Hay resultado, pero no hay tabla (castigos) configurada para el parámetro capturado p_imp = Parametro.objects.create( nombre="Impurezas", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="3.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() resp_vacio = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_vacio.status_code, 201) resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 422) self.assertIn("resultado_laboratorio", resp_calc.data.get("errors", {})) def test_calcular_deducciones_sin_deducciones_aplicables_retornar_estado(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "11000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impureza", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="2.000", descontar_kg_por_tonelada="0.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="1.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() resp_vacio = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_vacio.status_code, 201) resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 200) self.assertEqual(resp_calc.data.get("detail"), "Sin deducciones aplicables.") self.assertIn("calculo_calidad", resp_calc.data) self.assertEqual(resp_calc.data["calculo_calidad"].get("estado"), "sin_deducciones") self.assertFalse(resp_calc.data.get("rechazo")) self.assertEqual(resp_calc.data.get("rechazos"), []) self.assertEqual(resp_calc.data.get("deducciones"), []) self.assertEqual(resp_calc.data.get("bonificaciones"), []) self.assertEqual(str(resp_calc.data.get("total_bonificado")), "0") recepcion.refresh_from_db() self.assertTrue(recepcion.calidad_calculada) resp_cerrar = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {"decision": "aceptar_con_comentario", "comentario": "OK"}, format="json", ) self.assertIn(resp_cerrar.status_code, (200, 409)) def test_peso_manual_requires_motivo(self): resp = self._identificar() recepcion_id = resp.data["id"] resp_lleno = self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": False}, format="json", ) self.assertEqual(resp_lleno.status_code, 422) resp_lleno2 = self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": False, "motivo_manual": "Báscula sin señal"}, format="json", ) self.assertEqual(resp_lleno2.status_code, 201) def test_no_duplicate_peso_lleno(self): resp = self._identificar() recepcion_id = resp.data["id"] r1 = self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(r1.status_code, 201) r2 = self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1001.00", "es_automatico": True}, format="json", ) self.assertEqual(r2.status_code, 422) def test_peso_vacio_requires_peso_lleno(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) recepcion.muestreo.estado = "completado" recepcion.muestreo.save() r = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "800.00", "es_automatico": True}, format="json", ) self.assertEqual(r.status_code, 422) def test_peso_vacio_no_puede_ser_igual_al_peso_lleno(self): recepcion_id = self._identificar().data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) recepcion.muestreo.estado = "completado" recepcion.muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save(update_fields=["estado", "updated_at"]) resp = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp.status_code, 422) self.assertEqual(resp.data["detail"], "El peso vacío no puede ser igual al peso lleno.") self.assertFalse(recepcion.pesos.filter(tipo="vacio").exists()) def test_peso_vacio_requires_muestreo_completado(self): resp = self._identificar() recepcion_id = resp.data["id"] resp_lleno = self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_lleno.status_code, 201) resp_muestreo = self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) self.assertEqual(resp_muestreo.status_code, 201) r = self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "800.00", "es_automatico": True}, format="json", ) self.assertEqual(r.status_code, 422) def test_cerrar_requires_pesos_and_muestreo(self): resp = self._identificar() recepcion_id = resp.data["id"] r1 = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {}, format="json", ) self.assertEqual(r1.status_code, 422) self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) r2 = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {}, format="json", ) self.assertEqual(r2.status_code, 422) from apps.recepciones.models import PesoRecepcion PesoRecepcion.objects.create( recepcion_id=recepcion_id, tipo=PesoRecepcion.TIPO_VACIO, valor="900.00", es_automatico=True, motivo_manual="", ) r3 = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {}, format="json", ) self.assertEqual(r3.status_code, 422) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo muestreo.estado = "completado" muestreo.save() resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impureza", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="3.000", descontar_kg_por_tonelada="0.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="2.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 200) self.assertEqual(resp_calc.data.get("detail"), "Sin deducciones aplicables.") r4 = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {"decision": "aceptar_con_comentario", "comentario": "OK"}, format="json", ) self.assertIn(r4.status_code, (200, 409)) def test_cerrar_requiere_campo_aunque_exista_origen_trigo_legado(self): self.boleta.campo = None self.boleta.save(update_fields=["campo", "updated_at"]) recepcion_id = self._identificar().data["id"] resp = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {}, format="json", ) self.assertEqual(resp.status_code, 422) self.assertIn("campo", str(resp.data)) def test_calcular_deducciones_requiere_datos_camion_y_bodega(self): self.boleta.campo = None self.boleta.placas_camion = "" self.boleta.chofer_nombre = "" self.boleta.save( update_fields=[ "campo", "placas_camion", "chofer_nombre", "updated_at", ] ) recepcion_id = self._identificar().data["id"] resp = self.client.post( f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json", ) self.assertEqual(resp.status_code, 422) errors = resp.data.get("errors", resp.data) self.assertIn("campo", errors) self.assertIn("placas_camion", errors) self.assertIn("chofer_nombre", errors) def test_calcular_deducciones_no_requiere_origen_trigo_legado(self): self.boleta.origen_trigo = "" self.boleta.save(update_fields=["origen_trigo", "updated_at"]) recepcion_id = self._identificar().data["id"] resp = self.client.post( f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json", ) self.assertEqual(resp.status_code, 422) errors = resp.data.get("errors", resp.data) self.assertNotIn("campo", errors) self.assertNotIn("origen_trigo", errors) self.assertIn("peso_lleno", errors) def test_cerrar_requires_firma_chofer_when_flag_is_true(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.patch( f"/recepciones/{recepcion_id}/", {"chofer_firma_en_dispositivo": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impureza", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="3.000", descontar_kg_por_tonelada="0.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="2.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "900.00", "es_automatico": True}, format="json", ) resp_calc = self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") self.assertEqual(resp_calc.status_code, 200) self.assertEqual(resp_calc.data.get("detail"), "Sin deducciones aplicables.") resp_cerrar = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {"decision": "aceptar_con_comentario", "comentario": "OK"}, format="json", ) self.assertEqual(resp_cerrar.status_code, 422) self.assertEqual(resp_cerrar.data.get("detail"), "Se requiere firma del chofer.") def test_subir_foto_y_descargar(self): resp = self._identificar() recepcion_id = resp.data["id"] from PIL import Image buf = BytesIO() Image.new("RGB", (1, 1), color=(255, 0, 0)).save(buf, format="PNG") buf.seek(0) upload = SimpleUploadedFile("foto.png", buf.read(), content_type="image/png") resp_foto = self.client.post( f"/recepciones/{recepcion_id}/fotos/", {"tipo": "frontal", "imagen": upload}, format="multipart", ) self.assertEqual(resp_foto.status_code, 201) recepcion = Recepcion.objects.get(pk=recepcion_id) foto = recepcion.fotos.first() self.assertIsNotNone(foto) resp_dl = self.client.get(f"/recepciones/fotos/{foto.id}/download/") self.assertEqual(resp_dl.status_code, 200) def test_subir_foto_extension_invalida(self): resp = self._identificar() recepcion_id = resp.data["id"] upload = SimpleUploadedFile("foto.gif", b"GIF89a", content_type="image/gif") resp_foto = self.client.post( f"/recepciones/{recepcion_id}/fotos/", {"tipo": "frontal", "imagen": upload}, format="multipart", ) self.assertEqual(resp_foto.status_code, 422) def test_peso_lleno_sigue_funcionando_sin_bascula(self): resp = self._identificar() recepcion_id = resp.data["id"] resp_lleno = self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_lleno.status_code, 201) self.assertIsNone(resp_lleno.data.get("bascula")) @override_settings(BASCULA_ACCESS_TOKEN="scale-token-123") def test_peso_lleno_desde_bascula_sin_autenticacion(self): resp = self._identificar() recepcion_id = resp.data["id"] bascula = Bascula.objects.create( codigo="B01", modelo="Modelo X", oficina=self.oficina, ) anon_client = APIClient() # Sin credenciales de usuario y sin token de báscula: debe rechazarse. resp_forbidden = anon_client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00"}, format="json", ) self.assertIn(resp_forbidden.status_code, (401, 403)) # Token incorrecto: debe rechazarse también. resp_bad = anon_client.post( f"/recepciones/{recepcion_id}/peso-lleno/", { "bascula_id": bascula.id, "access_token": "wrong", "valor": "1000.00", }, format="json", ) self.assertIn(resp_bad.status_code, (401, 403)) # Báscula con token correcto, sin usuario autenticado: debe registrarse. resp_ok = anon_client.post( f"/recepciones/{recepcion_id}/peso-lleno/", { "bascula_id": bascula.id, "access_token": "scale-token-123", "valor": "1000.00", }, format="json", ) self.assertEqual(resp_ok.status_code, 201) self.assertEqual(resp_ok.data["valor"], "1000.00") self.assertTrue(resp_ok.data["es_automatico"]) self.assertEqual(resp_ok.data["bascula"], bascula.id) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertEqual(recepcion.estado, Recepcion.ESTADO_PESO_LLENO) @override_settings(BASCULA_ACCESS_TOKEN="scale-token-123") def test_peso_lleno_por_qr_desde_bascula(self): resp = self._identificar() recepcion_id = resp.data["id"] bascula = Bascula.objects.create( codigo="B02", modelo="Modelo Y", oficina=self.oficina, ) anon_client = APIClient() # Boleta inexistente en el QR: 422 (validación) porque resolve_recepcion lanza ValidationError. resp_not_found = anon_client.post( "/recepciones/peso-lleno/", { "bascula_id": bascula.id, "access_token": "scale-token-123", "valor": "1000.00", "qr": "zafra:999999|folio:no-existe", }, format="json", ) self.assertEqual(resp_not_found.status_code, 422) # QR real de la boleta identificada: debe resolver la recepción y registrar el peso. resp_ok = anon_client.post( "/recepciones/peso-lleno/", { "bascula_id": bascula.id, "access_token": "scale-token-123", "valor": "1500.00", "qr": self.boleta.qr_texto, }, format="json", ) self.assertEqual(resp_ok.status_code, 201) self.assertEqual(resp_ok.data["valor"], "1500.00") self.assertTrue(resp_ok.data["es_automatico"]) self.assertEqual(resp_ok.data["bascula"], bascula.id) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertEqual(recepcion.estado, Recepcion.ESTADO_PESO_LLENO) # Ya existe peso lleno: debe rechazarse si se intenta de nuevo. resp_dup = anon_client.post( "/recepciones/peso-lleno/", { "bascula_id": bascula.id, "access_token": "scale-token-123", "valor": "1500.00", "qr": self.boleta.qr_texto, }, format="json", ) self.assertEqual(resp_dup.status_code, 422) @override_settings(BASCULA_ACCESS_TOKEN="scale-token-123") def test_peso_lleno_por_qr_requiere_qr_o_folio(self): bascula = Bascula.objects.create( codigo="B03", modelo="Modelo Z", oficina=self.oficina, ) anon_client = APIClient() resp = anon_client.post( "/recepciones/peso-lleno/", { "bascula_id": bascula.id, "access_token": "scale-token-123", "valor": "1000.00", }, format="json", ) self.assertEqual(resp.status_code, 422) def _crear_recepcion_cerrada(self): resp = self._identificar() recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.calidad_calculada = True recepcion.save() self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "900.00", "es_automatico": True}, format="json", ) resp_cerrar = self.client.post( f"/recepciones/{recepcion_id}/cerrar/", {"decision": "aceptar_con_comentario", "comentario": "OK"}, format="json", ) self.assertEqual(resp_cerrar.status_code, 200) return recepcion_id def _login_molino(self): molino_group, _ = Group.objects.get_or_create(name="Molino") validate_perm = Permission.objects.get( codename="validate", content_type__app_label="recepciones" ) reopen_perm = Permission.objects.get( codename="reopen", content_type__app_label="recepciones" ) resolve_dispute_perm = Permission.objects.get( codename="resolve_dispute", content_type__app_label="recepciones" ) molino_group.permissions.set([validate_perm, reopen_perm, resolve_dispute_perm]) molino_user = User.objects.create_user( username="molino", email="molino@example.com", password="molino12345", ) molino_user.groups.add(molino_group) molino_user.profile.oficinas.add(self.oficina) resp = self.client.post( "/auth/token/", {"username": "molino", "password": "molino12345"}, format="json", ) self.assertEqual(resp.status_code, 200) token = resp.data["access"] self.client.credentials( HTTP_AUTHORIZATION=f"Bearer {token}", HTTP_X_OFFICE_ID=str(self.oficina.id), ) return molino_user def test_validar_recepcion_cerrada_pasa_a_validada(self): recepcion_id = self._crear_recepcion_cerrada() resp = self.client.post( f"/recepciones/{recepcion_id}/validar/", {}, format="json", ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["estado"], Recepcion.ESTADO_VALIDADA) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertIsNotNone(recepcion.fecha_validacion) self.assertIsNone(recepcion.fecha_edicion) log_validada = recepcion.logs.filter(accion=RecepcionLog.ACCION_VALIDADA).first() self.assertIsNotNone(log_validada) self.assertEqual(log_validada.estado_anterior, Recepcion.ESTADO_CERRADA) self.assertEqual(log_validada.estado_nuevo, Recepcion.ESTADO_VALIDADA) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_CERRADA) def test_boleta_entrada_pdf_disponible_en_recepcion_validada_o_editada(self): recepcion_id = self._crear_recepcion_cerrada() recepcion = Recepcion.objects.get(pk=recepcion_id) with patch("apps.recepciones.views.build_boleta_entrada_pdf_bytes", return_value=b"%PDF-test"): for estado in (Recepcion.ESTADO_VALIDADA, Recepcion.ESTADO_EDITADA): with self.subTest(estado=estado): recepcion.estado = estado recepcion.save(update_fields=["estado", "updated_at"]) resp = self.client.get(f"/recepciones/{recepcion_id}/boleta-entrada-pdf/") self.assertEqual(resp.status_code, 200) self.assertEqual(resp["Content-Type"], "application/pdf") def test_validar_recepcion_reabierta_pasa_a_editada(self): recepcion_id = self._crear_recepcion_cerrada() self.client.post(f"/recepciones/{recepcion_id}/validar/", {}, format="json") resp_reabrir = self.client.post( f"/recepciones/{recepcion_id}/reabrir/", {"motivo": "Corrección de peso"}, format="json", ) self.assertEqual(resp_reabrir.status_code, 200) self.assertEqual(resp_reabrir.data["estado"], Recepcion.ESTADO_BORRADOR) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_ASIGNADA) resp_validar = self.client.post( f"/recepciones/{recepcion_id}/validar/", {}, format="json", ) self.assertEqual(resp_validar.status_code, 200) self.assertEqual(resp_validar.data["estado"], Recepcion.ESTADO_EDITADA) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertIsNotNone(recepcion.fecha_validacion) self.assertIsNotNone(recepcion.fecha_edicion) log_editada = recepcion.logs.filter(accion=RecepcionLog.ACCION_EDITADA).first() self.assertIsNotNone(log_editada) self.assertEqual(log_editada.estado_anterior, Recepcion.ESTADO_BORRADOR) self.assertEqual(log_editada.estado_nuevo, Recepcion.ESTADO_EDITADA) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_CERRADA) def test_reabrir_requiere_motivo(self): recepcion_id = self._crear_recepcion_cerrada() self.client.post(f"/recepciones/{recepcion_id}/validar/", {}, format="json") resp = self.client.post( f"/recepciones/{recepcion_id}/reabrir/", {}, format="json", ) self.assertEqual(resp.status_code, 422) self.assertIn("motivo", resp.data.get("errors", {})) def test_reabrir_desde_estado_no_validado_rechazado(self): recepcion_id = self._crear_recepcion_cerrada() resp = self.client.post( f"/recepciones/{recepcion_id}/reabrir/", {"motivo": "No debería funcionar"}, format="json", ) self.assertEqual(resp.status_code, 422) self.assertIn("estado", resp.data.get("errors", {})) def test_validar_requiere_permiso(self): recepcion_id = self._crear_recepcion_cerrada() self._login_operador(with_process_perm=False) resp = self.client.post( f"/recepciones/{recepcion_id}/validar/", {}, format="json", ) self.assertEqual(resp.status_code, 403) def test_molino_puede_validar_y_reabrir(self): recepcion_id = self._crear_recepcion_cerrada() self._login_molino() resp_validar = self.client.post( f"/recepciones/{recepcion_id}/validar/", {}, format="json", ) self.assertEqual(resp_validar.status_code, 200) self.assertEqual(resp_validar.data["estado"], Recepcion.ESTADO_VALIDADA) resp_reabrir = self.client.post( f"/recepciones/{recepcion_id}/reabrir/", {"motivo": "Ajuste"}, format="json", ) self.assertEqual(resp_reabrir.status_code, 200) self.assertEqual(resp_reabrir.data["estado"], Recepcion.ESTADO_BORRADOR) def test_listado_excluye_validadas_y_editadas_por_defecto(self): recepcion_id = self._crear_recepcion_cerrada() self.client.post(f"/recepciones/{recepcion_id}/validar/", {}, format="json") resp = self.client.get("/recepciones/", format="json") self.assertEqual(resp.status_code, 200) ids = [r["id"] for r in resp.data.get("results", resp.data)] self.assertNotIn(recepcion_id, ids) def test_listado_excluye_rechazadas_por_defecto(self): recepcion_id = self._identificar().data["id"] Recepcion.objects.filter(pk=recepcion_id).update(estado=Recepcion.ESTADO_RECHAZADA) resp = self.client.get("/recepciones/", format="json") self.assertEqual(resp.status_code, 200) ids = [r["id"] for r in resp.data.get("results", resp.data)] self.assertNotIn(recepcion_id, ids) resp_filtrada = self.client.get( "/recepciones/", {"estado": Recepcion.ESTADO_RECHAZADA}, format="json", ) self.assertEqual(resp_filtrada.status_code, 200) ids_filtrados = [r["id"] for r in resp_filtrada.data.get("results", resp_filtrada.data)] self.assertIn(recepcion_id, ids_filtrados) def test_listado_permite_filtrar_validadas_y_editadas(self): recepcion_id = self._crear_recepcion_cerrada() self.client.post(f"/recepciones/{recepcion_id}/validar/", {}, format="json") resp = self.client.get( "/recepciones/", {"estado": Recepcion.ESTADO_VALIDADA}, format="json", ) self.assertEqual(resp.status_code, 200) ids = [r["id"] for r in resp.data.get("results", resp.data)] self.assertIn(recepcion_id, ids) def _crear_boleta_sin_contrato(self): folio = build_folio(self.oficina.codigo, self.productor.id, self.zafra.fecha_inicio.year, 2) return Boleta.objects.create( zafra=self.zafra, oficina=self.oficina, productor=self.productor, contrato=None, folio=folio, qr_texto=build_qr_text(self.zafra.id, folio), peso_estimado=100, campo=self.campo, origen_trigo="Campo 1", placas_camion="ABC-124", marca_unidad="Kenworth", modelo_unidad="T680", chofer_nombre="Chofer 1", chofer_telefono="555-0001", linea_nombre="Línea 1", created_by=self.admin, updated_by=self.admin, ) def _crear_recepcion_con_rechazo_calidad(self, boleta=None): boleta = boleta or self.boleta resp = self.client.post( "/recepciones/identificar/", {"qr": boleta.qr_texto, "bodega_destino": self.bodega.id}, format="json", ) self.assertEqual(resp.status_code, 200) recepcion_id = resp.data["id"] self.client.post( f"/recepciones/{recepcion_id}/peso-lleno/", {"valor": "11000.00", "es_automatico": True}, format="json", ) self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) recepcion = Recepcion.objects.get(pk=recepcion_id) muestreo = recepcion.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impurezas", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="4.100", descontar_kg_por_tonelada="10.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="4.200", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() recepcion.estado = Recepcion.ESTADO_MUESTREO_COMPLETADO recepcion.save() self.client.post( f"/recepciones/{recepcion_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.client.post(f"/recepciones/{recepcion_id}/calcular-deducciones/", {}, format="json") return recepcion_id def test_rechazo_calidad_con_contrato_va_a_en_disputa(self): recepcion_id = self._crear_recepcion_con_rechazo_calidad() recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertEqual(recepcion.estado, Recepcion.ESTADO_EN_DISPUTA) self.assertTrue(recepcion.rechazo_calidad) def test_rechazo_calidad_sin_contrato_es_rechazada(self): boleta = self._crear_boleta_sin_contrato() recepcion_id = self._crear_recepcion_con_rechazo_calidad(boleta=boleta) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertEqual(recepcion.estado, Recepcion.ESTADO_RECHAZADA) self.assertIsNotNone(recepcion.fecha_cierre) boleta.refresh_from_db() self.assertEqual(boleta.estado, Boleta.ESTADO_RECHAZADA) def test_resolver_disputa_autorizar(self): recepcion_id = self._crear_recepcion_con_rechazo_calidad() self._login_molino() resp = self.client.post( f"/recepciones/{recepcion_id}/resolver-disputa/", {"decision": "autorizar", "comentario": "Se autoriza la entrada."}, format="json", ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["estado"], Recepcion.ESTADO_CERRADA_EN_DISPUTA) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertIsNotNone(recepcion.fecha_cierre) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_RECIBIDA) def test_resolver_disputa_rechazar(self): recepcion_id = self._crear_recepcion_con_rechazo_calidad() self._login_molino() resp = self.client.post( f"/recepciones/{recepcion_id}/resolver-disputa/", {"decision": "rechazar", "comentario": "Se rechaza la entrada."}, format="json", ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["estado"], Recepcion.ESTADO_RECHAZADA_EN_DISPUTA) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertIsNotNone(recepcion.fecha_cierre) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_RECHAZADA) def test_validar_rechazada_en_disputa_requiere_bodega_rechazo(self): recepcion_id = self._crear_recepcion_con_rechazo_calidad() self._login_molino() self.client.post( f"/recepciones/{recepcion_id}/resolver-disputa/", {"decision": "rechazar"}, format="json", ) resp = self.client.post(f"/recepciones/{recepcion_id}/validar/", {}, format="json") self.assertEqual(resp.status_code, 422) self.assertIn("bodega_rechazo", resp.data.get("errors", {})) def test_validar_rechazada_en_disputa_con_bodega_rechazo(self): recepcion_id = self._crear_recepcion_con_rechazo_calidad() bodega_rechazo = Bodega.objects.create( codigo="BR-1", nombre="Bodega Rechazo", oficina=self.oficina, activa=True, es_rechazo=True, codigo_sap="SAP-RECHAZO", created_by=self.admin, updated_by=self.admin, ) self._login_molino() self.client.post( f"/recepciones/{recepcion_id}/resolver-disputa/", {"decision": "rechazar"}, format="json", ) resp = self.client.post(f"/recepciones/{recepcion_id}/validar/", {}, format="json") self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["estado"], Recepcion.ESTADO_VALIDADA) recepcion = Recepcion.objects.get(pk=recepcion_id) self.assertIsNotNone(recepcion.fecha_validacion) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_RECHAZADA) def test_validar_cerrada_en_disputa(self): recepcion_id = self._crear_recepcion_con_rechazo_calidad() self._login_molino() self.client.post( f"/recepciones/{recepcion_id}/resolver-disputa/", {"decision": "autorizar"}, format="json", ) resp = self.client.post(f"/recepciones/{recepcion_id}/validar/", {}, format="json") self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["estado"], Recepcion.ESTADO_VALIDADA) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_CERRADA) def test_resolver_disputa_requiere_permiso(self): recepcion_id = self._crear_recepcion_con_rechazo_calidad() self._login_operador(with_process_perm=True) resp = self.client.post( f"/recepciones/{recepcion_id}/resolver-disputa/", {"decision": "autorizar"}, format="json", ) self.assertEqual(resp.status_code, 403) def _crear_segunda_boleta(self, variedad=None): folio = build_folio(self.oficina.codigo, self.productor.id, self.zafra.fecha_inicio.year, 2) variedad = variedad or self.variedad return Boleta.objects.create( zafra=self.zafra, oficina=self.oficina, productor=self.productor, contrato=self.contrato, folio=folio, qr_texto=build_qr_text(self.zafra.id, folio), peso_estimado=100, campo=self.campo, origen_trigo="Campo 1", placas_camion="ABC-125", marca_unidad="Kenworth", modelo_unidad="T680", chofer_nombre="Chofer 1", chofer_telefono="555-0001", linea_nombre="Línea 1", variedad=variedad, created_by=self.admin, updated_by=self.admin, ) def test_flujo_flete_dos_boletas(self): boleta2 = self._crear_segunda_boleta() resp1 = self.client.post( "/recepciones/identificar/", {"qr": self.boleta.qr_texto, "bodega_destino": self.bodega.id}, format="json", ) self.assertEqual(resp1.status_code, 200) recepcion1_id = resp1.data["id"] resp2 = self.client.post( f"/recepciones/{recepcion1_id}/agregar-boleta/", {"qr": boleta2.qr_texto}, format="json", ) self.assertEqual(resp2.status_code, 200) flete_id = resp2.data["id"] self.assertEqual(len(resp2.data["recepciones"]), 2) flete = Flete.objects.get(pk=flete_id) self.assertEqual(flete.recepciones.count(), 2) # Registrar datos del camión self.client.patch( f"/recepciones/{recepcion1_id}/", {"placas_camion": "ABC-123", "chofer_nombre": "Juan Pérez"}, format="json", ) # Pesos a nivel flete self.client.post( f"/recepciones/fletes/{flete_id}/peso-lleno/", {"valor": "11000.00", "es_automatico": True}, format="json", ) # Muestreo a nivel flete self.client.post( f"/recepciones/fletes/{flete_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) muestreo = flete.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) p_imp = Parametro.objects.create( nombre="Impureza", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ) Castigo.objects.create( parametro=p_imp, limite_superior="3.000", descontar_kg_por_tonelada="0.00", bonificar_por_tonelada="0.00", created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=p_imp, valor="2.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() self.client.post( f"/recepciones/fletes/{flete_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) recepciones = list(flete.recepciones.order_by("id").all()) recepcion2_id = recepciones[1].id # Distribuir peso neto (10,000 kg) entre ambas recepciones self.client.post( f"/recepciones/fletes/{flete_id}/distribuir-peso/", { "distribuciones": [ {"recepcion_id": recepcion1_id, "peso_neto_asignado": "6000.000"}, {"recepcion_id": recepcion2_id, "peso_neto_asignado": "4000.000"}, ] }, format="json", ) # Calcular deducciones para todas las recepciones del flete resp_deducciones = self.client.post( f"/recepciones/fletes/{flete_id}/calcular-deducciones/", {}, format="json", ) self.assertEqual(resp_deducciones.status_code, 200) self.assertEqual(resp_deducciones.data["flete_id"], flete_id) self.assertEqual(len(resp_deducciones.data["resultados"]), 2) self.assertEqual( {item["recepcion_id"] for item in resp_deducciones.data["resultados"]}, {recepcion1_id, recepcion2_id}, ) self.assertTrue( all(item["calculo_calidad"]["estado"] == "sin_deducciones" for item in resp_deducciones.data["resultados"]) ) # Cerrar flete resp_cerrar = self.client.post( f"/recepciones/fletes/{flete_id}/cerrar/", {"decision": "aceptar_con_comentario", "comentario": "OK"}, format="json", ) self.assertEqual(resp_cerrar.status_code, 200) self.assertEqual(resp_cerrar.data["estado"], Recepcion.ESTADO_CERRADA) flete.refresh_from_db() for r in flete.recepciones.all(): self.assertEqual(r.estado, Recepcion.ESTADO_CERRADA) self.assertIsNotNone(r.peso_neto_asignado) # Validar flete self._login_molino() resp_validar = self.client.post( f"/recepciones/fletes/{flete_id}/validar/", {}, format="json", ) self.assertEqual(resp_validar.status_code, 200) self.assertEqual(resp_validar.data["estado"], Recepcion.ESTADO_VALIDADA) flete.refresh_from_db() for r in flete.recepciones.all(): self.assertEqual(r.estado, Recepcion.ESTADO_VALIDADA) self.boleta.refresh_from_db() self.assertEqual(self.boleta.estado, Boleta.ESTADO_CERRADA) def test_pesos_flete_via_endpoints_clasicos(self): boleta2 = self._crear_segunda_boleta() resp1 = self.client.post( "/recepciones/identificar/", {"qr": self.boleta.qr_texto, "bodega_destino": self.bodega.id}, format="json", ) self.assertEqual(resp1.status_code, 200) recepcion1_id = resp1.data["id"] resp2 = self.client.post( f"/recepciones/{recepcion1_id}/agregar-boleta/", {"qr": boleta2.qr_texto}, format="json", ) self.assertEqual(resp2.status_code, 200) flete_id = resp2.data["id"] flete = Flete.objects.get(pk=flete_id) recepciones = list(flete.recepciones.order_by("id")) recepcion2_id = recepciones[1].id # Peso lleno mediante endpoint clásico (recepción hija) resp_lleno = self.client.post( f"/recepciones/{recepcion2_id}/peso-lleno/", {"valor": "5000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_lleno.status_code, 201) self.assertEqual(resp_lleno.data["tipo"], "lleno") self.assertEqual(resp_lleno.data["valor"], "5000.00") flete.refresh_from_db() self.assertEqual(flete.estado, Recepcion.ESTADO_PESO_LLENO) self.assertTrue(flete.pesos.filter(tipo="lleno").exists()) for r in flete.recepciones.all(): self.assertEqual(r.estado, Recepcion.ESTADO_PESO_LLENO) # Muestreo y completado self.client.post( f"/recepciones/fletes/{flete_id}/generar-muestreo/", {"tipo_muestreo": "aleatorio"}, format="json", ) muestreo = flete.muestreo resultado = ResultadoLaboratorio.objects.create( muestreo=muestreo, created_by=self.admin, updated_by=self.admin, ) ParametroResultado.objects.create( resultado_laboratorio=resultado, parametro=Parametro.objects.create( nombre="Impureza", unidad="%", activo=True, created_by=self.admin, updated_by=self.admin, ), valor="2.000", created_by=self.admin, updated_by=self.admin, ) muestreo.estado = "completado" muestreo.save() # Peso vacío mediante endpoint clásico (recepción madre) resp_vacio = self.client.post( f"/recepciones/{recepcion1_id}/peso-vacio/", {"valor": "1000.00", "es_automatico": True}, format="json", ) self.assertEqual(resp_vacio.status_code, 201) self.assertEqual(resp_vacio.data["tipo"], "vacio") self.assertEqual(resp_vacio.data["valor"], "1000.00") self.assertEqual(resp_vacio.data["peso_bruto"], "4000.00") flete.refresh_from_db() self.assertEqual(flete.estado, Recepcion.ESTADO_PESO_VACIO) self.assertTrue(flete.pesos.filter(tipo="vacio").exists()) for r in flete.recepciones.all(): self.assertEqual(r.estado, Recepcion.ESTADO_PESO_VACIO) def test_reabrir_recepcion_de_flete_reabre_todo_el_flete(self): boleta2 = self._crear_segunda_boleta() recepcion1_id = self.client.post( "/recepciones/identificar/", {"qr": self.boleta.qr_texto, "bodega_destino": self.bodega.id}, format="json", ).data["id"] flete_id = self.client.post( f"/recepciones/{recepcion1_id}/agregar-boleta/", {"qr": boleta2.qr_texto}, format="json", ).data["id"] flete = Flete.objects.get(pk=flete_id) flete.estado = Recepcion.ESTADO_VALIDADA flete.fecha_cierre = timezone.now() flete.fecha_validacion = timezone.now() flete.save(update_fields=["estado", "fecha_cierre", "fecha_validacion", "updated_at"]) flete.recepciones.update( estado=Recepcion.ESTADO_VALIDADA, fecha_cierre=timezone.now(), fecha_validacion=timezone.now(), ) Boleta.objects.filter(recepcion__flete=flete).update(estado=Boleta.ESTADO_CERRADA) self._login_molino() resp = self.client.post( f"/recepciones/{recepcion1_id}/reabrir/", {"motivo": "Corregir datos del flete"}, format="json", ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["estado"], Recepcion.ESTADO_BORRADOR) flete.refresh_from_db() self.assertEqual(flete.estado, Recepcion.ESTADO_BORRADOR) self.assertIsNone(flete.fecha_cierre) for recepcion in flete.recepciones.select_related("boleta"): self.assertEqual(recepcion.estado, Recepcion.ESTADO_BORRADOR) self.assertIsNone(recepcion.fecha_cierre) self.assertEqual(recepcion.boleta.estado, Boleta.ESTADO_ASIGNADA) self.assertTrue( recepcion.logs.filter( accion=RecepcionLog.ACCION_REABIERTA, detalle__motivo="Corregir datos del flete", ).exists() ) def test_agregar_boleta_variedad_diferente_falla(self): variedad2 = Variedad.objects.create(nombre="V2") boleta2 = self._crear_segunda_boleta(variedad=variedad2) resp1 = self.client.post( "/recepciones/identificar/", {"qr": self.boleta.qr_texto, "bodega_destino": self.bodega.id}, format="json", ) self.assertEqual(resp1.status_code, 200) recepcion1_id = resp1.data["id"] resp2 = self.client.post( f"/recepciones/{recepcion1_id}/agregar-boleta/", {"qr": boleta2.qr_texto}, format="json", ) self.assertEqual(resp2.status_code, 422) self.assertIn("variedad", str(resp2.data).lower())