Permission UploadViewSet (#4297)

* Permission UploadViewSet

* Fix

---------

Co-authored-by: kflemin <[email protected]>

seed/lib/superperms/orgs/decorators.py

@@ -225,7 +225,7 @@ def _wrapped(request, *args, **kwargs):
     return decorator
 
 
-def assert_hierarchy_access(request, property_id_kwarg=None, property_view_id_kwarg=None, param_property_view_id=None, taxlot_view_id_kwarg=None, import_file_id_kwarg=None, param_import_file_id=None, import_record_id_kwarg=None, body_ali_id=None, body_import_file_id=None, analysis_id_kwarg=None, *args, **kwargs):
+def assert_hierarchy_access(request, property_id_kwarg=None, property_view_id_kwarg=None, param_property_view_id=None, taxlot_view_id_kwarg=None, import_file_id_kwarg=None, param_import_file_id=None, import_record_id_kwarg=None, body_ali_id=None, body_import_file_id=None, analysis_id_kwarg=None, body_import_record_id=None, param_import_record_id=None, *args, **kwargs):
     """Helper function to has_hierarchy_access"""
     body = request.data
     params = request.GET
@@ -266,6 +266,14 @@ def assert_hierarchy_access(request, property_id_kwarg=None, property_view_id_kw
             import_file = ImportFile.objects.get(pk=body[body_import_file_id])
             requests_ali = import_file.access_level_instance
 
+        elif body_import_record_id and body_import_record_id in body:
+            import_record = ImportRecord.objects.get(pk=body[body_import_record_id])
+            requests_ali = import_record.access_level_instance
+
+        elif param_import_record_id and param_import_record_id in params:
+            import_record = ImportRecord.objects.get(pk=params[param_import_record_id])
+            requests_ali = import_record.access_level_instance
+
         elif analysis_id_kwarg and analysis_id_kwarg in kwargs:
             analysis = Analysis.objects.get(pk=kwargs[analysis_id_kwarg])
             requests_ali = analysis.access_level_instance
@@ -288,17 +296,17 @@ def assert_hierarchy_access(request, property_id_kwarg=None, property_view_id_kw
         }, status=status.HTTP_404_NOT_FOUND)
 
 
-def has_hierarchy_access(property_id_kwarg=None, property_view_id_kwarg=None, param_property_view_id=None, taxlot_view_id_kwarg=None, import_file_id_kwarg=None, param_import_file_id=None, import_record_id_kwarg=None, body_ali_id=None, body_import_file_id=None, analysis_id_kwarg=None):
+def has_hierarchy_access(property_id_kwarg=None, property_view_id_kwarg=None, param_property_view_id=None, taxlot_view_id_kwarg=None, import_file_id_kwarg=None, param_import_file_id=None, import_record_id_kwarg=None, body_ali_id=None, body_import_file_id=None, analysis_id_kwarg=None, body_import_record_id=None, param_import_record_id=None):
     """Must be called after has_perm_class"""
     def decorator(fn):
         if 'self' in signature(fn).parameters:
             @wraps(fn)
             def _wrapped(self, request, *args, **kwargs):
-                return assert_hierarchy_access(request, property_id_kwarg, property_view_id_kwarg, param_property_view_id, taxlot_view_id_kwarg, import_file_id_kwarg, param_import_file_id, import_record_id_kwarg, body_ali_id, body_import_file_id, analysis_id_kwarg, *args, **kwargs) or fn(self, request, *args, **kwargs)
+                return assert_hierarchy_access(request, property_id_kwarg, property_view_id_kwarg, param_property_view_id, taxlot_view_id_kwarg, import_file_id_kwarg, param_import_file_id, import_record_id_kwarg, body_ali_id, body_import_file_id, analysis_id_kwarg, body_import_record_id, param_import_record_id, *args, **kwargs) or fn(self, request, *args, **kwargs)
         else:
             @wraps(fn)
             def _wrapped(request, *args, **kwargs):
-                return assert_hierarchy_access(request, property_id_kwarg, property_view_id_kwarg, param_property_view_id, taxlot_view_id_kwarg, import_file_id_kwarg, param_import_file_id, import_record_id_kwarg, body_ali_id, body_import_file_id, analysis_id_kwarg, *args, **kwargs) or fn(request, *args, **kwargs)
+                return assert_hierarchy_access(request, property_id_kwarg, property_view_id_kwarg, param_property_view_id, taxlot_view_id_kwarg, import_file_id_kwarg, param_import_file_id, import_record_id_kwarg, body_ali_id, body_import_file_id, analysis_id_kwarg, body_import_record_id, param_import_record_id, *args, **kwargs) or fn(request, *args, **kwargs)
 
         return _wrapped
 

seed/tests/test_portfoliomanager.py

@@ -8,6 +8,7 @@
 """
 import json
 import os
+from os import path
 from pathlib import Path
 from unittest import skip, skipIf
 
@@ -17,7 +18,9 @@
 from django.urls import reverse_lazy
 from xlrd import open_workbook
 
+from seed.data_importer.models import ImportRecord
 from seed.landing.models import SEEDUser as User
+from seed.tests.util import AccessLevelBaseTestCase
 from seed.utils.organizations import create_organization
 from seed.views.v3.portfolio_manager import PortfolioManagerImport
 
@@ -434,6 +437,50 @@ def test_single_property_template_for_upload(self):
         self.assertEqual(200, response.status_code)
 
 
+class UploadViewSetPermission(AccessLevelBaseTestCase):
+    def setUp(self):
+        super().setUp()
+        self.root_property = self.property_factory.get_property(access_level_instance=self.root_level_instance)
+        self.child_property = self.property_factory.get_property(access_level_instance=self.child_level_instance)
+
+        self.import_record = ImportRecord.objects.create(
+            owner=self.root_member_user, last_modified_by=self.root_member_user, super_organization=self.org, access_level_instance=self.org.root
+        )
+
+    def test_create(self):
+        filename = path.join(path.dirname(__file__), 'data', 'property_sample_data.json')
+        with open(filename, 'rb') as f:
+
+            url = reverse_lazy('api:v3:upload-list')
+            url += "?organization_id=" + str(self.org.id)
+            url += "&import_record=" + str(self.import_record.id)
+            params = {"file": f}
+
+            self.login_as_child_member()
+            response = self.client.post(url, data=params)
+            assert response.status_code == 404
+
+            self.login_as_root_member()
+            response = self.client.post(url, data=params)
+            assert response.status_code == 200
+
+    def test_create_from_pm_import(self):
+        url = reverse_lazy('api:v3:upload-create-from-pm-import')
+        params = json.dumps({
+            'properties': [],
+            'import_record_id': self.import_record.pk,
+            'organization_id': self.org.pk
+        })
+
+        self.login_as_child_member()
+        response = self.client.post(url, params, content_type='application/json')
+        assert response.status_code == 404
+
+        self.login_as_root_member()
+        response = self.client.post(url, params, content_type='application/json')
+        assert response.status_code == 200
+
+
 class PortfolioManagerSingleReportXSLX(TestCase):
     """Test downloading a single ESPM report in XSLX format."""
 

seed/views/v3/uploads.py

@@ -20,7 +20,10 @@
 
 from seed.data_importer.models import ImportFile, ImportRecord
 from seed.decorators import ajax_request_class
-from seed.lib.superperms.orgs.decorators import has_perm_class
+from seed.lib.superperms.orgs.decorators import (
+    has_hierarchy_access,
+    has_perm_class
+)
 from seed.models import PORTFOLIO_RAW, SEED_DATA_SOURCES
 from seed.utils.api import OrgMixin, api_endpoint_class
 from seed.utils.api_schema import AutoSchemaHelper
@@ -80,6 +83,7 @@ class UploadViewSet(viewsets.ViewSet, OrgMixin):
     @api_endpoint_class
     @ajax_request_class
     @has_perm_class('can_modify_data')
+    @has_hierarchy_access(param_import_record_id="import_record")
     def create(self, request):
         """
         Upload a new file to an import_record. This is a multipart/form upload.
@@ -206,6 +210,7 @@ def _get_pint_var_from_pm_value_object(pm_value):
     @api_endpoint_class
     @ajax_request_class
     @has_perm_class('can_modify_data')
+    @has_hierarchy_access(body_import_record_id="import_record_id")
     @action(detail=False, methods=['POST'], parser_classes=(JSONParser,))
     def create_from_pm_import(self, request):
         """