T: add basic tests for associations

rsconcept/backend/apps/oss/models/PropagationEngine.py

@@ -140,16 +140,26 @@ class PropagationEngine:
     def inherit_association(self, target: int, items: list[Association]) -> None:
         ''' Execute inheritance of Associations. '''
         operation = self.cache.operation_by_id[target]
-        if operation.result is None:
+        if operation.result is None or not items:
             return
 
         self.cache.ensure_loaded_subs()
+
+        existing_associations = set(
+            Association.objects.filter(
+                container__schema_id=operation.result_id,
+            ).values_list('container_id', 'associate_id')
+        )
+
         new_associations: list[Association] = []
         for assoc in items:
             new_container = self.cache.get_inheritor(assoc.container_id, target)
             new_associate = self.cache.get_inheritor(assoc.associate_id, target)
-            if new_container is None or new_associate is None:
+            if new_container is None or new_associate is None \
+                    or new_associate == new_container \
+                    or (new_container, new_associate) in existing_associations:
                 continue
+
             new_associations.append(Association(
                 container_id=new_container,
                 associate_id=new_associate

rsconcept/backend/apps/rsform/serializers/__init__.py

@@ -12,6 +12,7 @@ from .basics import (
     WordFormSerializer
 )
 from .data_access import (
+    AssociationCreateSerializer,
     AssociationDataSerializer,
     CrucialUpdateSerializer,
     CstCreateSerializer,

rsconcept/backend/apps/rsform/serializers/data_access.py

@@ -49,6 +49,23 @@ class AssociationDataSerializer(StrictSerializer):
         return attrs
 
 
+class AssociationCreateSerializer(AssociationDataSerializer):
+    ''' Serializer: Data for creating new association. '''
+
+    def validate(self, attrs):
+        attrs = super().validate(attrs)
+        if attrs['container'].pk == attrs['associate'].pk:
+            raise serializers.ValidationError({
+                'container': msg.associationSelf()
+            })
+        if Association.objects.filter(container=attrs['container'], associate=attrs['associate']).exists():
+            raise serializers.ValidationError({
+                'associate': msg.associationAlreadyExists()
+            })
+
+        return attrs
+
+
 class CstBaseSerializer(StrictModelSerializer):
     ''' Serializer: Constituenta all data. '''
     class Meta:

rsconcept/backend/apps/rsform/tests/s_views/__init__.py

@@ -1,4 +1,5 @@
 ''' Tests for REST API. '''
+from .t_associations import *
 from .t_cctext import *
 from .t_constituenta import *
 from .t_rsforms import *

rsconcept/backend/apps/rsform/tests/s_views/t_associations.py

@@ -0,0 +1,100 @@
+''' Testing API: Association. '''
+import io
+import os
+from zipfile import ZipFile
+
+from cctext import ReferenceType
+from rest_framework import status
+
+from apps.library.models import AccessPolicy, LibraryItem, LibraryItemType, LocationHead
+from apps.rsform.models import Association, Constituenta, CstType, RSForm
+from shared.EndpointTester import EndpointTester, decl_endpoint
+from shared.testing_utils import response_contains
+
+
+class TestAssociationsEndpoints(EndpointTester):
+    ''' Testing basic Association API. '''
+
+    def setUp(self):
+        super().setUp()
+        self.owned = RSForm.create(title='Test', alias='T1', owner=self.user)
+        self.owned_id = self.owned.model.pk
+        self.unowned = RSForm.create(title='Test2', alias='T2')
+        self.unowned_id = self.unowned.model.pk
+        self.n1 = self.owned.insert_last('N1')
+        self.x1 = self.owned.insert_last('X1')
+        self.n2 = self.owned.insert_last('N2')
+        self.unowned_cst = self.unowned.insert_last('C1')
+        self.invalid_id = self.n2.pk + 1337
+
+
+    @decl_endpoint('/api/rsforms/{item}/create-association', method='post')
+    def test_create_association(self):
+        self.executeBadData({}, item=self.owned_id)
+
+        data = {'container': self.n1.pk, 'associate': self.invalid_id}
+        self.executeBadData(data, item=self.owned_id)
+
+        data['associate'] = self.unowned_cst.pk
+        self.executeBadData(data, item=self.owned_id)
+
+        data['associate'] = data['container']
+        self.executeBadData(data, item=self.owned_id)
+
+        data = {'container': self.n1.pk, 'associate': self.x1.pk}
+        self.executeBadData(data, item=self.unowned_id)
+
+        response = self.executeCreated(data, item=self.owned_id)
+        associations = response.data['association']
+        self.assertEqual(len(associations), 1)
+        self.assertEqual(associations[0]['container'], self.n1.pk)
+        self.assertEqual(associations[0]['associate'], self.x1.pk)
+
+
+    @decl_endpoint('/api/rsforms/{item}/create-association', method='post')
+    def test_create_association_duplicate(self):
+        data = {'container': self.n1.pk, 'associate': self.x1.pk}
+        self.executeCreated(data, item=self.owned_id)
+        self.executeBadData(data, item=self.owned_id)
+
+
+    @decl_endpoint('/api/rsforms/{item}/delete-association', method='patch')
+    def test_delete_association(self):
+        data = {'container': self.n1.pk, 'associate': self.x1.pk}
+        self.executeForbidden(data, item=self.unowned_id)
+        self.executeBadData(data, item=self.owned_id)
+
+        Association.objects.create(
+            container=self.n1,
+            associate=self.x1
+        )
+        self.executeForbidden(data, item=self.unowned_id)
+        response = self.executeOK(data, item=self.owned_id)
+        associations = response.data['association']
+        self.assertEqual(len(associations), 0)
+
+
+    @decl_endpoint('/api/rsforms/{item}/clear-associations', method='patch')
+    def test_clear_associations(self):
+        data = {'target': self.n1.pk}
+        self.executeForbidden(data, item=self.unowned_id)
+        self.executeNotFound(data, item=self.invalid_id)
+        self.executeOK(data, item=self.owned_id)
+
+        Association.objects.create(
+            container=self.n1,
+            associate=self.x1
+        )
+        Association.objects.create(
+            container=self.n1,
+            associate=self.n2
+        )
+        Association.objects.create(
+            container=self.n2,
+            associate=self.n1
+        )
+        response = self.executeOK(data, item=self.owned_id)
+        associations = response.data['association']
+        self.assertEqual(len(associations), 1)
+        self.assertEqual(associations[0]['container'], self.n2.pk)
+        self.assertEqual(associations[0]['associate'], self.n1.pk)

rsconcept/backend/apps/rsform/views/rsforms.py

@@ -287,7 +287,7 @@ class RSFormViewSet(viewsets.GenericViewSet, generics.ListAPIView, generics.Retr
     @extend_schema(
         summary='create Association',
         tags=['Constituenta'],
-        request=s.AssociationDataSerializer,
+        request=s.AssociationCreateSerializer,
         responses={
             c.HTTP_201_CREATED: s.RSFormParseSerializer,
             c.HTTP_400_BAD_REQUEST: None,
@@ -299,13 +299,15 @@ class RSFormViewSet(viewsets.GenericViewSet, generics.ListAPIView, generics.Retr
     def create_association(self, request: Request, pk) -> HttpResponse:
         ''' Create Association. '''
         item = self._get_item()
-        serializer = s.AssociationDataSerializer(data=request.data, context={'schema': item})
+        serializer = s.AssociationCreateSerializer(data=request.data, context={'schema': item})
         serializer.is_valid(raise_exception=True)
+        container = serializer.validated_data['container']
+        associate = serializer.validated_data['associate']
 
         with transaction.atomic():
             new_association = m.Association.objects.create(
-                container=serializer.validated_data['container'],
+                container=container,
-                associate=serializer.validated_data['associate']
+                associate=associate
             )
             PropagationFacade.after_create_association(item.pk, [new_association])
             item.save(update_fields=['time_update'])

rsconcept/backend/shared/messages.py

@@ -10,6 +10,14 @@ def constituentsInvalid(constituents: list[int]):
     return f'некорректные конституенты для схемы: {constituents}'
 
 
+def associationSelf():
+    return 'Рефлексивная ассоциация не допускается'
+
+
+def associationAlreadyExists():
+    return 'Отношение уже существует'
+
+
 def constituentaNotInRSform(title: str):
     return f'Конституента не принадлежит схеме: {title}'