99 lines
5.9 KiB
Python
99 lines
5.9 KiB
Python
"""USB station: guarded intake updates and idempotent inspection snapshots."""
|
|
import hashlib
|
|
import json
|
|
from django.db import transaction
|
|
from django.shortcuts import get_object_or_404
|
|
from rest_framework import serializers
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
from .api import SecureAPIResponseMixin, StrictSerializer, DeviceSerializer, apply_changes
|
|
from .api_identity import canonical_hwid
|
|
from .models import Device, DeviceIdentity, ChoiceValue, InspectionSubmission, InspectionRecord, AuditLog
|
|
from .services import add_audit
|
|
from .workflow import advance_device
|
|
|
|
class IdentityRequest(StrictSerializer):
|
|
hwid = serializers.CharField(max_length=100)
|
|
expected_revision = serializers.IntegerField(min_value=1)
|
|
def validate_hwid(self, value): return canonical_hwid(value)
|
|
|
|
class SpecsSerializer(StrictSerializer):
|
|
model = serializers.CharField(max_length=200, required=False)
|
|
cpu = serializers.CharField(max_length=200, required=False)
|
|
ram_gb = serializers.IntegerField(min_value=1, max_value=2048, required=False)
|
|
ssd_gb = serializers.IntegerField(min_value=1, required=False)
|
|
hdd_gb = serializers.IntegerField(min_value=1, required=False)
|
|
|
|
class IntakeRequest(IdentityRequest):
|
|
specifications = SpecsSerializer()
|
|
|
|
class ResultSerializer(StrictSerializer):
|
|
outcome = serializers.ChoiceField(choices=['NOT_TESTED','NOT_APPLICABLE','PASS','HARDWARE_FAILURE','UNAVAILABLE','SETUP_PROBLEM','NEEDS_REVIEW'])
|
|
notes = serializers.CharField(max_length=10000, allow_blank=True, required=False, default='')
|
|
evidence = serializers.CharField(max_length=100000, allow_blank=True, required=False, default='')
|
|
history = serializers.ListField(child=serializers.JSONField(), max_length=500, required=False, default=list)
|
|
|
|
class InspectionRequest(IdentityRequest):
|
|
request_id = serializers.UUIDField()
|
|
phase = serializers.ChoiceField(choices=['STARTED','IN_PROGRESS','COMPLETE'])
|
|
technician = serializers.CharField(max_length=200, allow_blank=True)
|
|
notes = serializers.CharField(max_length=10000, allow_blank=True)
|
|
results = serializers.DictField(child=ResultSerializer())
|
|
def validate_results(self, value):
|
|
if len(value)>50 or len(json.dumps(value))>500000:
|
|
raise serializers.ValidationError('Too many inspection results.')
|
|
if any(len(k)>80 for k in value): raise serializers.ValidationError('Invalid test name.')
|
|
return value
|
|
def validate(self, attrs):
|
|
if attrs['phase']=='COMPLETE' and (not attrs['technician'].strip() or not attrs['results'] or any(r['outcome'] in ['NOT_TESTED','NEEDS_REVIEW'] for r in attrs['results'].values())):
|
|
raise serializers.ValidationError('Complete all tests and enter the technician name first.')
|
|
return attrs
|
|
|
|
def locked_device(number, hwid):
|
|
d=get_object_or_404(Device.objects.select_for_update(), number=number, archived_at__isnull=True)
|
|
if not DeviceIdentity.objects.filter(device=d,hwid=hwid).exists():
|
|
raise serializers.ValidationError('Hardware identity does not match this device.')
|
|
return d
|
|
|
|
class IntakeSyncView(SecureAPIResponseMixin, APIView):
|
|
@transaction.atomic
|
|
def post(self, request, number):
|
|
s=IntakeRequest(data=request.data);s.is_valid(raise_exception=True);v=s.validated_data
|
|
d=locked_device(number,v['hwid'])
|
|
if d.stage != Device.Stage.INTAKE or d.revision != v['expected_revision']:
|
|
return Response({'detail':'Device state changed. Refresh before intake sync.'},status=409)
|
|
changes={k:x for k,x in v['specifications'].items() if getattr(d,k)!=x}
|
|
if changes:
|
|
d=apply_changes(d,changes,user=request.user,api_client=request.auth.name,expected_revision=d.revision)
|
|
return Response(DeviceSerializer(d).data)
|
|
|
|
class InspectionSyncView(SecureAPIResponseMixin, APIView):
|
|
@transaction.atomic
|
|
def post(self, request, number):
|
|
s=InspectionRequest(data=request.data);s.is_valid(raise_exception=True);v=s.validated_data
|
|
d=locked_device(number,v['hwid'])
|
|
payload={k:x for k,x in request.data.items() if k not in ['expected_revision','request_id']}
|
|
digest=hashlib.sha256(json.dumps(payload,sort_keys=True,separators=(',',':')).encode()).hexdigest()
|
|
previous=InspectionSubmission.objects.filter(request_id=v['request_id']).first()
|
|
if previous:
|
|
if previous.device_id != d.pk or previous.payload_hash != digest:
|
|
return Response({'detail':'Request ID already used for different results.'},status=409)
|
|
return Response(DeviceSerializer(d).data)
|
|
if d.revision != v['expected_revision'] or d.stage not in [Device.Stage.INTAKE,Device.Stage.INSPECTION,Device.Stage.REPAIR]:
|
|
return Response({'detail':'Device state changed. Refresh before inspection sync.'},status=409)
|
|
if d.stage != Device.Stage.INSPECTION:
|
|
d=advance_device(d.pk,Device.Stage.INSPECTION,user=request.user,note='USB inspection',expected_revision=d.revision)
|
|
state=ChoiceValue.objects.get(kind='status',label='Inspektion',active=True)
|
|
if d.status_id != state.pk:
|
|
old=d.status;d.status=state;d.save(update_fields=['status'])
|
|
add_audit(request.user,d,AuditLog.Action.UPDATE,'Status',old,state,api_client=request.auth.name)
|
|
else:
|
|
d.save(update_fields=['updated_at'])
|
|
InspectionSubmission.objects.create(request_id=v['request_id'],device=d,payload_hash=digest,payload=payload,recorded_by=request.user)
|
|
lines=[f"USB {v['phase']} · {v['technician']}",v['notes']]
|
|
lines += [f"{key}: {result['outcome']} — {result['notes']}" for key,result in v['results'].items()]
|
|
summary='\n'.join(lines)
|
|
InspectionRecord.objects.create(device=d,defects=summary,recorded_by=request.user)
|
|
add_audit(request.user,d,AuditLog.Action.UPDATE,'USB inspection','',summary,api_client=request.auth.name)
|
|
return Response(DeviceSerializer(d).data,status=201)
|