import json from datetime import date from unittest import mock from unittest.mock import ANY from django.contrib.auth.models import Permission from django.contrib.auth.models import User from django.db import connection from django.test import override_settings from django.test.utils import CaptureQueriesContext from guardian.shortcuts import assign_perm from rest_framework import status from rest_framework.test import APITestCase from documents.data_models import DocumentMetadataOverrides from documents.models import CustomField from documents.models import CustomFieldInstance from documents.models import Document from documents.serialisers import CustomFieldInstanceSerializer from documents.serialisers import DocumentSerializer from documents.tests.factories import DocumentFactory from documents.tests.utils import DirectoriesMixin class TestCustomFieldsAPI(DirectoriesMixin, APITestCase): ENDPOINT = "/api/custom_fields/" def setUp(self) -> None: self.user = User.objects.create_superuser(username="temp_admin") self.client.force_authenticate(user=self.user) return super().setUp() def test_create_custom_field(self) -> None: """ GIVEN: - Each of the supported data types is created WHEN: - API request to create custom metadata is made THEN: - the field is created - the field returns the correct fields """ for field_type, name in [ ("string", "Custom Text"), ("url", "Wikipedia Link"), ("date", "Invoiced Date"), ("integer", "Invoice #"), ("boolean", "Is Active"), ("float", "Average Value"), ("monetary", "Total Paid"), ("documentlink", "Related Documents"), ]: resp = self.client.post( self.ENDPOINT, data=json.dumps( { "data_type": field_type, "name": name, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) data = resp.json() self.assertEqual(data["name"], name) self.assertEqual(data["data_type"], field_type) resp = self.client.post( self.ENDPOINT, json.dumps( { "data_type": "select", "name": "Select Field", "extra_data": { "select_options": [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 2", "id": "def-456"}, ], }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) data = resp.json() self.assertCountEqual( data["extra_data"]["select_options"], [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 2", "id": "def-456"}, ], ) def test_create_custom_field_nonunique_name(self) -> None: """ GIVEN: - Custom field exists WHEN: - API request to create custom field with the same name THEN: - HTTP 400 is returned """ CustomField.objects.create( name="Test Custom Field", data_type=CustomField.FieldDataType.STRING, ) resp = self.client.post( self.ENDPOINT, data={ "data_type": "string", "name": "Test Custom Field", }, ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_create_custom_field_select_invalid_options(self) -> None: """ GIVEN: - Custom field does not exist WHEN: - API request to create custom field with invalid select options THEN: - HTTP 400 is returned """ # Not a list resp = self.client.post( self.ENDPOINT, json.dumps( { "data_type": "select", "name": "Select Field", "extra_data": { "select_options": "not a list", }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) # No options resp = self.client.post( self.ENDPOINT, json.dumps( { "data_type": "select", "name": "Select Field", }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_custom_field_select_unique_ids(self) -> None: """ GIVEN: - Existing custom field WHEN: - API request to create custom field with select options without id THEN: - Unique ids are generated for each option """ resp = self.client.post( self.ENDPOINT, json.dumps( { "data_type": "select", "name": "Select Field", "extra_data": { "select_options": [ {"label": "Option 1"}, {"label": "Option 2"}, ], }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) data = resp.json() self.assertCountEqual( data["extra_data"]["select_options"], [ {"label": "Option 1", "id": ANY}, {"label": "Option 2", "id": ANY}, ], ) # Add a new option resp = self.client.patch( f"{self.ENDPOINT}{data['id']}/", json.dumps( { "extra_data": { "select_options": data["extra_data"]["select_options"] + [{"label": "Option 3"}], }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertCountEqual( data["extra_data"]["select_options"], [ {"label": "Option 1", "id": ANY}, {"label": "Option 2", "id": ANY}, {"label": "Option 3", "id": ANY}, ], ) @override_settings(CELERY_TASK_ALWAYS_EAGER=True) def test_custom_field_select_options_pruned(self) -> None: """ GIVEN: - Select custom field exists and document instance with one of the options WHEN: - API request to remove an option from the select field THEN: - The option is removed from the field - The option is removed from the document instance """ custom_field_select = CustomField.objects.create( name="Select Field", data_type=CustomField.FieldDataType.SELECT, extra_data={ "select_options": [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 2", "id": "def-456"}, {"label": "Option 3", "id": "ghi-789"}, ], }, ) doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) CustomFieldInstance.objects.create( document=doc, field=custom_field_select, value_select="def-456", ) resp = self.client.patch( f"{self.ENDPOINT}{custom_field_select.id}/", data=json.dumps( { "extra_data": { "select_options": [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 3", "id": "ghi-789"}, ], }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertCountEqual( data["extra_data"]["select_options"], [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 3", "id": "ghi-789"}, ], ) doc.refresh_from_db() _cf_1 = doc.custom_fields.first() assert _cf_1 is not None self.assertEqual(_cf_1.value, None) @mock.patch("documents.signals.handlers.process_cf_select_update.apply_async") def test_custom_field_update_offloaded_once(self, mock_delay) -> None: """ GIVEN: - A select custom field attached to multiple documents WHEN: - The select options are updated THEN: - The async update task is enqueued once """ cf_select = CustomField.objects.create( name="Select Field", data_type=CustomField.FieldDataType.SELECT, extra_data={ "select_options": [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 2", "id": "def-456"}, ], }, ) documents = [ Document.objects.create( title="WOW", content="the content", checksum=f"{i}", mime_type="application/pdf", ) for i in range(3) ] for document in documents: CustomFieldInstance.objects.create( document=document, field=cf_select, value_select="def-456", ) cf_select.extra_data = { "select_options": [ {"label": "Option 1", "id": "abc-123"}, ], } cf_select.save() mock_delay.assert_called_once_with(kwargs={"custom_field": cf_select}) def test_create_custom_field_monetary_validation(self) -> None: """ GIVEN: - Custom field does not exist WHEN: - API request to create custom field with invalid default currency option - API request to create custom field with valid default currency option THEN: - HTTP 400 is returned - HTTP 201 is returned """ # not a string resp = self.client.post( self.ENDPOINT, json.dumps( { "data_type": "monetary", "name": "Monetary Field", "extra_data": { "default_currency": 123, }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) # not a 3-letter currency code resp = self.client.post( self.ENDPOINT, json.dumps( { "data_type": "monetary", "name": "Monetary Field", "extra_data": { "default_currency": "EU", }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) # valid currency code resp = self.client.post( self.ENDPOINT, json.dumps( { "data_type": "monetary", "name": "Monetary Field", "extra_data": { "default_currency": "EUR", }, }, ), content_type="application/json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) def test_create_custom_field_instance(self) -> None: """ GIVEN: - Field of each data type is created WHEN: - API request to create custom metadata instance with each data type THEN: - the field instance is created - the field returns the correct fields and values - the field is attached to the given document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) doc2 = Document.objects.create( title="WOW2", content="the content2", checksum="1234", mime_type="application/pdf", ) custom_field_string = CustomField.objects.create( name="Test Custom Field String", data_type=CustomField.FieldDataType.STRING, ) custom_field_date = CustomField.objects.create( name="Test Custom Field Date", data_type=CustomField.FieldDataType.DATE, ) custom_field_int = CustomField.objects.create( name="Test Custom Field Int", data_type=CustomField.FieldDataType.INT, ) custom_field_boolean = CustomField.objects.create( name="Test Custom Field Boolean", data_type=CustomField.FieldDataType.BOOL, ) custom_field_url = CustomField.objects.create( name="Test Custom Field Url", data_type=CustomField.FieldDataType.URL, ) custom_field_float = CustomField.objects.create( name="Test Custom Field Float", data_type=CustomField.FieldDataType.FLOAT, ) custom_field_monetary = CustomField.objects.create( name="Test Custom Field Monetary", data_type=CustomField.FieldDataType.MONETARY, ) custom_field_monetary2 = CustomField.objects.create( name="Test Custom Field Monetary 2", data_type=CustomField.FieldDataType.MONETARY, ) custom_field_documentlink = CustomField.objects.create( name="Test Custom Field Doc Link", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) custom_field_select = CustomField.objects.create( name="Test Custom Field Select", data_type=CustomField.FieldDataType.SELECT, extra_data={ "select_options": [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 2", "id": "def-456"}, ], }, ) date_value = date.today() resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_string.id, "value": "test value", }, { "field": custom_field_date.id, "value": date_value.isoformat(), }, { "field": custom_field_int.id, "value": 3, }, { "field": custom_field_boolean.id, "value": True, }, { "field": custom_field_url.id, "value": "https://example.com", }, { "field": custom_field_float.id, "value": 12.3456, }, { "field": custom_field_monetary.id, "value": "EUR11.10", }, { "field": custom_field_monetary2.id, "value": 11.10, # Legacy format }, { "field": custom_field_documentlink.id, "value": [doc2.id], }, { "field": custom_field_select.id, "value": "abc-123", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) resp_data = resp.json()["custom_fields"] self.assertCountEqual( resp_data, [ {"field": custom_field_string.id, "value": "test value"}, {"field": custom_field_date.id, "value": date_value.isoformat()}, {"field": custom_field_int.id, "value": 3}, {"field": custom_field_boolean.id, "value": True}, {"field": custom_field_url.id, "value": "https://example.com"}, {"field": custom_field_float.id, "value": 12.3456}, {"field": custom_field_monetary.id, "value": "EUR11.10"}, {"field": custom_field_monetary2.id, "value": "11.1"}, {"field": custom_field_documentlink.id, "value": [doc2.id]}, {"field": custom_field_select.id, "value": "abc-123"}, ], ) doc.refresh_from_db() self.assertEqual(len(doc.custom_fields.all()), 10) def test_document_serializer_custom_fields_validation_batches_field_lookup( self, ) -> None: """ GIVEN: - A document is being validated with several custom field values at once (as happens on every PATCH/PUT/POST) WHEN: - The serializer is validated THEN: - The referenced CustomField objects are resolved with a single query, not one query per custom field """ doc = DocumentFactory(mime_type="application/pdf") custom_fields = [ CustomField.objects.create( name=f"Test Custom Field {i}", data_type=CustomField.FieldDataType.STRING, ) for i in range(5) ] serializer = DocumentSerializer( doc, data={ "custom_fields": [ {"field": custom_field.id, "value": "test value"} for custom_field in custom_fields ], }, partial=True, ) with CaptureQueriesContext(connection) as ctx: self.assertTrue(serializer.is_valid(), serializer.errors) custom_field_lookups = [ query for query in ctx.captured_queries if 'FROM "documents_customfield" WHERE "documents_customfield"."id"' in query["sql"] ] self.assertEqual( len(custom_field_lookups), 1, "Expected a single batched query to resolve the custom fields, " f"got {len(custom_field_lookups)}: {custom_field_lookups}", ) def test_custom_field_lookup_reuses_shared_context_cache(self) -> None: """ GIVEN: - A CustomField has already been resolved once, by a serializer sharing a given `context` dict WHEN: - A second, separately-instantiated CustomFieldInstanceSerializer validates the same field id, sharing that same context (this is what drf-writable-nested does: it rebuilds a fresh serializer -- and fresh field instances -- per item while matching existing vs. new instances during save()) THEN: - No additional query is issued to resolve the CustomField """ custom_field = CustomField.objects.create( name="Test Custom Field", data_type=CustomField.FieldDataType.STRING, ) context: dict = {} first_pass = CustomFieldInstanceSerializer( data={"field": custom_field.id, "value": "a"}, context=context, ) self.assertTrue(first_pass.is_valid(), first_pass.errors) second_pass = CustomFieldInstanceSerializer( data={"field": custom_field.id, "value": "b"}, context=context, ) with CaptureQueriesContext(connection) as ctx: self.assertTrue(second_pass.is_valid(), second_pass.errors) custom_field_lookups = [ query for query in ctx.captured_queries if 'FROM "documents_customfield" WHERE "documents_customfield"."id"' in query["sql"] ] self.assertEqual( len(custom_field_lookups), 0, "Expected the second, separately-instantiated serializer to reuse " f"the already-resolved CustomField, got: {custom_field_lookups}", ) def test_custom_field_validation_rejects_malformed_field_value(self) -> None: """ GIVEN: - A document is being validated with a malformed custom_fields entry whose "field" value is neither a valid CustomField id nor a type DRF's own PrimaryKeyRelatedField can safely reject on its own (unhashable, or a non-numeric scalar) WHEN: - The serializer is validated THEN: - A normal validation error is raised, not an unhandled TypeError/ValueError escaping past DRF's validation layer """ doc = DocumentFactory(mime_type="application/pdf") bad_field_values = { "unhashable-list": [], "unhashable-dict": {}, "non-numeric-scalar": "abc", } for case_id, bad_field_value in bad_field_values.items(): with self.subTest(case_id): serializer = DocumentSerializer( doc, data={ "custom_fields": [ {"field": bad_field_value, "value": "test value"}, ], }, partial=True, ) self.assertFalse(serializer.is_valid()) self.assertIn("custom_fields", serializer.errors) def test_document_metadata_overrides_from_document_batches_field_lookup( self, ) -> None: """ GIVEN: - A document has several custom field values - The document instance has just been refreshed from the database, which drops any prefetched related objects (as send_websocket_document_updated does before building overrides) WHEN: - DocumentMetadataOverrides.from_document() reads the document's custom field values THEN: - The referenced CustomField objects are resolved with a single query, not one query per custom field """ doc = DocumentFactory(mime_type="application/pdf") for i in range(5): CustomFieldInstance.objects.create( document=doc, field=CustomField.objects.create( name=f"Test Custom Field {i}", data_type=CustomField.FieldDataType.STRING, ), value_text="value", ) doc.refresh_from_db() with CaptureQueriesContext(connection) as ctx: overrides = DocumentMetadataOverrides.from_document(doc) self.assertEqual(len(overrides.custom_fields), 5) unbatched_field_lookups = [ query for query in ctx.captured_queries if 'FROM "documents_customfield" WHERE "documents_customfield"."id"' in query["sql"] ] self.assertEqual( unbatched_field_lookups, [], "Expected CustomField data to come from the CustomFieldInstance " "join, not a separate per-instance lookup, " f"got: {unbatched_field_lookups}", ) def test_change_custom_field_instance_value(self) -> None: """ GIVEN: - Custom field instance is created and attached to document WHEN: - API request to create change the value of the custom field THEN: - the field instance is updated - the field returns the correct fields and values """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_string = CustomField.objects.create( name="Test Custom Field String", data_type=CustomField.FieldDataType.STRING, ) self.assertEqual(CustomFieldInstance.objects.count(), 0) # Create resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_string.id, "value": "test value", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(CustomFieldInstance.objects.count(), 1) _cf_2 = doc.custom_fields.first() assert _cf_2 is not None self.assertEqual(_cf_2.value, "test value") # Update resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_string.id, "value": "a new test value", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(CustomFieldInstance.objects.count(), 1) _cf_3 = doc.custom_fields.first() assert _cf_3 is not None self.assertEqual(_cf_3.value, "a new test value") def test_delete_custom_field_instance(self) -> None: """ GIVEN: - Multiple custom field instances are created and attached to document WHEN: - API request to remove a field THEN: - the field instance is removed - the other field remains unchanged - the field returns the correct fields and values """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_string = CustomField.objects.create( name="Test Custom Field String", data_type=CustomField.FieldDataType.STRING, ) custom_field_date = CustomField.objects.create( name="Test Custom Field Date", data_type=CustomField.FieldDataType.DATE, ) date_value = date.today() resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_string.id, "value": "a new test value", }, { "field": custom_field_date.id, "value": date_value.isoformat(), }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(CustomFieldInstance.objects.count(), 2) self.assertEqual(len(doc.custom_fields.all()), 2) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_date.id, "value": date_value.isoformat(), }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(CustomFieldInstance.objects.count(), 1) self.assertEqual(Document.objects.count(), 1) self.assertEqual(len(doc.custom_fields.all()), 1) _cf_4 = doc.custom_fields.first() assert _cf_4 is not None self.assertEqual(_cf_4.value, date_value) def test_custom_field_validation(self) -> None: """ GIVEN: - Document exists with no fields WHEN: - API request to remove a field - API request is not valid THEN: - HTTP 400 is returned - No field created - No field attached to the document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_string = CustomField.objects.create( name="Test Custom Field String", data_type=CustomField.FieldDataType.STRING, ) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_string.id, # Whoops, spelling "valeu": "a new test value", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) self.assertEqual(len(doc.custom_fields.all()), 0) def test_custom_field_value_url_validation(self) -> None: """ GIVEN: - Document & custom field exist WHEN: - API request to set a field value to something which is or is not a link THEN: - HTTP 400 is returned - No field instance is created or attached to the document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_url = CustomField.objects.create( name="Test Custom Field URL", data_type=CustomField.FieldDataType.URL, ) for value in ["not a url", "file:"]: with self.subTest(f"Test value {value}"): resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_url.id, "value": value, }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) self.assertEqual(len(doc.custom_fields.all()), 0) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_url.id, "value": "tel:+1-816-555-1212", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) def test_custom_field_value_integer_validation(self) -> None: """ GIVEN: - Document & custom field exist WHEN: - API request to set a field value to something not an integer THEN: - HTTP 400 is returned - No field instance is created or attached to the document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_int = CustomField.objects.create( name="Test Custom Field INT", data_type=CustomField.FieldDataType.INT, ) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_int.id, "value": "not an int", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) self.assertEqual(len(doc.custom_fields.all()), 0) def test_custom_field_value_monetary_validation(self) -> None: """ GIVEN: - Document & custom field exist WHEN: - API request to set a field value to something not a valid monetary decimal (legacy) or not a new monetary format e.g. USD12.34 THEN: - HTTP 400 is returned - No field instance is created or attached to the document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_money = CustomField.objects.create( name="Test Custom Field MONETARY", data_type=CustomField.FieldDataType.MONETARY, ) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_money.id, # Too many places past decimal "value": 12.123, }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_money.id, # Too many places past decimal "value": "GBP12.123", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ { "field": custom_field_money.id, # Not a 3-letter currency code "value": "G12.12", }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) self.assertEqual(len(doc.custom_fields.all()), 0) def test_custom_field_value_short_text_validation(self) -> None: """ GIVEN: - Document & custom field exist WHEN: - API request to set a field value to a too long string THEN: - HTTP 400 is returned - No field instance is created or attached to the document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_string = CustomField.objects.create( name="Test Custom Field STRING", data_type=CustomField.FieldDataType.STRING, ) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ {"field": custom_field_string.id, "value": "a" * 129}, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) self.assertEqual(len(doc.custom_fields.all()), 0) def test_custom_field_value_select_validation(self) -> None: """ GIVEN: - Document & custom field exist WHEN: - API request to set a field value to something not in the select options THEN: - HTTP 400 is returned - No field instance is created or attached to the document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_select = CustomField.objects.create( name="Test Custom Field SELECT", data_type=CustomField.FieldDataType.SELECT, extra_data={ "select_options": [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 2", "id": "def-456"}, ], }, ) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ {"field": custom_field_select.id, "value": "not an option"}, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) self.assertEqual(len(doc.custom_fields.all()), 0) def test_custom_field_value_documentlink_validation(self) -> None: """ GIVEN: - Document & custom field exist WHEN: - API request to set a field value to a document that does not exist - API request to set a field value to empty string THEN: - HTTP 400 is returned - No field instance is created or attached to the document """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) custom_field_documentlink = CustomField.objects.create( name="Test Custom Field Doc Link", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ {"field": custom_field_documentlink.id, "value": [999]}, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) self.assertEqual(len(doc.custom_fields.all()), 0) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": [ {"field": custom_field_documentlink.id, "value": ""}, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(CustomFieldInstance.objects.count(), 0) def test_custom_field_not_null(self) -> None: """ GIVEN: - Existing document WHEN: - API request with custom_fields set to null THEN: - HTTP 400 is returned """ doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", ) resp = self.client.patch( f"/api/documents/{doc.id}/", data={ "custom_fields": None, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_symmetric_doclink_fields(self) -> None: """ GIVEN: - Existing document WHEN: - Doc links are added or removed THEN: - Symmetrical link is created or removed as expected """ doc1 = Document.objects.create( title="WOW1", content="1", checksum="1", mime_type="application/pdf", ) doc2 = Document.objects.create( title="WOW2", content="the content2", checksum="2", mime_type="application/pdf", ) doc3 = Document.objects.create( title="WOW3", content="the content3", checksum="3", mime_type="application/pdf", ) doc4 = Document.objects.create( title="WOW4", content="the content4", checksum="4", mime_type="application/pdf", ) custom_field_doclink = CustomField.objects.create( name="Test Custom Field Doc Link", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) # Add links, creates bi-directional resp = self.client.patch( f"/api/documents/{doc1.id}/", data={ "custom_fields": [ { "field": custom_field_doclink.id, "value": [2, 3, 4], }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(CustomFieldInstance.objects.count(), 4) _cf_5 = doc2.custom_fields.first() assert _cf_5 is not None self.assertEqual(_cf_5.value, [1]) _cf_6 = doc3.custom_fields.first() assert _cf_6 is not None self.assertEqual(_cf_6.value, [1]) _cf_7 = doc4.custom_fields.first() assert _cf_7 is not None self.assertEqual(_cf_7.value, [1]) # Add links appends if necessary resp = self.client.patch( f"/api/documents/{doc3.id}/", data={ "custom_fields": [ { "field": custom_field_doclink.id, "value": [1, 4], }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) _cf_8 = doc4.custom_fields.first() assert _cf_8 is not None self.assertEqual(_cf_8.value, [1, 3]) # Remove one of the links, removed on other doc resp = self.client.patch( f"/api/documents/{doc1.id}/", data={ "custom_fields": [ { "field": custom_field_doclink.id, "value": [2, 3], }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) _cf_9 = doc2.custom_fields.first() assert _cf_9 is not None self.assertEqual(_cf_9.value, [1]) _cf_10 = doc3.custom_fields.first() assert _cf_10 is not None self.assertEqual(_cf_10.value, [1, 4]) _cf_11 = doc4.custom_fields.first() assert _cf_11 is not None self.assertEqual(_cf_11.value, [3]) # Removes the field entirely resp = self.client.patch( f"/api/documents/{doc1.id}/", data={ "custom_fields": [], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) _cf_12 = doc2.custom_fields.first() assert _cf_12 is not None self.assertEqual(_cf_12.value, []) _cf_13 = doc3.custom_fields.first() assert _cf_13 is not None self.assertEqual(_cf_13.value, [4]) _cf_14 = doc4.custom_fields.first() assert _cf_14 is not None self.assertEqual(_cf_14.value, [3]) # If field exists on target doc but value is None doc5 = Document.objects.create( title="WOW5", content="the content4", checksum="5", mime_type="application/pdf", ) CustomFieldInstance.objects.create(document=doc5, field=custom_field_doclink) resp = self.client.patch( f"/api/documents/{doc1.id}/", data={ "custom_fields": [ { "field": custom_field_doclink.id, "value": [doc5.id], }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) _cf_15 = doc5.custom_fields.first() assert _cf_15 is not None self.assertEqual(_cf_15.value, [1]) def test_documentlink_patch_requires_change_permission_on_target_documents( self, ) -> None: source_owner = User.objects.create_user(username="source-owner") source_owner.user_permissions.add( Permission.objects.get(codename="change_document"), ) other_user = User.objects.create_user(username="other-user") source_doc = Document.objects.create( title="Source", checksum="source", mime_type="application/pdf", owner=source_owner, ) target_doc = Document.objects.create( title="Target", checksum="target", mime_type="application/pdf", owner=other_user, ) custom_field_doclink = CustomField.objects.create( name="Test Custom Field Doc Link", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) self.client.force_authenticate(user=source_owner) resp = self.client.patch( f"/api/documents/{source_doc.id}/", data={ "custom_fields": [ { "field": custom_field_doclink.id, "value": [target_doc.id], }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual( CustomFieldInstance.objects.filter(field=custom_field_doclink).count(), 0, ) def test_documentlink_patch_allowed_with_change_permission_on_target_documents( self, ) -> None: source_owner = User.objects.create_user(username="source-owner") source_owner.user_permissions.add( Permission.objects.get(codename="change_document"), ) other_user = User.objects.create_user(username="other-user") source_doc = Document.objects.create( title="Source", checksum="source", mime_type="application/pdf", owner=source_owner, ) target_doc = Document.objects.create( title="Target", checksum="target", mime_type="application/pdf", owner=other_user, ) custom_field_doclink = CustomField.objects.create( name="Test Custom Field Doc Link", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) assign_perm("change_document", source_owner, target_doc) self.client.force_authenticate(user=source_owner) resp = self.client.patch( f"/api/documents/{source_doc.id}/", data={ "custom_fields": [ { "field": custom_field_doclink.id, "value": [target_doc.id], }, ], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) target_doc.refresh_from_db() self.assertEqual( target_doc.custom_fields.get(field=custom_field_doclink).value, [source_doc.id], ) def test_custom_field_filters(self) -> None: custom_field_string = CustomField.objects.create( name="Test Custom Field String", data_type=CustomField.FieldDataType.STRING, ) custom_field_date = CustomField.objects.create( name="Test Custom Field Date", data_type=CustomField.FieldDataType.DATE, ) custom_field_int = CustomField.objects.create( name="Test Custom Field Int", data_type=CustomField.FieldDataType.INT, ) response = self.client.get( f"{self.ENDPOINT}?id={custom_field_string.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) response = self.client.get( f"{self.ENDPOINT}?id__in={custom_field_string.id},{custom_field_date.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) response = self.client.get( f"{self.ENDPOINT}?name__icontains=Int", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["name"], custom_field_int.name) def test_custom_fields_document_count(self) -> None: custom_field_string = CustomField.objects.create( name="Test Custom Field String", data_type=CustomField.FieldDataType.STRING, ) doc = Document.objects.create( title="WOW", content="the content", checksum="123", mime_type="application/pdf", owner=self.user, ) response = self.client.get( f"{self.ENDPOINT}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(results[0]["document_count"], 0) CustomFieldInstance.objects.create( document=doc, field=custom_field_string, value_text="test value", ) response = self.client.get( f"{self.ENDPOINT}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(results[0]["document_count"], 1) # Test as user without access to the document non_superuser = User.objects.create_user(username="non_superuser") non_superuser.user_permissions.add( *Permission.objects.exclude(codename="view_global_statistics"), ) non_superuser.save() self.client.force_authenticate(user=non_superuser) self.client.force_login(user=non_superuser) response = self.client.get( f"{self.ENDPOINT}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(results[0]["document_count"], 0) def test_patch_document_invalid_date_custom_field_returns_validation_error( self, ) -> None: """ GIVEN: - A date custom field - A document WHEN: - Patching the document with a date string in the wrong format THEN: - HTTP 400 is returned instead of an internal server error - No custom field instance is created """ cf_date = CustomField.objects.create( name="datefield", data_type=CustomField.FieldDataType.DATE, ) doc = Document.objects.create( title="Doc", checksum="123", mime_type="application/pdf", ) response = self.client.patch( f"/api/documents/{doc.pk}/", { "custom_fields": [ { "field": cf_date.pk, "value": "10.03.2026", }, ], }, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("custom_fields", response.data) self.assertEqual(CustomFieldInstance.objects.count(), 0)