from decimal import Decimal import shutil import tempfile from io import BytesIO from django.contrib.auth import get_user_model from django.contrib.auth.models import Group from django.core.files.uploadedfile import SimpleUploadedFile from django.test import override_settings from rest_framework.test import APIClient, APITestCase from apps.boletas.models import Boleta from apps.boletas.utils import build_qr_text from apps.catalogos.models import Contrato, ContratoVariedad, Oficina, Parametro, Productor, Variedad, Zafra from apps.muestreo_laboratorio.models import ResultadoLaboratorio from apps.recepciones.models import Recepcion from apps.muestreo_laboratorio.permissions import CALIDAD_GROUP_NAME User = get_user_model() TEMP_MEDIA_ROOT = tempfile.mkdtemp() @override_settings(MEDIA_ROOT=TEMP_MEDIA_ROOT) class MuestreoLaboratorioFlowTests(APITestCase): @classmethod def tearDownClass(cls): super().tearDownClass() shutil.rmtree(TEMP_MEDIA_ROOT, ignore_errors=True) def setUp(self): self.grp_tecnicos = Group.objects.create(name="Técnicos") self.grp_labs = Group.objects.create(name="Laboratoristas") self.admin = User.objects.create_superuser( username="admin_ml", email="admin_ml@example.com", password="admin12345", ) self.tecnico = User.objects.create_user( username="tecnico1", email="tecnico1@example.com", password="tecnico12345", ) self.tecnico.groups.add(self.grp_tecnicos) self.laboratorista = User.objects.create_user( username="lab1", email="lab1@example.com", password="lab12345", ) self.laboratorista.groups.add(self.grp_labs) self.oficina = Oficina.objects.create(codigo="OF-1", nombre="Oficina 1", activa=True) self.tecnico.profile.oficinas.add(self.oficina) self.laboratorista.profile.oficinas.add(self.oficina) self.zafra = Zafra.objects.create( nombre="Z 2026 ML", fecha_inicio="2026-01-01", fecha_fin="2026-12-31", oficina=self.oficina, ) self.variedad = Variedad.objects.create(nombre="V ML", oficina=self.oficina) self.productor = Productor.objects.create(nombre="Prod ML", rfc="ABC123456T12") self.contrato = Contrato.objects.create( numero_contrato="C-ML", productor=self.productor, zafra=self.zafra, oficina=self.oficina, activo=True, ) ContratoVariedad.objects.create( contrato=self.contrato, variedad=self.variedad, created_by=self.admin, updated_by=self.admin, ) self.boleta = Boleta.objects.create( zafra=self.zafra, productor=self.productor, contrato=self.contrato, folio=10, qr_texto=build_qr_text(self.zafra.id, 10), peso_estimado=100, created_by=self.admin, updated_by=self.admin, ) def _login(self, username, password): resp = self.client.post( "/auth/token/", {"username": username, "password": password}, 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 _crear_recepcion_y_muestreo(self): self._login("admin_ml", "admin12345") resp = self.client.post( "/recepciones/identificar/", {"qr": self.boleta.qr_texto}, format="json", ) self.assertEqual(resp.status_code, 200) recepcion_id = resp.data["id"] resp_mu = self.client.post( f"/recepciones/{recepcion_id}/generar-muestreo/", {"tipo_muestreo": "manual", "tecnico_id": self.tecnico.id}, format="json", ) self.assertEqual(resp_mu.status_code, 201) return recepcion_id, resp_mu.data["id"] def test_tomar_muestra_y_subir_foto_y_descarga_protegida(self): _, muestreo_id = self._crear_recepcion_y_muestreo() self._login("tecnico1", "tecnico12345") resp_toma = self.client.post( f"/muestreo-laboratorio/ordenes/{muestreo_id}/tomar-muestra/", {"latitud": "20.000001", "longitud": "-103.000001", "submuestras": [{"n": 1}]}, format="json", ) self.assertEqual(resp_toma.status_code, 200) 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"/muestreo-laboratorio/ordenes/{muestreo_id}/fotos/", {"tipo": "general", "imagen": upload}, format="multipart", ) self.assertEqual(resp_foto.status_code, 201) foto_id = resp_foto.data["foto"]["id"] client2 = APIClient() resp_no_auth = client2.get(f"/muestreo-laboratorio/fotos-muestra/{foto_id}/download/") self.assertEqual(resp_no_auth.status_code, 401) resp_dl = self.client.get(f"/muestreo-laboratorio/fotos-muestra/{foto_id}/download/") self.assertEqual(resp_dl.status_code, 200) @override_settings(MUESTREO_LAB_PARAMETROS_REQUERIDOS=["proteina"]) def test_captura_parametros_y_publicar(self): _, muestreo_id = self._crear_recepcion_y_muestreo() self._login("lab1", "lab12345") resp_res = self.client.post( "/muestreo-laboratorio/resultados/", {"muestreo_id": muestreo_id, "observaciones": ""}, format="json", ) self.assertEqual(resp_res.status_code, 201) resultado = ResultadoLaboratorio.objects.get(muestreo_id=muestreo_id) p_proteina = Parametro.objects.create( nombre="Proteina", unidad="%", umbral_min=None, umbral_max=Decimal("12"), activo=True, created_by=self.admin, updated_by=self.admin, ) p_humedad = Parametro.objects.create( nombre="Humedad", unidad="%", umbral_min=Decimal("10"), umbral_max=Decimal("14"), activo=True, created_by=self.admin, updated_by=self.admin, ) resp_params = self.client.post( f"/muestreo-laboratorio/resultados/{resultado.id}/parametros/", [ {"parametro": p_proteina.id, "valor": "13"}, {"parametro": p_humedad.id, "valor": "11"}, ], format="json", ) self.assertEqual(resp_params.status_code, 200) resultado.refresh_from_db() self.assertIn("proteina_alta", resultado.tags) self.assertTrue(resultado.alertas.get("fuera_de_rango")) resp_pub = self.client.post( f"/muestreo-laboratorio/resultados/{resultado.id}/publicar/", {}, format="json", ) self.assertEqual(resp_pub.status_code, 200) resultado.refresh_from_db() self.assertEqual(resultado.estado, ResultadoLaboratorio.ESTADO_PUBLICADO) self.assertIsNotNone(resultado.fecha_publicacion) resultado.muestreo.refresh_from_db() self.assertEqual(resultado.muestreo.estado, "completado") recepcion = resultado.muestreo.recepcion recepcion.refresh_from_db() self.assertEqual(recepcion.estado, Recepcion.ESTADO_MUESTREO_COMPLETADO) @override_settings(MUESTREO_LAB_PARAMETROS_REQUERIDOS=["proteina"]) def test_publicar_sin_adjuntar_archivos_y_unidad_desde_catalogo(self): _, muestreo_id = self._crear_recepcion_y_muestreo() self._login("lab1", "lab12345") p_proteina = Parametro.objects.create( nombre="Proteina", unidad="%", umbral_min=None, umbral_max=Decimal("12"), activo=True, created_by=self.admin, updated_by=self.admin, ) resp_params = self.client.post( f"/muestreo-laboratorio/ordenes/{muestreo_id}/parametros/", [{"parametro": p_proteina.id, "valor": "13", "unidad": "kg"}], format="json", ) self.assertEqual(resp_params.status_code, 200) resultado_id = resp_params.data["resultado_id"] resultado = ResultadoLaboratorio.objects.get(pk=resultado_id) pr = resultado.parametros.select_related("parametro").get(parametro_id=p_proteina.id) self.assertEqual(pr.unidad, "%") resp_pub = self.client.post( f"/muestreo-laboratorio/ordenes/{muestreo_id}/publicar/", {}, format="json", ) self.assertEqual(resp_pub.status_code, 200) resultado.refresh_from_db() self.assertEqual(resultado.estado, ResultadoLaboratorio.ESTADO_PUBLICADO) self.assertFalse(bool(resultado.informe)) @override_settings(MUESTREO_LAB_PARAMETROS_REQUERIDOS=["proteina"]) def test_captura_parametros_por_orden_sin_crear_resultado(self): _, muestreo_id = self._crear_recepcion_y_muestreo() self._login("lab1", "lab12345") p_proteina = Parametro.objects.create( nombre="Proteina", unidad="%", umbral_min=None, umbral_max=Decimal("12"), activo=True, created_by=self.admin, updated_by=self.admin, ) resp_params = self.client.post( f"/muestreo-laboratorio/ordenes/{muestreo_id}/parametros/", [{"parametro": p_proteina.id, "valor": "13"}], format="json", ) self.assertEqual(resp_params.status_code, 200) resultado_id = resp_params.data.get("resultado_id") self.assertIsNotNone(resultado_id) resultado = ResultadoLaboratorio.objects.get(pk=resultado_id) self.assertEqual(resultado.muestreo_id, muestreo_id) resp_pub = self.client.post( f"/muestreo-laboratorio/ordenes/{muestreo_id}/publicar/", {}, format="json", ) self.assertEqual(resp_pub.status_code, 200) resultado.refresh_from_db() self.assertEqual(resultado.estado, ResultadoLaboratorio.ESTADO_PUBLICADO) resultado.muestreo.refresh_from_db() self.assertEqual(resultado.muestreo.estado, "completado") def test_adjuntar_informe_y_descarga_protegida(self): _, muestreo_id = self._crear_recepcion_y_muestreo() self._login("lab1", "lab12345") self.client.post( "/muestreo-laboratorio/resultados/", {"muestreo_id": muestreo_id, "observaciones": ""}, format="json", ) resultado = ResultadoLaboratorio.objects.get(muestreo_id=muestreo_id) pdf = SimpleUploadedFile("informe.pdf", b"%PDF-1.4\n%EOF", content_type="application/pdf") resp_adj = self.client.post( f"/muestreo-laboratorio/resultados/{resultado.id}/adjuntar-informe/", {"informe": pdf}, format="multipart", ) self.assertEqual(resp_adj.status_code, 200) client2 = APIClient() resp_no_auth = client2.get(f"/muestreo-laboratorio/resultados/{resultado.id}/informe/download/") self.assertEqual(resp_no_auth.status_code, 401) resp_dl = self.client.get(f"/muestreo-laboratorio/resultados/{resultado.id}/informe/download/") self.assertEqual(resp_dl.status_code, 200) @override_settings(MEDIA_ROOT=TEMP_MEDIA_ROOT) class CalidadGroupAccessTests(APITestCase): """El rol 'Calidad' debe tener el mismo acceso al módulo de Laboratorio que 'Laboratoristas'.""" def setUp(self): self.grp_calidad = Group.objects.create(name=CALIDAD_GROUP_NAME) self.calidad_user = User.objects.create_user( username="calidad_ml", email="calidad_ml@example.com", password="calidad12345", ) self.calidad_user.groups.add(self.grp_calidad) self.oficina = Oficina.objects.create(codigo="OF-CAL-ML", nombre="Oficina Calidad ML", activa=True) self.calidad_user.profile.oficinas.add(self.oficina) def _login(self, username, password): resp = self.client.post( "/auth/token/", {"username": username, "password": password}, 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 test_calidad_puede_listar_ordenes_y_resultados(self): self._login("calidad_ml", "calidad12345") resp_ordenes = self.client.get("/muestreo-laboratorio/ordenes/") self.assertEqual(resp_ordenes.status_code, 200) resp_resultados = self.client.get("/muestreo-laboratorio/resultados/") self.assertEqual(resp_resultados.status_code, 200)