[Fixes #12124] Fix broken tests

geonode/assets/local.py

@@ -60,7 +60,7 @@ def remove_data(self, asset: LocalAsset):
         for dir in removed_dir:
             if not os.listdir(dir):
                 logger.info(f"Removing empty asset directory {dir}")
-                os.remove(dir)
+                os.rmdir(dir)
 
     def replace_data(self, asset: LocalAsset, files: list):
         self.remove_data(asset)

geonode/base/api/tests.py

@@ -2308,7 +2308,7 @@ def test_resource_service_copy_with_perms_dataset(self):
     @override_settings(ASYNC_SIGNALS=False, FILE_UPLOAD_DIRECTORY_PERMISSIONS=0o777, FILE_UPLOAD_PERMISSIONS=0o7777)
     def test_resource_service_copy_with_perms_dataset_set_default_perms(self):
         with self.settings(ASYNC_SIGNALS=False):
-           
+
             files_as_dict, resource = self._import_dataset()
 
             _, _ = create_asset_and_link(
@@ -2327,7 +2327,7 @@ def test_resource_service_copy_with_perms_dataset_set_default_perms(self):
 
             self.assertTrue(self.client.login(username="admin", password="admin"))
             payload = QueryDict("", mutable=True)
-            payload.update({"defaults": '{"title": ' + resource.title + ' }'})
+            payload.update({"defaults": '{"title": ' + resource.title + " }"})
             response = self.client.put(copy_url, data=payload)
             self.assertEqual(response.status_code, 200)
 
@@ -2343,23 +2343,21 @@ def test_resource_service_copy_with_perms_dataset_set_default_perms(self):
 
     def _import_dataset(self):
         files_as_dict = {
-                "base_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.shp"),
-                "dbf_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.dbf"),
-                "prj_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.shx"),
-                "shx_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.prj"),
-            }
-        payload = {
-                _filename: open(_file, "rb") for _filename, _file in files_as_dict.items()
-            }
+            "base_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.shp"),
+            "dbf_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.dbf"),
+            "prj_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.shx"),
+            "shx_file": os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.prj"),
+        }
+        payload = {_filename: open(_file, "rb") for _filename, _file in files_as_dict.items()}
 
         _url = reverse("importer_upload")
         self.client.force_login(get_user_model().objects.get(username="admin"))
 
         response = self.client.post(_url, data=payload)
         self.assertEqual(201, response.status_code)
 
-        resource = ResourceHandlerInfo.objects.get(execution_request_id=response.json()['execution_id'])
-        return files_as_dict,resource
+        resource = ResourceHandlerInfo.objects.get(execution_request_id=response.json()["execution_id"])
+        return files_as_dict, resource
 
     def test_resource_service_copy_with_perms_doc(self):
         files = os.path.join(gisdata.GOOD_DATA, "vector/san_andres_y_providencia_water.shp")

geonode/base/models.py

@@ -88,7 +88,6 @@
 from urllib.parse import urlsplit, urljoin
 from geonode.storage.manager import storage_manager
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -607,10 +606,14 @@ def upload_files(resource_id, files, force=False):
     @staticmethod
     def cleanup_uploaded_files(resource_id):
         """Remove uploaded files, if any"""
+        from geonode.assets.utils import get_default_asset
+
         if ResourceBase.objects.filter(id=resource_id).exists():
             _resource = ResourceBase.objects.filter(id=resource_id).get()
             _uploaded_folder = None
-            if _resource.files:
+            asset = get_default_asset(_resource)
+            files = asset.location if asset else []
+            if files:
                 for _file in _resource.files:
                     try:
                         if storage_manager.exists(_file):

geonode/documents/tests.py

@@ -418,7 +418,9 @@ def test_ajax_document_permissions(self, create_thumb):
         # Setup some document names to work with
         superuser = get_user_model().objects.get(pk=2)
         document = resource_manager.create(
-            None, resource_type=Document, defaults=dict(files=[TEST_GIF], owner=superuser, title="theimg", is_approved=True)
+            None,
+            resource_type=Document,
+            defaults=dict(files=[TEST_GIF], owner=superuser, title="theimg", is_approved=True),
         )
         document_id = document.id
         invalid_document_id = 20
@@ -809,7 +811,7 @@ def test_document_link_with_permissions(self):
         # Access resource with user logged-in
         self.client.login(username=self.not_admin.username, password="very-secret")
         response = self.client.get(self.doc_link_url)
-        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response.status_code, 200)
         # test document link with external url
         doc = resource_manager.create(
             None,

geonode/proxy/templatetags/proxy_lib_tags.py

@@ -17,6 +17,7 @@
 #
 #########################################################################
 
+from geonode.assets.utils import get_default_asset
 from geonode.base.models import ResourceBase
 import traceback
 
@@ -52,7 +53,10 @@ def original_link_available(context, resourceid, url):
     dataset_files = []
     if isinstance(instance, ResourceBase):
         try:
-            for file in instance.files:
+            asset_obj = get_default_asset(instance)
+            # Copy all Dataset related files into a temporary folder
+            files = asset_obj.location if asset_obj else []
+            for file in files:
                 dataset_files.append(file)
                 if not storage_manager.exists(file):
                     return False

geonode/proxy/tests.py

@@ -30,6 +30,7 @@
 from urllib.parse import urljoin
 
 from django.conf import settings
+from geonode.assets.utils import create_asset_and_link
 from geonode.proxy.templatetags.proxy_lib_tags import original_link_available
 from django.test.client import RequestFactory
 from django.core.files.uploadedfile import SimpleUploadedFile
@@ -264,12 +265,15 @@ def test_download_url_with_existing_files(self, fopen, fexists):
         fopen.return_value = SimpleUploadedFile("foo_file.shp", b"scc")
         dataset = Dataset.objects.all().first()
 
-        dataset.files = [
+        dataset_files = [
             "/tmpe1exb9e9/foo_file.dbf",
             "/tmpe1exb9e9/foo_file.prj",
             "/tmpe1exb9e9/foo_file.shp",
             "/tmpe1exb9e9/foo_file.shx",
         ]
+        asset, link = create_asset_and_link(
+            dataset, get_user_model().objects.get(username="admin"), dataset_files, clone_files=False
+        )
 
         dataset.save()
 
@@ -287,6 +291,9 @@ def test_download_url_with_existing_files(self, fopen, fexists):
         self.assertEqual("application/zip", response.headers.get("Content-Type"))
         self.assertEqual('attachment; filename="CA.zip"', response.headers.get("Content-Disposition"))
 
+        link.delete()
+        asset.delete()
+
     @patch("geonode.storage.manager.storage_manager.exists")
     @patch("geonode.storage.manager.storage_manager.open")
     @on_ogc_backend(geoserver.BACKEND_PACKAGE)
@@ -295,12 +302,15 @@ def test_download_files(self, fopen, fexists):
         fopen.return_value = SimpleUploadedFile("foo_file.shp", b"scc")
         dataset = Dataset.objects.all().first()
 
-        dataset.files = [
+        dataset_files = [
             "/tmpe1exb9e9/foo_file.dbf",
             "/tmpe1exb9e9/foo_file.prj",
             "/tmpe1exb9e9/foo_file.shp",
             "/tmpe1exb9e9/foo_file.shx",
         ]
+        asset, link = create_asset_and_link(
+            dataset, get_user_model().objects.get(username="admin"), dataset_files, clone_files=False
+        )
 
         dataset.save()
 
@@ -324,6 +334,9 @@ def test_download_files(self, fopen, fexists):
         self.assertIn(".shx", "".join(zip_files))
         self.assertIn(".prj", "".join(zip_files))
 
+        link.delete()
+        asset.delete()
+
 
 class OWSApiTestCase(GeoNodeBaseTestSupport):
     def setUp(self):
@@ -376,16 +389,23 @@ def test_should_return_true_if_files_are_available(self, fexists):
 
         assert upload
 
-        self.resource.files = [
+        dataset_files = [
             "/tmpe1exb9e9/foo_file.dbf",
             "/tmpe1exb9e9/foo_file.prj",
             "/tmpe1exb9e9/foo_file.shp",
             "/tmpe1exb9e9/foo_file.shx",
         ]
 
+        asset, link = create_asset_and_link(
+            self.resource, get_user_model().objects.get(username="admin"), dataset_files, clone_files=False
+        )
+
         self.resource.save()
 
         self.resource.refresh_from_db()
 
         actual = original_link_available(self.context, self.resource.resourcebase_ptr_id, self.url)
         self.assertTrue(actual)
+
+        link.delete()
+        asset.delete()

geonode/proxy/views.py

@@ -53,6 +53,7 @@
 from geonode import geoserver  # noqa
 from geonode.base import register_event
 from geonode.base.auth import get_auth_user, get_token_from_auth_header
+from geonode.assets.utils import get_default_asset
 
 BUFFER_CHUNK_SIZE = 64 * 1024
 
@@ -281,8 +282,9 @@ def download(request, resourceid, sender=Dataset):
         dataset_files = []
         file_list = []  # Store file info to be returned
         try:
-            files = instance.resourcebase_ptr.files
+            asset_obj = get_default_asset(instance)
             # Copy all Dataset related files into a temporary folder
+            files = asset_obj.location if asset_obj else []
             for file_path in files:
                 if storage_manager.exists(file_path):
                     dataset_files.append(file_path)

geonode/resource/manager.py

@@ -40,6 +40,7 @@
 
 
 from geonode.base.models import ResourceBase, LinkedResource, Link
+from geonode.geoapps.models import GeoApp
 from geonode.thumbs.thumbnails import _generate_thumbnail_name
 from geonode.documents.tasks import create_document_thumbnail
 from geonode.security.permissions import PermSpecCompact, DATA_STYLABLE_RESOURCES_SUBTYPES
@@ -554,10 +555,13 @@ def copy(
                     links = self._copy_data(instance, target=_resource)
                     # we're just merging all the files together: it won't work once we have multiple assets per resource
                     # TODO: get the files from the proper Asset
-                    files = []
-                    for link in links:
-                        files.extend(link.asset.location)
-                    to_update = {"files": files}
+                    to_update = {}
+
+                    if not isinstance(instance.get_real_instance(), (Map, GeoApp)):
+                        files = [link.asset.location for link in links]
+                        if files:
+                            to_update = {"files": files}
+
                     _resource = self._concrete_resource_manager.copy(instance, uuid=_resource.uuid, defaults=to_update)
 
             except Exception as e:

geonode/resource/tests.py

@@ -157,7 +157,7 @@ def _copy_assert_resource(res, title):
         res = self.rm.ingest(
             dt_files,
             resource_type=Dataset,
-            defaults={"owner": self.user, "title": "Testing Dataset", "files": dt_files},
+            defaults={"owner": self.user, "title": "Testing Dataset"},
         )
         self.assertTrue(isinstance(res, Dataset))
         _copy_assert_resource(res, "Testing Dataset 2")

geonode/resource/utils.py

@@ -457,8 +457,10 @@ def resourcebase_post_save(instance, *args, **kwargs):
         if hasattr(instance, "abstract") and not getattr(instance, "abstract", None):
             instance.abstract = _("No abstract provided")
         if hasattr(instance, "title") and not getattr(instance, "title", None) or getattr(instance, "title", "") == "":
-            if isinstance(instance, Document) and instance.files:
-                instance.title = os.path.basename(instance.files[0])
+            asset = get_default_asset(instance)
+            files = asset.location if asset else []
+            if isinstance(instance, Document) and files:
+                instance.title = os.path.basename(files[0])
             if hasattr(instance, "name") and getattr(instance, "name", None):
                 instance.title = instance.name
         if (

geonode/security/tests.py

@@ -785,7 +785,8 @@ def test_dataset_permissions(self):
 
         # test view_resourcebase permission on anonymous user
         response = requests.get(url)
-        self.assertTrue(response.status_code, 404)
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(b"Could not find layer" in response.content)
         self.assertEqual(response.headers.get("Content-Type"), "application/vnd.ogc.se_xml;charset=UTF-8")
 
         # test WMS with authenticated user that has access to the Dataset
@@ -795,7 +796,7 @@ def test_dataset_permissions(self):
                 username=settings.OGC_SERVER["default"]["USER"], password=settings.OGC_SERVER["default"]["PASSWORD"]
             ),
         )
-        self.assertTrue(response.status_code, 200)
+        self.assertEqual(response.status_code, 200)
         self.assertEqual(response.headers.get("Content-Type"), "image/png")
 
         # test WMS with authenticated user that has no view_resourcebase:

geonode/storage/manager.py

@@ -181,7 +181,7 @@ def copy(self, resource, target=None):
         #     updated_files["files"] = self.copy_files_list(resource.files)
         # return updated_files
 
-    def copy_files_list(self, files: List[str], dir=None, dir_prefix=None, dir_suffix=None):
+    def copy_files_list(self, files: List[str], dir=settings.MEDIA_ROOT, dir_prefix=None, dir_suffix=None):
         from geonode.utils import mkdtemp
 
         out = []