Add initial implementations for forms and organization apps with serializers, factories, and admin configurations
Some checks failed
CI/CD Pipeline / Run Tests (push) Failing after 45s
CI/CD Pipeline / Code Quality Checks (push) Failing after 48s
CI/CD Pipeline / Build Docker Images (push) Has been skipped
CI/CD Pipeline / Push to Gitea Registry (push) Has been skipped
CI/CD Pipeline / Deploy to Server (push) Has been skipped

This commit is contained in:
2026-03-28 18:23:06 +01:00
parent 8ed3e1175c
commit 345b1d0cc8
201 changed files with 15097 additions and 6691 deletions

View File

@@ -11,7 +11,7 @@ from apps.core.viewsets import (
OwnerViewSet,
ReadOnlyViewSet,
)
from apps.parsers.models import Proxy
from apps.organization.models import Organization
from apps.user.models import Profile, User
from django.test import TestCase, override_settings
from django.urls import include, path
@@ -21,30 +21,31 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.routers import DefaultRouter
from rest_framework.test import APITestCase
from tests.apps.parsers.factories import ProxyFactory, fake
from tests.apps.organization.factories import OrganizationFactory, fake
from tests.apps.user.factories import ProfileFactory, UserFactory
def _proxy_payload() -> dict[str, Any]:
proxy = ProxyFactory.build()
def _organization_payload() -> dict[str, Any]:
organization = OrganizationFactory.build()
return {
"address": proxy.address,
"is_active": proxy.is_active,
"fail_count": proxy.fail_count,
"description": proxy.description,
"name": organization.name,
"inn": fake.unique.numerify("##########"),
"ogrn": organization.ogrn,
"kpp": organization.kpp,
"okpo": organization.okpo,
}
class ProxySerializer(serializers.ModelSerializer):
class OrganizationSerializer(serializers.ModelSerializer):
class Meta:
model = Proxy
fields = ["id", "address", "is_active", "fail_count", "description"]
model = Organization
fields = ["id", "name", "inn", "ogrn", "kpp", "okpo"]
class ProxyListSerializer(serializers.ModelSerializer):
class OrganizationListSerializer(serializers.ModelSerializer):
class Meta:
model = Proxy
fields = ["id", "address"]
model = Organization
fields = ["id", "name"]
class ProfileSerializer(serializers.ModelSerializer):
@@ -60,27 +61,27 @@ class UserSerializer(serializers.ModelSerializer):
fields = ["id", "email", "username"]
class ProxyViewSet(BaseViewSet[Proxy]):
queryset = Proxy.objects.all()
serializer_class = ProxySerializer
serializer_classes = {"list": ProxyListSerializer}
only_fields = ["id", "address"]
class OrganizationViewSet(BaseViewSet[Organization]):
queryset = Organization.objects.all()
serializer_class = OrganizationSerializer
serializer_classes = {"list": OrganizationListSerializer}
only_fields = ["id", "name"]
class DeferProxyViewSet(BaseViewSet[Proxy]):
queryset = Proxy.objects.all()
serializer_class = ProxySerializer
defer_fields = ["description"]
class DeferOrganizationViewSet(BaseViewSet[Organization]):
queryset = Organization.objects.all()
serializer_class = OrganizationSerializer
defer_fields = ["okpo"]
class ReadOnlyProxyViewSet(ReadOnlyViewSet[Proxy]):
queryset = Proxy.objects.all()
serializer_class = ProxySerializer
class ReadOnlyOrganizationViewSet(ReadOnlyViewSet[Organization]):
queryset = Organization.objects.all()
serializer_class = OrganizationSerializer
class NoPaginationProxyViewSet(BaseViewSet[Proxy]):
queryset = Proxy.objects.all()
serializer_class = ProxySerializer
class NoPaginationOrganizationViewSet(BaseViewSet[Organization]):
queryset = Organization.objects.all()
serializer_class = OrganizationSerializer
pagination_class = None
@@ -113,9 +114,9 @@ class OwnerProfileViewSet(OwnerViewSet[Profile]):
serializer_class = ProfileSerializer
class BulkProxyViewSet(BulkMixin, BaseViewSet[Proxy]):
queryset = Proxy.objects.all()
serializer_class = ProxySerializer
class BulkOrganizationViewSet(BulkMixin, BaseViewSet[Organization]):
queryset = Organization.objects.all()
serializer_class = OrganizationSerializer
bulk_max_items = 2
@action(detail=False, methods=["post"])
@@ -132,16 +133,32 @@ class BulkProxyViewSet(BulkMixin, BaseViewSet[Proxy]):
router = DefaultRouter()
router.register("proxies", ProxyViewSet, basename="proxy")
router.register("proxies-defer", DeferProxyViewSet, basename="proxy-defer")
router.register("proxies-readonly", ReadOnlyProxyViewSet, basename="proxy-readonly")
router.register("proxies-nopage", NoPaginationProxyViewSet, basename="proxy-nopage")
router.register("organizations", OrganizationViewSet, basename="organization")
router.register(
"organizations-defer",
DeferOrganizationViewSet,
basename="organization-defer",
)
router.register(
"organizations-readonly",
ReadOnlyOrganizationViewSet,
basename="organization-readonly",
)
router.register(
"organizations-nopage",
NoPaginationOrganizationViewSet,
basename="organization-nopage",
)
router.register("profiles-select", ProfileSelectViewSet, basename="profile-select")
router.register("profiles-old", ProfileOldStyleViewSet, basename="profile-old")
router.register("users-prefetch", UserPrefetchViewSet, basename="user-prefetch")
router.register("users-old", UserOldStyleViewSet, basename="user-old")
router.register("profiles-owner", OwnerProfileViewSet, basename="profile-owner")
router.register("bulk-proxies", BulkProxyViewSet, basename="bulk-proxy")
router.register(
"bulk-organizations",
BulkOrganizationViewSet,
basename="bulk-organization",
)
urlpatterns = [path("", include(router.urls))]
@@ -227,22 +244,20 @@ class BaseViewSetIntegrationTest(APITestCase):
self.client.force_authenticate(self.user)
def test_list_paginated_uses_list_serializer(self):
ProxyFactory.create_batch(3)
OrganizationFactory.create_batch(3)
response = self.client.get("/proxies/?page=1&page_size=2")
response = self.client.get("/organizations/?page=1&page_size=2")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response.data["success"])
self.assertEqual(len(response.data["data"]), 2)
self.assertIn("pagination", response.data["meta"])
self.assertSetEqual(
set(response.data["data"][0].keys()), {"id", "address"}
)
self.assertSetEqual(set(response.data["data"][0].keys()), {"id", "name"})
def test_list_without_pagination(self):
ProxyFactory.create_batch(2)
OrganizationFactory.create_batch(2)
response = self.client.get("/proxies-nopage/")
response = self.client.get("/organizations-nopage/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response.data["success"])
@@ -250,30 +265,30 @@ class BaseViewSetIntegrationTest(APITestCase):
self.assertIsNone(response.data["meta"])
def test_retrieve_uses_default_serializer(self):
proxy = ProxyFactory()
organization = OrganizationFactory()
response = self.client.get(f"/proxies/{proxy.pk}/")
response = self.client.get(f"/organizations/{organization.pk}/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("fail_count", response.data["data"])
self.assertIn("inn", response.data["data"])
def test_create_update_delete(self):
payload = _proxy_payload()
payload = _organization_payload()
created = self.client.post("/proxies/", payload, format="json")
created = self.client.post("/organizations/", payload, format="json")
self.assertEqual(created.status_code, status.HTTP_201_CREATED)
proxy_id = created.data["data"]["id"]
new_description = fake.sentence(nb_words=3)
organization_id = created.data["data"]["id"]
new_name = fake.company()
updated = self.client.patch(
f"/proxies/{proxy_id}/",
{"description": new_description},
f"/organizations/{organization_id}/",
{"name": new_name},
format="json",
)
self.assertEqual(updated.status_code, status.HTTP_200_OK)
self.assertEqual(updated.data["data"]["description"], new_description)
self.assertEqual(updated.data["data"]["name"], new_name)
deleted = self.client.delete(f"/proxies/{proxy_id}/")
deleted = self.client.delete(f"/organizations/{organization_id}/")
self.assertEqual(deleted.status_code, status.HTTP_204_NO_CONTENT)
@@ -284,16 +299,16 @@ class ReadOnlyViewSetIntegrationTest(APITestCase):
self.client.force_authenticate(self.user)
def test_readonly_list_and_retrieve(self):
proxy = ProxyFactory()
ProxyFactory.create_batch(2)
organization = OrganizationFactory()
OrganizationFactory.create_batch(2)
list_response = self.client.get("/proxies-readonly/")
list_response = self.client.get("/organizations-readonly/")
self.assertEqual(list_response.status_code, status.HTTP_200_OK)
self.assertTrue(list_response.data["success"])
detail_response = self.client.get(f"/proxies-readonly/{proxy.pk}/")
detail_response = self.client.get(f"/organizations-readonly/{organization.pk}/")
self.assertEqual(detail_response.status_code, status.HTTP_200_OK)
self.assertEqual(detail_response.data["data"]["id"], proxy.pk)
self.assertEqual(detail_response.data["data"]["id"], str(organization.pk))
@override_settings(ROOT_URLCONF=__name__)
@@ -334,50 +349,52 @@ class BulkMixinIntegrationTest(APITestCase):
self.client.force_authenticate(self.user)
def test_bulk_create_empty_items(self):
response = self.client.post("/bulk-proxies/bulk_create/", {}, format="json")
response = self.client.post(
"/bulk-organizations/bulk_create/", {}, format="json"
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertFalse(response.data["success"])
def test_bulk_create_too_many(self):
items = [_proxy_payload() for _ in range(3)]
items = [_organization_payload() for _ in range(3)]
response = self.client.post(
"/bulk-proxies/bulk_create/", {"items": items}, format="json"
"/bulk-organizations/bulk_create/", {"items": items}, format="json"
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["errors"][0]["code"], "too_many_items")
def test_bulk_create_update_delete(self):
items = [_proxy_payload(), _proxy_payload()]
items = [_organization_payload(), _organization_payload()]
created = self.client.post(
"/bulk-proxies/bulk_create/", {"items": items}, format="json"
"/bulk-organizations/bulk_create/", {"items": items}, format="json"
)
self.assertEqual(created.status_code, status.HTTP_201_CREATED)
created_ids = [item["id"] for item in created.data["data"]]
update_items = [
{"id": created_ids[0], "description": fake.sentence(nb_words=2)},
{"id": created_ids[0], "name": fake.company()},
{
"id": fake.random_int(min=999999, max=9999999),
"description": fake.word(),
"id": fake.uuid4(),
"name": fake.company(),
},
]
updated = self.client.patch(
"/bulk-proxies/bulk_update/", {"items": update_items}, format="json"
"/bulk-organizations/bulk_update/", {"items": update_items}, format="json"
)
self.assertEqual(updated.status_code, status.HTTP_200_OK)
self.assertEqual(len(updated.data["data"]["updated"]), 1)
self.assertEqual(len(updated.data["data"]["errors"]), 1)
deleted = self.client.delete(
"/bulk-proxies/bulk_delete/", {"ids": created_ids}, format="json"
"/bulk-organizations/bulk_delete/", {"ids": created_ids}, format="json"
)
self.assertEqual(deleted.status_code, status.HTTP_200_OK)
self.assertEqual(deleted.data["data"]["deleted"], len(created_ids))
def test_bulk_update_missing_ids(self):
response = self.client.patch(
"/bulk-proxies/bulk_update/",
{"items": [{"address": fake.word()}]},
"/bulk-organizations/bulk_update/",
{"items": [{"name": fake.company()}]},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@@ -409,6 +426,6 @@ class QuerysetOptimizationIntegrationTest(APITestCase):
self.assertEqual(response_old.status_code, status.HTTP_200_OK)
def test_defer_fields_branch(self):
ProxyFactory.create_batch(2)
response = self.client.get("/proxies-defer/")
OrganizationFactory.create_batch(2)
response = self.client.get("/organizations-defer/")
self.assertEqual(response.status_code, status.HTTP_200_OK)