Complete worked example
python
from drf_lab import setup
setup()
from django.db import IntegrityError, transaction
from rest_framework import generics, serializers
from rest_framework.authentication import TokenAuthentication
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.test import APIRequestFactory, force_authenticate
from drf_lab import reset_fixture
from drf_lab.models import Role
class Query(serializers.Serializer):
country = serializers.ChoiceField(choices=["UA", "IL", "US", "CA"], default="UA")
class PublicRole(serializers.ModelSerializer):
slots = serializers.IntegerField(min_value=1, max_value=3, default=1)
class Meta:
model = Role
fields = ["id", "title", "country", "slots"]
read_only_fields = ["id"]
class SmallPage(LimitOffsetPagination):
default_limit = 1
max_limit = 2
class Roles(generics.ListCreateAPIView):
serializer_class = PublicRole
permission_classes = [IsAuthenticated]
authentication_classes = [TokenAuthentication]
pagination_class = SmallPage
def get_queryset(self):
query = Query(data=self.request.query_params)
query.is_valid(raise_exception=True)
return Role.objects.filter(owner=self.request.user, country=query.validated_data["country"]).order_by("id")
def perform_create(self, serializer):
serializer.save(owner=self.request.user, company_id=1)
def create(self, request, *args, **kwargs):
try:
with transaction.atomic():
return super().create(request, *args, **kwargs)
except IntegrityError:
return Response({"detail": "Role already exists."}, status=409)
def error_codes(value):
if isinstance(value, dict):
return {key: error_codes(item) for key, item in value.items()}
if isinstance(value, list):
return [error_codes(item) for item in value]
return getattr(value, "code", str(value))
def inspect_example(payload):
alice, bob, _ = reset_fixture()
factory = APIRequestFactory()
method = payload.get("method", "get")
request = factory.post("/roles/", payload.get("data", {}), format="json") if method == "post" else factory.get("/roles/", payload.get("query", {}))
if payload.get("user"):
force_authenticate(request, user=alice if payload["user"] == "alice" else bob)
response = Roles.as_view()(request)
if response.status_code >= 400:
return {"status": response.status_code, "body": error_codes(response.data), "rows": Role.objects.count()}
if method == "post":
row = Role.objects.get(title=response.data["title"])
return {"status": response.status_code, "keys": sorted(response.data), "title": row.title,
"stored_owner": row.owner.username, "rows": Role.objects.count()}
rows = response.data["results"]
return {"status": response.status_code, "count": response.data["count"],
"ids": [row["id"] for row in rows], "keys": sorted(rows[0]) if rows else []}- Reasoning
- Collection selection validates country, filters by request.user and uses a stable order before pagination. The public serializer excludes owner, company authority and private_note. Creation supplies owner from the verified principal and company from trusted fixture context rather than accepting either from an arbitrary body field. The create operation runs inside atomic and translates a uniqueness conflict to a deliberate 409 response after rollback. The test reads the stored owner and row count for accepted writes. This makes an incorrect service-level owner assignment observable even when ordinary input values remain valid.