Implement backend for versions

This commit is contained in:
IRBorisov 2024-03-04 15:26:23 +03:00
parent b2ada0d630
commit ce945711e2
5 changed files with 165 additions and 13 deletions

View File

@ -130,11 +130,20 @@ class ExpressionParseSerializer(serializers.Serializer):
class VersionSerializer(serializers.ModelSerializer):
''' Serializer: Version data. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'id', 'version', 'item', 'description', 'time_create'
read_only_fields = ('id', 'item', 'time_create')
class VersionInnerSerializer(serializers.ModelSerializer):
''' Serializer: Version data for list of versions. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'id', 'version', 'description', 'time_create'
read_only_fields = ('item', 'id', 'time_create')
read_only_fields = ('id', 'item', 'time_create')
class VersionCreateSerializer(serializers.ModelSerializer):
@ -160,7 +169,7 @@ class LibraryItemDetailsSerializer(serializers.ModelSerializer):
return [item.pk for item in instance.subscribers()]
def get_versions(self, instance: LibraryItem) -> list:
return [VersionSerializer(item).data for item in instance.versions()]
return [VersionInnerSerializer(item).data for item in instance.versions()]
class ConstituentaSerializer(serializers.ModelSerializer):
@ -426,10 +435,10 @@ class RSFormSerializer(serializers.ModelSerializer):
for cst in schema.constituents().order_by('order'):
result['items'].append(ConstituentaSerializer(cst).data)
return result
def to_versioned_data(self) -> dict:
''' Create serializable version representation without redundant data. '''
result = self.to_representation(self.instance)
result = self.to_representation(cast(LibraryItem, self.instance))
del result['versions']
del result['subscribers']
@ -442,7 +451,7 @@ class RSFormSerializer(serializers.ModelSerializer):
def from_versioned_data(self, version: int, data: dict) -> dict:
''' Load data from version. '''
result = self.to_representation(self.instance)
result = self.to_representation(cast(LibraryItem, self.instance))
result['version'] = version
return result | data
@ -651,3 +660,8 @@ class NewCstResponse(serializers.Serializer):
''' Serializer: Create cst response. '''
new_cst = ConstituentaSerializer()
schema = RSFormParseSerializer()
class NewVersionResponse(serializers.Serializer):
''' Serializer: Create cst response. '''
version = serializers.IntegerField()
schema = RSFormParseSerializer()

View File

@ -828,6 +828,86 @@ class TestVersionViews(APITestCase):
self.assertTrue(response.data['version'] in [v['id'] for v in response.data['schema']['versions']])
def test_retrieve_version(self):
data = {'version': '1.0.0', 'description': 'test'}
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
version_id = response.data['version']
invalid_id = 1338
response = self.client.get(f'/api/rsforms/{invalid_id}/versions/{invalid_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
response = self.client.get(f'/api/rsforms/{self.owned.item.id}/versions/{invalid_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
response = self.client.get(f'/api/rsforms/{invalid_id}/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
response = self.client.get(f'/api/rsforms/{self.unowned.item.id}/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.owned.item.alias = 'NewName'
self.owned.item.save()
self.x1.alias = 'X33'
self.x1.save()
response = self.client.get(f'/api/rsforms/{self.owned.item.id}/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertNotEqual(response.data['alias'], self.owned.item.alias)
self.assertNotEqual(response.data['items'][0]['alias'], self.x1.alias)
self.assertEqual(response.data['version'], version_id)
def test_access_version(self):
data = {'version': '1.0.0', 'description': 'test'}
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
version_id = response.data['version']
invalid_id = version_id + 1337
response = self.client.get(f'/api/versions/{invalid_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.client.logout()
response = self.client.get(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['version'], data['version'])
self.assertEqual(response.data['description'], data['description'])
self.assertEqual(response.data['item'], self.owned.item.id)
response = self.client.patch(
f'/api/versions/{version_id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
response = self.client.delete(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.client.force_authenticate(user=self.user)
data = {'version': '1.1.0', 'description': 'test1'}
response = self.client.patch(
f'/api/versions/{version_id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.get(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['version'], data['version'])
self.assertEqual(response.data['description'], data['description'])
response = self.client.delete(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
response = self.client.get(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
class TestRSLanguageViews(APITestCase):
def setUp(self):
self.factory = APIRequestFactory()

View File

@ -9,11 +9,14 @@ library_router.register('rsforms', views.RSFormViewSet)
urlpatterns = [
path('library/active', views.LibraryActiveView.as_view(), name='library'),
path('library/templates', views.LibraryTemplatesView.as_view(), name='library'),
path('library/templates', views.LibraryTemplatesView.as_view(), name='templates'),
path('constituents/<int:pk>', views.ConstituentAPIView.as_view(), name='constituenta-detail'),
path('rsforms/import-trs', views.TrsImportView.as_view()),
path('rsforms/create-detailed', views.create_rsform),
path('versions/<int:pk>', views.VersionAPIView.as_view()),
path('rsforms/<int:pk_item>/versions/create', views.create_version),
path('rsforms/<int:pk_item>/versions/<int:pk_version>', views.retrieve_version),
path('rslang/parse-expression', views.parse_expression),
path('rslang/to-ascii', views.convert_to_ascii),

View File

@ -34,6 +34,15 @@ class SchemaOwnerOrAdmin(BasePermission):
return request.user.is_staff # type: ignore
class ItemOwnerOrAdmin(BasePermission):
''' Permission for object ownership restriction '''
def has_object_permission(self, request, view, obj):
if request.user == obj.item.owner:
return True
if not hasattr(request.user, 'is_staff'):
return False
return request.user.is_staff # type: ignore
def read_trs(file) -> dict:
''' Read JSON from TRS file '''
with ZipFile(file, 'r') as archive:
@ -51,7 +60,7 @@ def write_trs(json_data: dict) -> bytes:
return content.getvalue()
def apply_pattern(text: str, mapping: dict[str, str], pattern: re.Pattern[str]) -> str:
''' Apply mapping to matching in regular expression patter subgroup 1. '''
''' Apply mapping to matching in regular expression patter subgroup 1 '''
if text == '' or pattern == '':
return text
pos_input: int = 0

View File

@ -58,13 +58,29 @@ class ConstituentAPIView(generics.RetrieveUpdateAPIView):
def get_permissions(self):
result = super().get_permissions()
if self.request.method.lower() == 'get':
if self.request.method.upper() == 'GET':
result.append(permissions.AllowAny())
else:
result.append(utils.SchemaOwnerOrAdmin())
return result
@extend_schema(tags=['Version'])
@extend_schema_view()
class VersionAPIView(generics.RetrieveUpdateDestroyAPIView):
''' Endpoint: Get / Update Constituenta. '''
queryset = m.Version.objects.all()
serializer_class = s.VersionSerializer
def get_permissions(self):
result = super().get_permissions()
if self.request.method.upper() == 'GET':
result.append(permissions.AllowAny())
else:
result.append(utils.ItemOwnerOrAdmin())
return result
# pylint: disable=too-many-ancestors
@extend_schema(tags=['Library'])
@extend_schema_view()
@ -196,10 +212,10 @@ class RSFormViewSet(viewsets.GenericViewSet, generics.ListAPIView, generics.Retr
''' Determine permission class. '''
if self.action in ['load_trs', 'cst_create', 'cst_delete_multiple',
'reset_aliases', 'cst_rename', 'cst_substitute']:
permission_classes = [utils.ObjectOwnerOrAdmin]
permission_list = [utils.ObjectOwnerOrAdmin]
else:
permission_classes = [permissions.AllowAny]
return [permission() for permission in permission_classes]
permission_list = [permissions.AllowAny]
return [permission() for permission in permission_list]
@extend_schema(
summary='create constituenta',
@ -547,10 +563,10 @@ def _prepare_rsform_data(data: dict, request: Request, owner: Union[m.User, None
@extend_schema(
summary='save version for RSForm copying current content',
tags=['Versions'],
tags=['Version'],
request=s.VersionCreateSerializer,
responses={
c.HTTP_201_CREATED: s.RSFormParseSerializer,
c.HTTP_201_CREATED: s.NewVersionResponse,
c.HTTP_403_FORBIDDEN: None,
c.HTTP_404_NOT_FOUND: None
}
@ -584,6 +600,36 @@ def create_version(request: Request, pk_item: int):
)
@extend_schema(
summary='retrieve versioned data for RSForm',
tags=['Version'],
request=None,
responses={
c.HTTP_200_OK: s.RSFormParseSerializer,
c.HTTP_404_NOT_FOUND: None
}
)
@api_view(['GET'])
def retrieve_version(request: Request, pk_item: int, pk_version: int):
''' Endpoint: Retrieve version for RSForm. '''
try:
item = m.LibraryItem.objects.get(pk=pk_item)
except m.LibraryItem.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
try:
version = m.Version.objects.get(pk=pk_version)
except m.Version.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
if version.item != item:
return Response(status=c.HTTP_404_NOT_FOUND)
data = s.RSFormSerializer(item).from_versioned_data(version.pk, version.data)
return Response(
status=c.HTTP_200_OK,
data=data
)
@extend_schema(
summary='RS expression into Syntax Tree',
tags=['FormalLanguage'],