Feat/devsu 2815 make create report access project specific

pori_python/ipr/connection.py

@@ -93,10 +93,54 @@ def delete(self, uri: str, data: Dict = {}, **kwargs) -> Dict:
             **kwargs,
         )
 
+    def check_upload_permission(self, project_name: str) -> None:
+        """Check that the current user has permission to upload to the given project.
+
+        Fetches all projects and the current user info (including groups and
+        projects) up front. Checks for admin, manager, create report access,
+        all projects access, and project membership.
+        """
+        projects = self.get('project')
+        project_exists = any(p['name'] == project_name for p in projects)
+        if not project_exists:
+            raise Exception(
+                f'Project {project_name} does not exist or user does not have permission to view it'
+            )
+
+        user = self.get('user/me')
+        user_groups = user.get('groups', []) if isinstance(user, dict) else []
+        group_names = {
+            group.get('name', '').strip().lower()
+            if isinstance(group, dict)
+            else group.strip().lower()
+            for group in user_groups
+        }
+
+        is_admin = 'admin' in group_names
+        is_manager = 'manager' in group_names
+        has_create_report_access = 'create report access' in group_names
+        has_all_projects_access = 'all projects access' in group_names
+
+        # admins and managers can always create reports
+        can_create_report = is_admin or is_manager or has_create_report_access
+
+        user_projects = user.get('projects', []) if isinstance(user, dict) else []
+        has_project_access = (
+            is_admin
+            or has_all_projects_access
+            or any(isinstance(p, dict) and p.get('name') == project_name for p in user_projects)
+        )
+
+        if not can_create_report:
+            raise Exception('User does not have report creation permission')
+
+        if not has_project_access:
+            raise Exception(f'User has no permission to create report in project {project_name}')
+
     def upload_report(
         self,
         content: Dict,
-        mins_to_wait: int = 5,
+        mins_to_wait: int = 10,
         async_upload: bool = False,
         ignore_extra_fields: bool = False,
     ) -> Dict:
@@ -105,19 +149,6 @@ def upload_report(
             # or 'report'. jobStatus is no longer available once the report is successfully
             # uploaded.
 
-            projects = self.get('project')
-            project_names = [item['name'] for item in projects]
-
-            # if project is not exist, create one
-            if content['project'] not in project_names:
-                logger.info(
-                    f'Project not found - attempting to create project {content["project"]}'
-                )
-                try:
-                    self.post('project', {'name': content['project']})
-                except Exception as err:
-                    raise Exception(f'Project creation failed due to {err}')
-
             if ignore_extra_fields:
                 initial_result = self.post('reports-async?ignore_extra_fields=true', content)
             else:

pori_python/ipr/main.py

@@ -152,7 +152,7 @@ def command_interface() -> None:
     )
     parser.add_argument(
         '--mins_to_wait',
-        default=5,
+        default=10,
         action='store',
         help='is using reports-async, number of minutes to wait before throwing error',
     )
@@ -337,7 +337,7 @@ def ipr_report(
     match_germline: bool = False,
     custom_kb_match_filter: Optional[Callable] = None,
     async_upload: bool = False,
-    mins_to_wait: int = 5,
+    mins_to_wait: int = 10,
     include_ipr_variant_text: bool = True,
     include_nonspecific_disease: bool = False,
     include_nonspecific_project: bool = False,
@@ -396,6 +396,10 @@ def ipr_report(
     else:
         logger.warning('No ipr_url given')
 
+    # Verify upload permission before doing any expensive processing
+    if ipr_upload and ipr_conn:
+        ipr_conn.check_upload_permission(content['project'])
+
     if validate_json:
         if not ipr_conn:
             raise ValueError('ipr_url required to validate json')

tests/test_ipr/test_connection.py

@@ -95,3 +95,125 @@ def request(*args, **kwargs):
                         )
                     },
                 )
+
+
+class TestCheckUploadPermission:
+    def _user_response(self, groups=None, projects=None):
+        return {
+            'groups': [{'name': g} for g in (groups or [])],
+            'projects': [{'name': p} for p in (projects or [])],
+        }
+
+    def test_rejects_user_without_create_report_access(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(
+            side_effect=[[{'name': 'TEST'}], self._user_response(projects=['TEST'])]
+        )
+        conn.post = mock.MagicMock()
+
+        with pytest.raises(Exception, match='User does not have report creation permission'):
+            conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()
+
+    def test_rejects_user_without_project_access(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(
+            side_effect=[
+                [{'name': 'TEST'}],
+                self._user_response(groups=['create report access'], projects=['OTHER']),
+            ]
+        )
+        conn.post = mock.MagicMock()
+
+        with pytest.raises(
+            Exception, match='User has no permission to create report in project TEST'
+        ):
+            conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()
+
+    def test_allows_user_with_project_and_create_report_access(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(
+            side_effect=[
+                [{'name': 'TEST'}],
+                self._user_response(groups=['create report access'], projects=['TEST']),
+            ]
+        )
+        conn.post = mock.MagicMock()
+
+        conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()
+
+    def test_project_not_found_raises(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(side_effect=[[{'name': 'OTHER'}]])
+        conn.post = mock.MagicMock()
+
+        with pytest.raises(Exception, match='Project TEST does not exist'):
+            conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()
+
+    def test_manager_without_project_membership_raises(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(
+            side_effect=[
+                [{'name': 'TEST'}],
+                self._user_response(groups=['manager'], projects=[]),
+            ]
+        )
+        conn.post = mock.MagicMock()
+
+        with pytest.raises(
+            Exception, match='User has no permission to create report in project TEST'
+        ):
+            conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()
+
+    def test_manager_with_project_membership_allowed(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(
+            side_effect=[
+                [{'name': 'TEST'}],
+                self._user_response(groups=['manager'], projects=['TEST']),
+            ]
+        )
+        conn.post = mock.MagicMock()
+
+        conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()
+
+    def test_admin_bypasses_all_checks(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(
+            side_effect=[
+                [{'name': 'TEST'}],
+                self._user_response(groups=['admin'], projects=[]),
+            ]
+        )
+        conn.post = mock.MagicMock()
+
+        conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()
+
+    def test_all_projects_access_without_project_membership(self):
+        conn = IprConnection('user', 'pass')
+        conn.get = mock.MagicMock(
+            side_effect=[
+                [{'name': 'TEST'}],
+                self._user_response(
+                    groups=['create report access', 'all projects access'], projects=[]
+                ),
+            ]
+        )
+        conn.post = mock.MagicMock()
+
+        conn.check_upload_permission('TEST')
+
+        conn.post.assert_not_called()

tests/test_ipr/test_main.py

@@ -110,6 +110,11 @@ def side_effect_function(*args, **kwargs):
             return [{'name': 'genomic', 'ident': '001'}]
         elif args[0] == 'project':
             return [{'name': 'TEST', 'ident': '001'}]
+        elif args[0] == 'user/me':
+            return {
+                'groups': [{'name': 'admin'}],
+                'projects': [{'name': 'TEST'}],
+            }
         else:
             return []
 

tests/test_ipr/test_probe.py

@@ -25,6 +25,11 @@ def side_effect_function(*args, **kwargs):
             return [{'name': 'genomic', 'ident': '001'}]
         elif args[0] == 'project':
             return [{'name': 'TEST', 'ident': '001'}]
+        elif args[0] == 'user/me':
+            return {
+                'groups': [{'name': 'admin'}],
+                'projects': [{'name': 'TEST'}],
+            }
         else:
             return []