ixgadmin
8ad83a7a65
Batch writes were closed in 88eede5, but every other app's viewset still used
IsAuthenticated, which in this platform means any buyer. A buyer could rewrite a
document record, create a QC result (which moves the batch status and raises a
CAPA), and reach the write endpoints for CAPAs, certificates, ESG reports and
shipments.
- Use the one IsOpsOrAdmin from apps.compliance.permissions; drop the duplicate
copy in apps/qc/views.py.
- Apply it to the whole viewset on shipments, qc, capa, esg, documents,
certificates-of-analysis and regulatory-certs. The buyer portal is
server-rendered through the ORM, so it needs none of these APIs.
- Keep the buyer chat working: AI interactions stay authenticated and scoped,
and agent reads stay open while agent writes become ops/admin.
- Close a buyer-to-admin escalation in the users API: writes are admin-only and
role/is_active are read-only, so a buyer can no longer PATCH their own row to
role=admin and then pass every IsOpsOrAdmin check.
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
from django.utils import timezone
|
|
from rest_framework import viewsets, status
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
|
|
from apps.compliance.permissions import IsOpsOrAdmin
|
|
from apps.compliance.services import create_coa_for_qc
|
|
|
|
from .models import QCResult
|
|
from .serializers import QCResultSerializer, QCResultCreateSerializer
|
|
|
|
|
|
class QCResultViewSet(viewsets.ModelViewSet):
|
|
queryset = QCResult.objects.all()
|
|
# Recording a QC result moves the batch's status, quarantines a failure and
|
|
# can raise a CAPA, so the whole viewset — read and write — is ops/admin
|
|
# only. The buyer portal reads QC results through the ORM
|
|
# (apps/buyer/views.py), not this API.
|
|
permission_classes = [IsOpsOrAdmin]
|
|
|
|
def get_serializer_class(self):
|
|
if self.action == "create":
|
|
return QCResultCreateSerializer
|
|
return QCResultSerializer
|
|
|
|
def perform_create(self, serializer):
|
|
serializer.save()
|
|
|
|
@action(detail=True, methods=["post"])
|
|
def verify(self, request, pk=None):
|
|
"""Four-eyes quality approval. A separate authorised user validates the
|
|
entered result. The submitter cannot approve their own result."""
|
|
qc = self.get_object()
|
|
user = request.user
|
|
|
|
if qc.verification_status == QCResult.VerificationStatus.APPROVED:
|
|
return Response({"detail": "QC result already approved"}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
if qc.entered_by is not None and qc.entered_by_id == user.id:
|
|
return Response(
|
|
{"detail": "The submitter cannot approve their own QC result (four-eyes control)"},
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
approved = request.data.get("approved", "true")
|
|
approved = str(approved).lower() in ("true", "1", "yes")
|
|
|
|
if approved:
|
|
if not qc.overall_pass:
|
|
return Response(
|
|
{"detail": "A failed QC result cannot be approved; it remains quarantined with an open CAPA"},
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
qc.verification_status = QCResult.VerificationStatus.APPROVED
|
|
qc.verified_by = user
|
|
qc.approved_at = timezone.now()
|
|
qc.save()
|
|
coa = create_coa_for_qc(qc, approved_by=user)
|
|
return Response(
|
|
{
|
|
"detail": "QC result approved",
|
|
"coa_id": str(coa.id),
|
|
"coa_status": coa.status,
|
|
"coa_url": coa.file_url,
|
|
}
|
|
)
|
|
|
|
qc.verification_status = QCResult.VerificationStatus.REJECTED
|
|
qc.verified_by = user
|
|
qc.approved_at = timezone.now()
|
|
qc.save()
|
|
return Response({"detail": "QC result rejected"})
|