from datetime import date from io import BytesIO from zipfile import ZipFile from django.contrib.auth import get_user_model from rest_framework.test import APITestCase from apps.boletas.models import Boleta, BoletaLog from apps.boletas.utils import build_folio from apps.catalogos.models import Bodega, Contrato, ContratoVariedad, Oficina, Productor, Variedad, Zafra from apps.recepciones.models import Recepcion, RecepcionLog User = get_user_model() class GeneracionBoletasTests(APITestCase): def setUp(self): self.admin = User.objects.create_superuser( username="admin", email="admin@example.com", password="admin12345", ) self.oficina = Oficina.objects.create(codigo="OF-1", nombre="Oficina 1", activa=True) self.zafra = Zafra.objects.create( nombre="Z 2026", fecha_inicio="2026-01-01", fecha_fin="2026-12-31", oficina=self.oficina, ) self.variedad = Variedad.objects.create(nombre="V1", oficina=self.oficina) 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, oficina=self.oficina, activo=True, ) ContratoVariedad.objects.create( contrato=self.contrato, variedad=self.variedad, peso_teorico=None, ) resp = self.client.post( "/auth/token/", {"username": "admin", "password": "admin12345"}, format="json", ) self.token = resp.data["access"] self.client.credentials( HTTP_AUTHORIZATION=f"Bearer {self.token}", HTTP_X_OFFICE_ID=str(self.oficina.id), ) def test_generar_boletas_por_productor(self): resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 2, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) self.assertEqual(resp.status_code, 201) self.assertEqual(resp.data["cantidad_boletas"], 2) self.assertEqual(Boleta.objects.count(), 2) self.assertFalse(Boleta.objects.exclude(contrato__isnull=True).exists()) self.assertFalse(Boleta.objects.exclude(variedad__isnull=True).exists()) folios = list(Boleta.objects.values_list("folio", flat=True).order_by("folio")) expected_folio = build_folio(self.oficina.codigo, self.productor.id, self.zafra.fecha_inicio.year, 1) self.assertEqual( folios, [expected_folio, build_folio(self.oficina.codigo, self.productor.id, self.zafra.fecha_inicio.year, 2)], ) resp2 = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 1, "generar_qr": True, }, format="json", ) self.assertEqual(resp2.status_code, 201) folios2 = list(Boleta.objects.values_list("folio", flat=True).order_by("folio")) self.assertEqual( folios2, [ expected_folio, build_folio(self.oficina.codigo, self.productor.id, self.zafra.fecha_inicio.year, 2), build_folio(self.oficina.codigo, self.productor.id, self.zafra.fecha_inicio.year, 3), ], ) def test_generar_boletas_por_contrato_asigna_contrato_y_variedad(self): resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "contrato_ids": [self.contrato.id], "cantidad_boletas_por_productor": 1, }, format="json", ) self.assertEqual(resp.status_code, 201) boleta = Boleta.objects.get(productor=self.productor) self.assertEqual(boleta.contrato, self.contrato) self.assertEqual(boleta.variedad, self.variedad) def test_generar_boletas_por_productor_sin_contrato(self): productor = Productor.objects.create(nombre="Prod sin contrato", rfc="DEF123456T12") resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [productor.id], "cantidad_boletas_por_productor": 1, }, format="json", ) self.assertEqual(resp.status_code, 201) boleta = Boleta.objects.get(productor=productor) self.assertIsNone(boleta.contrato) self.assertIsNone(boleta.variedad) def test_generar_boletas_por_productor_con_varios_contratos(self): Contrato.objects.create( numero_contrato="C-2", productor=self.productor, zafra=self.zafra, oficina=self.oficina, activo=True, ) resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 1, }, format="json", ) self.assertEqual(resp.status_code, 201) boleta = Boleta.objects.get(productor=self.productor) self.assertIsNone(boleta.contrato) self.assertIsNone(boleta.variedad) def test_generar_boletas_con_bodega_destino(self): bodega = Bodega.objects.create(codigo="B-1", nombre="Bodega 1", oficina=self.oficina, activa=True) resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 1, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) self.assertEqual(resp.status_code, 201) boleta = Boleta.objects.order_by("-id").first() self.assertIsNotNone(boleta) resp2 = self.client.post( "/recepciones/identificar/", { "folio": str(boleta.folio), "zafra_id": self.zafra.id, "bodega_destino": bodega.id, }, format="json", ) self.assertEqual(resp2.status_code, 200) boleta.refresh_from_db() self.assertTrue(hasattr(boleta, "recepcion")) self.assertEqual(boleta.recepcion.bodega_destino_id, bodega.id) def test_imprimir_masivo_zip(self): resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 2, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) self.assertEqual(resp.status_code, 201) boleta_ids = resp.data["boleta_ids"] zip_resp = self.client.post( "/boletas/imprimir-masivo/", {"boleta_ids": boleta_ids}, format="json", ) self.assertEqual(zip_resp.status_code, 200) self.assertIn("application/zip", zip_resp["Content-Type"]) zf = ZipFile(BytesIO(zip_resp.content)) self.assertEqual(len(zf.namelist()), 2) def test_generar_zip_endpoint(self): zip_resp = self.client.post( "/boletas/generar-zip/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 2, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) self.assertEqual(zip_resp.status_code, 200) self.assertIn("application/zip", zip_resp["Content-Type"]) zf = ZipFile(BytesIO(zip_resp.content)) self.assertEqual(len(zf.namelist()), 2) def test_descargar_rango_zip(self): resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 2, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) self.assertEqual(resp.status_code, 201) today = str(date.today()) zip_resp = self.client.get( "/boletas/descargar-rango/", {"fecha_inicio": today, "fecha_fin": today}, ) self.assertEqual(zip_resp.status_code, 200) self.assertIn("application/zip", zip_resp["Content-Type"]) zf = ZipFile(BytesIO(zip_resp.content)) self.assertEqual(len(zf.namelist()), 2) def test_descargar_rango_zip_filtra_por_oficina(self): self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 2, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) oficina2 = Oficina.objects.create(codigo="OF-2", nombre="Oficina 2", activa=True) zafra2 = Zafra.objects.create( nombre="Z 2026-2", fecha_inicio="2026-01-01", fecha_fin="2026-12-31", oficina=oficina2, ) variedad2 = Variedad.objects.create(nombre="V2", oficina=oficina2) contrato2 = Contrato.objects.create( numero_contrato="C-2", productor=self.productor, zafra=zafra2, oficina=oficina2, activo=True, ) ContratoVariedad.objects.create(contrato=contrato2, variedad=variedad2) Boleta.objects.create( zafra=zafra2, oficina=oficina2, productor=self.productor, contrato=contrato2, folio="B-OF2-1", ) today = str(date.today()) zip_resp = self.client.get( "/boletas/descargar-rango/", {"fecha_inicio": today, "fecha_fin": today}, ) self.assertEqual(zip_resp.status_code, 200) zf = ZipFile(BytesIO(zip_resp.content)) self.assertEqual(len(zf.namelist()), 2) for name in zf.namelist(): self.assertNotIn("OF2", name, msg=name) def test_descargar_rango_zip_requiere_oficina(self): self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.token}") resp = self.client.get( "/boletas/descargar-rango/", {"fecha_inicio": "2026-01-01", "fecha_fin": "2026-12-31"}, ) self.assertEqual(resp.status_code, 400) self.assertIn("office_id", resp.data) def test_list_oculta_boletas_cerradas_salvo_filtro_explicito(self): activa = Boleta.objects.create( zafra=self.zafra, oficina=self.oficina, productor=self.productor, contrato=self.contrato, folio="B-ACT-1", estado=Boleta.ESTADO_GENERADA, created_by=self.admin, updated_by=self.admin, ) cerrada = Boleta.objects.create( zafra=self.zafra, oficina=self.oficina, productor=self.productor, contrato=self.contrato, folio="B-CER-1", estado=Boleta.ESTADO_CERRADA, created_by=self.admin, updated_by=self.admin, ) resp = self.client.get("/boletas/", format="json") self.assertEqual(resp.status_code, 200) ids = [item["id"] for item in resp.data["results"]] self.assertIn(activa.id, ids) self.assertNotIn(cerrada.id, ids) resp_filtrada = self.client.get("/boletas/", {"estado": Boleta.ESTADO_CERRADA}, format="json") self.assertEqual(resp_filtrada.status_code, 200) ids_filtrados = [item["id"] for item in resp_filtrada.data["results"]] self.assertIn(cerrada.id, ids_filtrados) self.assertNotIn(activa.id, ids_filtrados) def test_cambio_estado_boleta_se_loguea(self): boleta = Boleta.objects.create( zafra=self.zafra, oficina=self.oficina, productor=self.productor, contrato=self.contrato, folio="B-LOG-1", estado=Boleta.ESTADO_GENERADA, created_by=self.admin, updated_by=self.admin, ) boleta.estado = Boleta.ESTADO_ASIGNADA boleta.updated_by = self.admin boleta.save() log = BoletaLog.objects.filter(boleta=boleta).first() self.assertIsNotNone(log) self.assertEqual(log.estado_anterior, Boleta.ESTADO_GENERADA) self.assertEqual(log.estado_nuevo, Boleta.ESTADO_ASIGNADA) self.assertEqual(log.usuario, self.admin) def test_historial_estados_incluye_boleta_y_recepcion(self): bodega = Bodega.objects.create(codigo="B-1", nombre="Bodega 1", oficina=self.oficina, activa=True) resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 1, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) self.assertEqual(resp.status_code, 201) boleta_id = resp.data["boleta_ids"][0] boleta = Boleta.objects.get(pk=boleta_id) self.client.post( "/recepciones/identificar/", {"folio": str(boleta.folio), "zafra_id": self.zafra.id, "bodega_destino": bodega.id}, format="json", ) recepcion = boleta.recepcion boleta.estado = Boleta.ESTADO_ASIGNADA boleta.updated_by = self.admin boleta.save() recepcion.estado = Recepcion.ESTADO_CERRADA recepcion.updated_by = self.admin recepcion.save() resp = self.client.get(f"/boletas/{boleta_id}/historial-estados/", format="json") self.assertEqual(resp.status_code, 200) tipos = {entry["tipo"] for entry in resp.data} self.assertIn("boleta", tipos) self.assertIn("recepcion", tipos) recepcion_acciones = { entry["accion"] for entry in resp.data if entry["tipo"] == "recepcion" } self.assertIn(RecepcionLog.ACCION_CAMBIO_ESTADO, recepcion_acciones) def _crear_boleta_con_recepcion(self): bodega = Bodega.objects.create(codigo="B-1", nombre="Bodega 1", oficina=self.oficina, activa=True) resp = self.client.post( "/boletas/generar/", { "zafra_id": self.zafra.id, "productor_ids": [self.productor.id], "cantidad_boletas_por_productor": 1, "generar_qr": True, "enviar_email": False, "imprimir_pdf": False, }, format="json", ) self.assertEqual(resp.status_code, 201) boleta_id = resp.data["boleta_ids"][0] boleta = Boleta.objects.get(pk=boleta_id) self.client.post( "/recepciones/identificar/", {"folio": str(boleta.folio), "zafra_id": self.zafra.id, "bodega_destino": bodega.id}, format="json", ) return boleta, boleta.recepcion def test_historial_global_filtra_por_boleta(self): boleta, recepcion = self._crear_boleta_con_recepcion() recepcion.estado = Recepcion.ESTADO_CERRADA recepcion.updated_by = self.admin recepcion.save() boleta.estado = Boleta.ESTADO_ASIGNADA boleta.updated_by = self.admin boleta.save() resp = self.client.get( "/boletas/historial-estados/", {"boleta_id": boleta.id}, format="json", ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["count"], 2) tipos = {entry["tipo"] for entry in resp.data["results"]} self.assertIn("boleta", tipos) self.assertIn("recepcion", tipos) def test_historial_global_filtra_por_tipo(self): boleta, recepcion = self._crear_boleta_con_recepcion() recepcion.estado = Recepcion.ESTADO_CERRADA recepcion.updated_by = self.admin recepcion.save() boleta.estado = Boleta.ESTADO_ASIGNADA boleta.updated_by = self.admin boleta.save() resp = self.client.get( "/boletas/historial-estados/", {"tipo": "recepcion"}, format="json", ) self.assertEqual(resp.status_code, 200) tipos = {entry["tipo"] for entry in resp.data["results"]} self.assertEqual(tipos, {"recepcion"}) def test_historial_global_paginado(self): boleta, recepcion = self._crear_boleta_con_recepcion() for i in range(5): boleta.estado = Boleta.ESTADO_ASIGNADA if i % 2 == 0 else Boleta.ESTADO_GENERADA boleta.updated_by = self.admin boleta.save() resp = self.client.get("/boletas/historial-estados/?page_size=3", format="json") self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data["results"]), 3)