Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .env_test
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ HARVESTING_MONITOR_DELAY=60
SITEURL=http://localhost:8000/

ALLOWED_HOSTS="['django', 'localhost', '127.0.0.1']"
# allow internal URLs pass validation for tests
SAFE_URL_CHECK_ENABLED=False

# Data Uploader
TIME_ENABLED=True
Expand Down
19 changes: 4 additions & 15 deletions geonode/documents/api/serializers.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,28 +18,15 @@
#########################################################################
import logging

from dynamic_rest.fields.fields import DynamicComputedField
from rest_framework import serializers
from geonode.base.api.serializers import ResourceBaseSerializer
from geonode.documents.models import Document

logger = logging.getLogger(__name__)


class GeonodeFilePathField(DynamicComputedField):
def get_attribute(self, instance):
return instance.files


class DocumentFieldField(DynamicComputedField):
def get_attribute(self, instance):
return instance.files


class DocumentSerializer(ResourceBaseSerializer):
title = serializers.CharField(required=False)
file_path = GeonodeFilePathField(required=False, write_only=True)
doc_file = DocumentFieldField(required=False, write_only=True)

class Meta:
model = Document
Expand All @@ -55,9 +42,11 @@ class Meta:
"subtype",
"extension",
"mime_type",
"file_path",
"doc_file",
"doc_url",
)
)
)
# name and extension determine where the document file lives on disk
# and how it is served. POST/PUT are blocked at the http_method_names
# layer; making them read-only locks them down on PATCH too.
read_only_fields = ("name", "extension")
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The PR description states that files is locked down, but it is not included in read_only_fields. If files is a field in the serializer, it remains mutable via PATCH requests. Please add files to read_only_fields to ensure it is locked down as intended.

Suggested change
read_only_fields = ("name", "extension")
read_only_fields = ("name", "extension", "files")

176 changes: 29 additions & 147 deletions geonode/documents/api/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,8 @@
from rest_framework.test import APITestCase

from guardian.shortcuts import assign_perm, get_anonymous_user
from geonode import settings

from geonode.base.populate_test_data import create_models
from geonode.base.enumerations import SOURCE_TYPE_REMOTE
from geonode.documents.models import Document

logger = logging.getLogger(__name__)
Expand All @@ -46,9 +44,6 @@ def setUpTestData(cls):
def setUp(self):
self.admin = get_user_model().objects.get(username="admin")
self.url = reverse("documents-list")
self.invalid_file_path = f"{settings.PROJECT_ROOT}/tests/data/thesaurus.rdf"
self.valid_file_path = f"{settings.PROJECT_ROOT}/base/fixtures/test_xml.xml"
self.no_title_file_path = f"{settings.PROJECT_ROOT}/base/fixtures/test_sld.sld"

def test_documents(self):
"""
Expand Down Expand Up @@ -91,47 +86,22 @@ def test_extra_metadata_included_with_param(self):
response = self.client.get(url, format="json")
self.assertNotIn("metadata", response.data["document"])

def test_creation_return_error_if_file_is_not_passed(self):
def test_post_not_allowed(self):
"""
If file_path is not available, should raise error
Document creation via the DRF endpoint is disabled; /documents/upload
is the only supported entry point.
"""
self.client.force_login(self.admin)
payload = {"document": {"title": "New document", "metadata_only": True}}
expected = {
"success": False,
"errors": ["A file, file path or URL must be speficied"],
"code": "document_exception",
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(400, actual.status_code)
self.assertDictEqual(expected, actual.json())
response = self.client.post(self.url, data=payload, format="json")
self.assertEqual(405, response.status_code)

def test_creation_return_error_if_file_is_none(self):
"""
If file_path is not available, should raise error
"""
self.client.force_login(self.admin)
payload = {"document": {"title": "New document", "metadata_only": True, "file_path": None, "doc_file": None}}
expected = {
"success": False,
"errors": ["A file, file path or URL must be speficied"],
"code": "document_exception",
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(400, actual.status_code)
self.assertDictEqual(expected, actual.json())

def test_creation_should_rase_exec_for_unsupported_files(self):
def test_put_not_allowed(self):
document = Document.objects.first()
url = urljoin(f"{self.url}/", f"{document.id}")
self.client.force_login(self.admin)
payload = {"document": {"title": "New document", "metadata_only": True, "file_path": self.invalid_file_path}}
expected = {
"success": False,
"errors": ["The file provided is not in the supported extensions list"],
"code": "document_exception",
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(400, actual.status_code)
self.assertDictEqual(expected, actual.json())
response = self.client.put(url, data={"title": "x"}, format="json")
self.assertEqual(405, response.status_code)

def test_document_listing_advertised(self):
document = Document.objects.first()
Expand Down Expand Up @@ -166,32 +136,6 @@ def test_document_listing_advertised(self):

Document.objects.update(advertised=True)

def test_creation_should_create_the_doc(self):
"""
If file_path is not available, should raise error
"""
self.client.force_login(self.admin)
payload = {
"document": {"title": "New document for testing", "metadata_only": True, "file_path": self.valid_file_path}
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(201, actual.status_code)
extension = actual.json().get("document", {}).get("extension", "")
self.assertEqual("xml", extension)
self.assertTrue(Document.objects.filter(title="New document for testing").exists())

def test_uploading_doc_without_title(self):
"""
A document should be uploaded without specifying a title
"""
self.client.force_login(self.admin)
payload = {"document": {"metadata_only": True, "file_path": self.no_title_file_path}}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(201, actual.status_code)
extension = actual.json().get("document", {}).get("extension", "")
self.assertEqual("sld", extension)
self.assertTrue(Document.objects.filter(title="test_sld.sld").exists())

def test_patch_point_of_contact(self):
document = Document.objects.first()
url = urljoin(f"{reverse('documents-list')}/", f"{document.id}")
Expand Down Expand Up @@ -494,34 +438,6 @@ def test_patch_principal_investigator(self):
)
)

def test_creation_should_create_the_doc_and_update_the_bbox(self):
"""
If file_path is not available, should raise error
"""
self.client.force_login(self.admin)
payload = {
"document": {
"title": "New document for testing",
"metadata_only": True,
"file_path": self.valid_file_path,
"extent": {"coords": [1123692.0, 5338214.0, 1339852.0, 5482615.0], "srid": "EPSG:3857"},
},
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(201, actual.status_code)
extension = actual.json().get("document", {}).get("extension", "")
self.assertEqual("xml", extension)
doc = Document.objects.filter(title="New document for testing").all()
self.assertTrue(doc.exists())
x = doc.first()
x.refresh_from_db()
self.assertEqual("EPSG:3857", x.srid)
self.assertEqual(actual.json()["document"].get("extent")["srid"], "EPSG:4326")
self.assertEqual(
actual.json()["document"].get("extent")["coords"],
[10.094296982428332, 43.1721654049465, 12.03609530058109, 44.11086592050112],
)

def test_file_path_and_doc_path_are_not_returned(self):
"""
If file_path and doc_path should not be visible
Expand All @@ -533,62 +449,28 @@ def test_file_path_and_doc_path_are_not_returned(self):
self.assertFalse("file_path" in _doc_payload)
self.assertFalse("doc_path" in _doc_payload)

def test_creation_from_url_should_create_the_doc(self):
def test_patch_cannot_mutate_file_location_fields(self):
"""
If file_path is not available, should raise error
PATCH must not let a caller rewrite the document's on-disk location.
name, extension and files are locked down by DocumentSerializer.update().
"""
self.client.force_login(self.admin)
doc_url = "https://example.com/image"
payload = {
"document": {
"title": "New document from URL for testing",
"metadata_only": False,
"doc_url": doc_url,
"extension": "jpeg",
}
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(201, actual.status_code)
created_doc_url = actual.json().get("document", {}).get("doc_url", "")
self.assertEqual(created_doc_url, doc_url)
document = Document.objects.first()
original_name = document.name
original_extension = document.extension
original_files = list(document.files or [])

def test_remote_document_is_marked_remote(self):
"""Tests creating an external document set its sourcetype to REMOTE."""
url = urljoin(f"{self.url}/", f"{document.id}")
self.client.force_login(self.admin)
doc_url = "https://example.com/image"
payload = {
"document": {
"title": "A remote document is remote",
"doc_url": doc_url,
"extension": "jpeg",
}
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(201, actual.status_code)
created_sourcetype = actual.json().get("document", {}).get("sourcetype", "")
self.assertEqual(created_sourcetype, SOURCE_TYPE_REMOTE)

def test_either_path_or_url_doc(self):
"""
If file_path is not available, should raise error
"""
self.client.force_login(self.admin)
doc_url = "https://example.com/image"
payload = {
"document": {
"title": "New document from URL for testing",
"metadata_only": False,
"doc_url": doc_url,
"file_path": self.valid_file_path,
"extension": "jpeg",
}
patch_data = {
"name": "pwned",
"extension": "exe",
"files": ["/etc/passwd"],
}
actual = self.client.post(self.url, data=payload, format="json")
expected = {
"success": False,
"errors": ["Either a file or a URL must be specified, not both"],
"code": "document_exception",
}
actual = self.client.post(self.url, data=payload, format="json")
self.assertEqual(400, actual.status_code)
self.assertDictEqual(expected, actual.json())
response = self.client.patch(url, data=patch_data, format="json")
self.assertEqual(200, response.status_code)

document.refresh_from_db()
self.assertEqual(document.name, original_name)
self.assertEqual(document.extension, original_extension)
self.assertEqual(list(document.files or []), original_files)
Loading
Loading