Allow quality reports to regular users (#8511)

- Allowed quality report computation for tasks to regular users (before this only admins could trigger custom report computations)
- Refactored quality control tests

---------

Co-authored-by: Maria Khrustaleva <[email protected]>

Author: zhiltsov-max

changelog.d/20241004_201336_mzhiltso_allow_quality_reports_for_regular_users.md

@@ -0,0 +1,4 @@
+### Changed
+
+- \[Server API\] Quality report computation is now allowed to regular users
+  (<https://github.com/cvat-ai/cvat/pull/8511>)

cvat/apps/iam/rules/utils.rego

@@ -58,6 +58,7 @@ IMPORT_DATASET := "import:dataset"
 IMPORT_BACKUP := "import:backup"
 EXPORT_BACKUP := "export:backup"
 UPDATE_ORG := "update:organization"
+VIEW_STATUS := "view:status"
 VIEW_VALIDATION_LAYOUT := "view:validation_layout"
 UPDATE_VALIDATION_LAYOUT := "update:validation_layout"
 

cvat/apps/quality_control/permissions.py

@@ -6,9 +6,9 @@
 from typing import Optional, Union, cast
 
 from django.conf import settings
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import PermissionDenied, ValidationError
 
-from cvat.apps.engine.models import Task
+from cvat.apps.engine.models import Project, Task
 from cvat.apps.engine.permissions import TaskPermission
 from cvat.apps.iam.permissions import OpenPolicyAgentPermission, StrEnum, get_iam_context
 
@@ -18,6 +18,7 @@
 class QualityReportPermission(OpenPolicyAgentPermission):
     obj: Optional[QualityReport]
     job_owner_id: Optional[int]
+    task_id: Optional[int]
 
     class Scopes(StrEnum):
         LIST = "list"
@@ -29,7 +30,7 @@ class Scopes(StrEnum):
     def create_scope_check_status(cls, request, job_owner_id: int, iam_context=None):
         if not iam_context and request:
             iam_context = get_iam_context(request, None)
-        return cls(**iam_context, scope="view:status", job_owner_id=job_owner_id)
+        return cls(**iam_context, scope=cls.Scopes.VIEW_STATUS, job_owner_id=job_owner_id)
 
     @classmethod
     def create_scope_view(cls, request, report: Union[int, QualityReport], iam_context=None):
@@ -59,11 +60,42 @@ def create(cls, request, view, obj, iam_context):
                 elif scope == Scopes.LIST and isinstance(obj, Task):
                     permissions.append(TaskPermission.create_scope_view(request, task=obj))
                 elif scope == Scopes.CREATE:
+                    # Note: POST /api/quality/reports is used to initiate report creation and to check the process status
+                    rq_id = request.query_params.get("rq_id")
                     task_id = request.data.get("task_id")
+
+                    if not (task_id or rq_id):
+                        raise PermissionDenied("Either task_id or rq_id must be specified")
+
+                    if rq_id:
+                        # There will be another check for this case during request processing
+                        continue
+
                     if task_id is not None:
-                        permissions.append(TaskPermission.create_scope_view(request, task_id))
+                        # The request may have a different org or org unset
+                        # Here we need to retrieve iam_context for this user, based on the task_id
+                        try:
+                            task = Task.objects.get(id=task_id)
+                        except Task.DoesNotExist:
+                            raise ValidationError("The specified task does not exist")
+
+                        iam_context = get_iam_context(request, task)
+
+                        permissions.append(
+                            TaskPermission.create_scope_view(request, task, iam_context=iam_context)
+                        )
+
+                    permissions.append(
+                        cls.create_base_perm(
+                            request,
+                            view,
+                            scope,
+                            iam_context,
+                            obj,
+                            task_id=task_id,
+                        )
+                    )
 
-                    permissions.append(cls.create_base_perm(request, view, scope, iam_context, obj))
                 else:
                     permissions.append(cls.create_base_perm(request, view, scope, iam_context, obj))
 
@@ -91,15 +123,28 @@ def get_scopes(request, view, obj):
     def get_resource(self):
         data = None
 
-        if self.obj:
-            task = self.obj.get_task()
-            if task.project:
-                organization = task.project.organization
+        if self.obj or self.scope == self.Scopes.CREATE:
+            task: Optional[Task] = None
+            project: Optional[Project] = None
+            obj_id: Optional[int] = None
+
+            if self.obj:
+                obj_id = self.obj.id
+                task = self.obj.get_task()
+            elif self.scope == self.Scopes.CREATE and self.task_id:
+                try:
+                    task = Task.objects.get(id=self.task_id)
+                except Task.DoesNotExist:
+                    raise ValidationError("The specified task does not exist")
+
+            if task and task.project:
+                project = task.project
+                organization = project.organization
             else:
-                organization = task.organization
+                organization = getattr(task, "organization", None)
 
             data = {
-                "id": self.obj.id,
+                "id": obj_id,
                 "organization": {"id": getattr(organization, "id", None)},
                 "task": (
                     {
@@ -111,15 +156,15 @@ def get_resource(self):
                 ),
                 "project": (
                     {
-                        "owner": {"id": getattr(task.project.owner, "id", None)},
-                        "assignee": {"id": getattr(task.project.assignee, "id", None)},
+                        "owner": {"id": getattr(project.owner, "id", None)},
+                        "assignee": {"id": getattr(project.assignee, "id", None)},
                     }
-                    if task.project
+                    if project
                     else None
                 ),
             }
         elif self.scope == self.Scopes.VIEW_STATUS:
-            data = {"owner": self.job_owner_id}
+            data = {"owner": {"id": self.job_owner_id}}
 
         return data
 

cvat/apps/quality_control/quality_reports.py

@@ -11,17 +11,21 @@
 from datetime import timedelta
 from functools import cached_property, partial
 from typing import Any, Callable, Dict, Hashable, List, Optional, Sequence, Tuple, Union, cast
-from uuid import uuid4
 
 import datumaro as dm
 import datumaro.util.mask_tools
 import django_rq
 import numpy as np
+import rq
 from attrs import asdict, define, fields_dict
 from datumaro.util import dump_json, parse_json
 from django.conf import settings
 from django.db import transaction
 from django.utils import timezone
+from django_rq.queues import DjangoRQ as RqQueue
+from rest_framework.request import Request
+from rq.job import Job as RqJob
+from rq_scheduler import Scheduler as RqScheduler
 from scipy.optimize import linear_sum_assignment
 
 from cvat.apps.dataset_manager.bindings import (
@@ -48,6 +52,7 @@
     User,
     ValidationMode,
 )
+from cvat.apps.engine.utils import define_dependent_job, get_rq_job_meta, get_rq_lock_by_user
 from cvat.apps.profiler import silk_profile
 from cvat.apps.quality_control import models
 from cvat.apps.quality_control.models import (
@@ -2115,25 +2120,30 @@ def generate_report(self) -> ComparisonReport:
 
 
 class QualityReportUpdateManager:
-    _QUEUE_JOB_PREFIX = "update-quality-metrics-task-"
+    _QUEUE_AUTOUPDATE_JOB_PREFIX = "update-quality-metrics-"
+    _QUEUE_CUSTOM_JOB_PREFIX = "quality-check-"
     _RQ_CUSTOM_QUALITY_CHECK_JOB_TYPE = "custom_quality_check"
     _JOB_RESULT_TTL = 120
 
     @classmethod
     def _get_quality_check_job_delay(cls) -> timedelta:
         return timedelta(seconds=settings.QUALITY_CHECK_JOB_DELAY)
 
-    def _get_scheduler(self):
+    def _get_scheduler(self) -> RqScheduler:
         return django_rq.get_scheduler(settings.CVAT_QUEUES.QUALITY_REPORTS.value)
 
-    def _get_queue(self):
+    def _get_queue(self) -> RqQueue:
         return django_rq.get_queue(settings.CVAT_QUEUES.QUALITY_REPORTS.value)
 
     def _make_queue_job_id_base(self, task: Task) -> str:
-        return f"{self._QUEUE_JOB_PREFIX}{task.id}"
+        return f"{self._QUEUE_AUTOUPDATE_JOB_PREFIX}task-{task.id}"
 
-    def _make_custom_quality_check_job_id(self) -> str:
-        return uuid4().hex
+    def _make_custom_quality_check_job_id(self, task_id: int, user_id: int) -> str:
+        # FUTURE-TODO: it looks like job ID template should not include user_id because:
+        # 1. There is no need to compute quality reports several times for different users
+        # 2. Each user (not only rq job owner) that has permission to access a task should
+        # be able to check the status of the computation process
+        return f"{self._QUEUE_CUSTOM_JOB_PREFIX}task-{task_id}-user-{user_id}"
 
     @classmethod
     def _get_last_report_time(cls, task: Task) -> Optional[timezone.datetime]:
@@ -2186,28 +2196,52 @@ def schedule_quality_autoupdate_job(self, task: Task):
             task_id=task.id,
         )
 
-    def schedule_quality_check_job(self, task: Task, *, user_id: int) -> str:
+    class JobAlreadyExists(QualityReportsNotAvailable):
+        def __str__(self):
+            return "Quality computation job for this task already enqueued"
+
+    def schedule_custom_quality_check_job(
+        self, request: Request, task: Task, *, user_id: int
+    ) -> str:
         """
         Schedules a quality report computation job, supposed for updates by a request.
         """
 
         self._check_quality_reporting_available(task)
 
-        rq_id = self._make_custom_quality_check_job_id()
-
         queue = self._get_queue()
-        queue.enqueue(
-            self._check_task_quality,
-            task_id=task.id,
-            job_id=rq_id,
-            meta={"user_id": user_id, "job_type": self._RQ_CUSTOM_QUALITY_CHECK_JOB_TYPE},
-            result_ttl=self._JOB_RESULT_TTL,
-            failure_ttl=self._JOB_RESULT_TTL,
-        )
+
+        with get_rq_lock_by_user(queue, user_id=user_id):
+            rq_id = self._make_custom_quality_check_job_id(task_id=task.id, user_id=user_id)
+            rq_job = queue.fetch_job(rq_id)
+            if rq_job:
+                if rq_job.get_status(refresh=False) in (
+                    rq.job.JobStatus.QUEUED,
+                    rq.job.JobStatus.STARTED,
+                    rq.job.JobStatus.SCHEDULED,
+                    rq.job.JobStatus.DEFERRED,
+                ):
+                    raise self.JobAlreadyExists()
+
+                rq_job.delete()
+
+            dependency = define_dependent_job(
+                queue, user_id=user_id, rq_id=rq_id, should_be_dependent=True
+            )
+
+            queue.enqueue(
+                self._check_task_quality,
+                task_id=task.id,
+                job_id=rq_id,
+                meta=get_rq_job_meta(request=request, db_obj=task),
+                result_ttl=self._JOB_RESULT_TTL,
+                failure_ttl=self._JOB_RESULT_TTL,
+                depends_on=dependency,
+            )
 
         return rq_id
 
-    def get_quality_check_job(self, rq_id: str):
+    def get_quality_check_job(self, rq_id: str) -> Optional[RqJob]:
         queue = self._get_queue()
         rq_job = queue.fetch_job(rq_id)
 
@@ -2216,8 +2250,8 @@ def get_quality_check_job(self, rq_id: str):
 
         return rq_job
 
-    def is_custom_quality_check_job(self, rq_job) -> bool:
-        return rq_job.meta.get("job_type") == self._RQ_CUSTOM_QUALITY_CHECK_JOB_TYPE
+    def is_custom_quality_check_job(self, rq_job: RqJob) -> bool:
+        return isinstance(rq_job.id, str) and rq_job.id.startswith(self._QUEUE_CUSTOM_JOB_PREFIX)
 
     @classmethod
     @silk_profile()

cvat/apps/quality_control/rules/quality_reports.rego

@@ -4,9 +4,10 @@ import rego.v1
 
 import data.utils
 import data.organizations
+import data.quality_utils
 
 # input: {
-#     "scope": <"view"|"list"|"create"|"view:status"> or null,
+#     "scope": <"create"|"view"|"view:status"|"list"> or null,
 #     "auth": {
 #         "user": {
 #             "id": <num>,
@@ -23,7 +24,7 @@ import data.organizations
 #         } or null,
 #     },
 #     "resource": {
-#         "id": <num>,
+#         "id": <num> or null,
 #         "owner": { "id": <num> },
 #         "organization": { "id": <num> } or null,
 #         "task": {
@@ -41,6 +42,8 @@ import data.organizations
 #     }
 # }
 
+
+
 default allow := false
 
 allow if {
@@ -57,6 +60,33 @@ allow if {
     organizations.is_member
 }
 
+allow if {
+    input.scope == utils.VIEW_STATUS
+    utils.is_resource_owner
+}
+
+allow if {
+    input.scope in {utils.CREATE, utils.VIEW}
+    utils.is_sandbox
+    quality_utils.is_task_staff(input.resource.task, input.resource.project, input.auth)
+    utils.has_perm(utils.WORKER)
+}
+
+allow if {
+    input.scope in {utils.CREATE, utils.VIEW}
+    input.auth.organization.id == input.resource.organization.id
+    utils.has_perm(utils.USER)
+    organizations.has_perm(organizations.MAINTAINER)
+}
+
+allow if {
+    input.scope in {utils.CREATE, utils.VIEW}
+    quality_utils.is_task_staff(input.resource.task, input.resource.project, input.auth)
+    input.auth.organization.id == input.resource.organization.id
+    utils.has_perm(utils.WORKER)
+    organizations.has_perm(organizations.WORKER)
+}
+
 filter := [] if { # Django Q object to filter list of entries
     utils.is_admin
     utils.is_sandbox

cvat/apps/quality_control/rules/quality_settings.rego

@@ -6,7 +6,7 @@ import data.utils
 import data.organizations
 
 # input: {
-#     "scope": <"view"> or null,
+#     "scope": <"list"> or null,
 #     "auth": {
 #         "user": {
 #             "id": <num>,

cvat/apps/quality_control/rules/quality_utils.rego

@@ -0,0 +1,40 @@
+package quality_utils
+
+import rego.v1
+
+
+is_task_owner(task_data, auth_data) if {
+    task_data.owner.id == auth_data.user.id
+}
+
+is_task_assignee(task_data, auth_data) if {
+    task_data.assignee.id == auth_data.user.id
+}
+
+is_project_owner(project_data, auth_data) if {
+    project_data.owner.id == auth_data.user.id
+}
+
+is_project_assignee(project_data, auth_data) if {
+    project_data.assignee.id == auth_data.user.id
+}
+
+is_project_staff(project_data, auth_data) if {
+    is_project_owner(project_data, auth_data)
+}
+
+is_project_staff(project_data, auth_data) if {
+    is_project_assignee(project_data, auth_data)
+}
+
+is_task_staff(task_data, project_data, auth_data) if {
+    is_project_staff(project_data, auth_data)
+}
+
+is_task_staff(task_data, project_data, auth_data) if {
+    is_task_owner(task_data, auth_data)
+}
+
+is_task_staff(task_data, project_data, auth_data) if {
+    is_task_assignee(task_data, auth_data)
+}