Added tests for preview endpoints (#5529)

cvat-sdk/cvat_sdk/core/proxies/projects.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import io
 import json
 from pathlib import Path
 from typing import TYPE_CHECKING, List, Optional
@@ -125,6 +126,12 @@ class Project(
     def get_tasks(self) -> List[Task]:
         return [Task(self._client, m) for m in self.api.list_tasks(id=self.id)[0].results]
 
+    def get_preview(
+        self,
+    ) -> io.RawIOBase:
+        (_, response) = self.api.retrieve_preview(self.id)
+        return io.BytesIO(response.data)
+
 
 class ProjectsRepo(
     _ProjectRepoBase,

tests/python/rest_api/test_cloud_storages.py

@@ -3,10 +3,12 @@
 #
 # SPDX-License-Identifier: MIT
 
+import io
 from http import HTTPStatus
 
 import pytest
 from deepdiff import DeepDiff
+from PIL import Image
 
 from shared.utils.config import get_method, patch_method, post_method
 
@@ -283,3 +285,80 @@ class TestPatchCloudStorage:
             self._test_can_update(username, storage_id, self._PRIVATE_BUCKET_SPEC, org_id=org_id)
         else:
             self._test_cannot_update(username, storage_id, self._PRIVATE_BUCKET_SPEC, org_id=org_id)
+
+
+@pytest.mark.usefixtures("restore_db_per_class")
+class TestGetCloudStoragePreview:
+    def _test_can_see(self, user, storage_id, **kwargs):
+        response = get_method(user, f"cloudstorages/{storage_id}/preview", **kwargs)
+
+        assert response.status_code == HTTPStatus.OK
+        (width, height) = Image.open(io.BytesIO(response.content)).size
+        assert width > 0 and height > 0
+
+    def _test_cannot_see(self, user, storage_id, **kwargs):
+        response = get_method(user, f"cloudstorages/{storage_id}/preview", **kwargs)
+
+        assert response.status_code == HTTPStatus.FORBIDDEN
+
+    @pytest.mark.parametrize("storage_id", [1])
+    @pytest.mark.parametrize(
+        "group, is_owner, is_allow",
+        [
+            ("admin", False, True),
+            ("business", False, False),
+            ("user", True, True),
+        ],
+    )
+    def test_sandbox_user_get_cloud_storage_preview(
+        self, storage_id, group, is_owner, is_allow, users, cloud_storages
+    ):
+        org = ""
+        cloud_storage = cloud_storages[storage_id]
+        username = (
+            cloud_storage["owner"]["username"]
+            if is_owner
+            else next(
+                (
+                    u
+                    for u in users
+                    if group in u["groups"] and u["id"] != cloud_storage["owner"]["id"]
+                )
+            )["username"]
+        )
+
+        if is_allow:
+            self._test_can_see(username, storage_id, org=org)
+        else:
+            self._test_cannot_see(username, storage_id, org=org)
+
+    @pytest.mark.parametrize("org_id", [2])
+    @pytest.mark.parametrize("storage_id", [2])
+    @pytest.mark.parametrize(
+        "role, is_owner, is_allow",
+        [
+            ("worker", True, True),
+            ("supervisor", False, True),
+            ("worker", False, False),
+        ],
+    )
+    def test_org_user_get_cloud_storage_preview(
+        self, org_id, storage_id, role, is_owner, is_allow, find_users, cloud_storages
+    ):
+        cloud_storage = cloud_storages[storage_id]
+        username = (
+            cloud_storage["owner"]["username"]
+            if is_owner
+            else next(
+                (
+                    u
+                    for u in find_users(role=role, org=org_id)
+                    if u["id"] != cloud_storage["owner"]["id"]
+                )
+            )["username"]
+        )
+
+        if is_allow:
+            self._test_can_see(username, storage_id, org_id=org_id)
+        else:
+            self._test_cannot_see(username, storage_id, org_id=org_id)

tests/python/rest_api/test_jobs.py

@@ -14,6 +14,7 @@ from typing import List
 import pytest
 from cvat_sdk.core.helpers import get_paginated_collection
 from deepdiff import DeepDiff
+from PIL import Image
 
 from shared.utils.config import make_api_client
 
@@ -605,3 +606,49 @@ class TestJobDataset:
             )  # images + annotation file
             content = zip_file.read(anno_file_name)
         check_func(content, values_to_be_checked)
+
+
+@pytest.mark.usefixtures("restore_db_per_class")
+class TestGetJobPreview:
+    def _test_get_job_preview_200(self, username, jid, **kwargs):
+        with make_api_client(username) as client:
+            (_, response) = client.jobs_api.retrieve_preview(jid, **kwargs)
+
+            assert response.status == HTTPStatus.OK
+            (width, height) = Image.open(BytesIO(response.data)).size
+            assert width > 0 and height > 0
+
+    def _test_get_job_preview_403(self, username, jid, **kwargs):
+        with make_api_client(username) as client:
+            (_, response) = client.jobs_api.retrieve(
+                jid, **kwargs, _check_status=False, _parse_response=False
+            )
+            assert response.status == HTTPStatus.FORBIDDEN
+
+    @pytest.mark.parametrize("org", [None, "", 1, 2])
+    def test_admin_get_job_preview(self, jobs, tasks, org):
+        jobs, kwargs = filter_jobs(jobs, tasks, org)
+
+        # keep only the reasonable amount of jobs
+        for job in jobs[:8]:
+            self._test_get_job_preview_200("admin2", job["id"], **kwargs)
+
+    @pytest.mark.parametrize("org_id", ["", None, 1, 2])
+    @pytest.mark.parametrize("groups", [["business"], ["user"], ["worker"], []])
+    def test_non_admin_get_job_preview(
+        self, org_id, groups, users, jobs, tasks, projects, org_staff
+    ):
+        # keep the reasonable amount of users and jobs
+        users = [u for u in users if u["groups"] == groups][:4]
+        jobs, kwargs = filter_jobs(jobs, tasks, org_id)
+        org_staff = org_staff(org_id)
+
+        for job in jobs[:8]:
+            job_staff = get_job_staff(job, tasks, projects)
+
+            # check if the specific user in job_staff to see the job preview
+            for user in users:
+                if user["id"] in job_staff | org_staff:
+                    self._test_get_job_preview_200(user["username"], job["id"], **kwargs)
+                else:
+                    self._test_get_job_preview_403(user["username"], job["id"], **kwargs)

tests/python/rest_api/test_projects.py

@@ -17,6 +17,7 @@ from typing import Dict, Optional
 import pytest
 from cvat_sdk.api_client import ApiClient, Configuration, models
 from deepdiff import DeepDiff
+from PIL import Image
 
 from shared.utils.config import BASE_URL, USER_PASS, get_method, make_api_client, patch_method
 
@@ -735,3 +736,124 @@ class TestPatchProjectLabel:
         )
         assert response.status_code == HTTPStatus.OK
         assert len(response.json()["labels"]) == len(project["labels"]) + 1
+
+
+@pytest.mark.usefixtures("restore_db_per_class")
+class TestGetProjectPreview:
+    def _test_response_200(self, username, project_id, **kwargs):
+        with make_api_client(username) as api_client:
+            (_, response) = api_client.projects_api.retrieve_preview(project_id, **kwargs)
+
+            assert response.status == HTTPStatus.OK
+            (width, height) = Image.open(BytesIO(response.data)).size
+            assert width > 0 and height > 0
+
+    def _test_response_403(self, username, project_id):
+        with make_api_client(username) as api_client:
+            (_, response) = api_client.projects_api.retrieve_preview(
+                project_id, _parse_response=False, _check_status=False
+            )
+            assert response.status == HTTPStatus.FORBIDDEN
+
+    def _test_response_404(self, username, project_id):
+        with make_api_client(username) as api_client:
+            (_, response) = api_client.projects_api.retrieve_preview(
+                project_id, _parse_response=False, _check_status=False
+            )
+            assert response.status == HTTPStatus.NOT_FOUND
+
+    # Admin can see any project preview even he has no ownerships for this project.
+    def test_project_preview_admin_accessibility(
+        self, projects, find_users, is_project_staff, org_staff
+    ):
+        users = find_users(privilege="admin")
+
+        user, project = next(
+            (user, project)
+            for user, project in product(users, projects)
+            if not is_project_staff(user["id"], project["organization"])
+            and user["id"] not in org_staff(project["organization"])
+            and project["tasks"]
+        )
+        self._test_response_200(user["username"], project["id"])
+
+    # Project owner or project assignee can see project preview.
+    def test_project_preview_owner_accessibility(self, projects):
+        for p in projects:
+            if not p["tasks"]:
+                continue
+            if p["owner"] is not None:
+                project_with_owner = p
+            if p["assignee"] is not None:
+                project_with_assignee = p
+
+        assert project_with_owner is not None
+        assert project_with_assignee is not None
+
+        self._test_response_200(project_with_owner["owner"]["username"], project_with_owner["id"])
+        self._test_response_200(
+            project_with_assignee["assignee"]["username"], project_with_assignee["id"]
+        )
+
+    def test_project_preview_not_found(self, projects):
+        for p in projects:
+            if p["tasks"]:
+                continue
+            if p["owner"] is not None:
+                project_with_owner = p
+            if p["assignee"] is not None:
+                project_with_assignee = p
+
+        assert project_with_owner is not None
+        assert project_with_assignee is not None
+
+        self._test_response_404(project_with_owner["owner"]["username"], project_with_owner["id"])
+        self._test_response_404(
+            project_with_assignee["assignee"]["username"], project_with_assignee["id"]
+        )
+
+    def test_user_cannot_see_project_preview(
+        self, projects, find_users, is_project_staff, org_staff
+    ):
+        users = find_users(exclude_privilege="admin")
+
+        user, project = next(
+            (user, project)
+            for user, project in product(users, projects)
+            if not is_project_staff(user["id"], project["organization"])
+            and user["id"] not in org_staff(project["organization"])
+        )
+        self._test_response_403(user["username"], project["id"])
+
+    @pytest.mark.parametrize("role", ("supervisor", "worker"))
+    def test_if_supervisor_or_worker_cannot_see_project_preview(
+        self, projects, is_project_staff, find_users, role
+    ):
+        user, pid = next(
+            (
+                (user, project["id"])
+                for user in find_users(role=role, exclude_privilege="admin")
+                for project in projects
+                if project["organization"] == user["org"]
+                and not is_project_staff(user["id"], project["id"])
+            )
+        )
+
+        self._test_response_403(user["username"], pid)
+
+    @pytest.mark.parametrize("role", ("maintainer", "owner"))
+    def test_if_maintainer_or_owner_can_see_project_preview(
+        self, find_users, projects, is_project_staff, role
+    ):
+        user, pid = next(
+            (
+                (user, project["id"])
+                for user in find_users(role=role, exclude_privilege="admin")
+                for project in projects
+                if project["organization"] == user["org"]
+                and not is_project_staff(user["id"], project["id"])
+                and project["tasks"]
+            )
+        )
+
+        self._test_response_200(user["username"], pid, org_id=user["org"])

tests/python/rest_api/test_tasks.py

@@ -3,6 +3,7 @@
 #
 # SPDX-License-Identifier: MIT
 
+import io
 import json
 import os.path as osp
 import subprocess
@@ -16,6 +17,7 @@ import pytest
 from cvat_sdk.api_client import apis, models
 from cvat_sdk.core.helpers import get_paginated_collection
 from deepdiff import DeepDiff
+from PIL import Image
 
 import shared.utils.s3 as s3
 from shared.utils.config import get_method, make_api_client, patch_method
@@ -819,3 +821,67 @@ class TestPostTaskData:
 
         status = self._test_cannot_create_task(self._USERNAME, task_spec, data_spec, org=org)
         assert mythical_file in status.message
+
+
+@pytest.mark.usefixtures("restore_db_per_class")
+class TestGetTaskPreview:
+    def _test_task_preview_200(self, username, task_id, **kwargs):
+        with make_api_client(username) as api_client:
+            (_, response) = api_client.tasks_api.retrieve_preview(task_id, **kwargs)
+
+            assert response.status == HTTPStatus.OK
+            (width, height) = Image.open(io.BytesIO(response.data)).size
+            assert width > 0 and height > 0
+
+    def _test_task_preview_403(self, username, task_id):
+        with make_api_client(username) as api_client:
+            (_, response) = api_client.tasks_api.retrieve_preview(
+                task_id, _parse_response=False, _check_status=False
+            )
+            assert response.status == HTTPStatus.FORBIDDEN
+
+    def _test_assigned_users_to_see_task_preview(self, tasks, users, is_task_staff, **kwargs):
+        for task in tasks:
+            staff_users = [user for user in users if is_task_staff(user["id"], task["id"])]
+            assert len(staff_users)
+
+            for user in staff_users:
+                self._test_task_preview_200(user["username"], task["id"], **kwargs)
+
+    def _test_assigned_users_cannot_see_task_preview(self, tasks, users, is_task_staff, **kwargs):
+        for task in tasks:
+            not_staff_users = [user for user in users if not is_task_staff(user["id"], task["id"])]
+            assert len(not_staff_users)
+
+            for user in not_staff_users:
+                self._test_task_preview_403(user["username"], task["id"], **kwargs)
+
+    @pytest.mark.parametrize("project_id, groups", [(1, "user")])
+    def test_task_assigned_to_see_task_preview(
+        self, project_id, groups, users, tasks, find_users, is_task_staff
+    ):
+        users = find_users(privilege=groups)
+        tasks = list(filter(lambda x: x["project_id"] == project_id and x["assignee"], tasks))
+        assert len(tasks)
+
+        self._test_assigned_users_to_see_task_preview(tasks, users, is_task_staff)
+
+    @pytest.mark.parametrize("org, project_id, role", [({"id": 2, "slug": "org2"}, 2, "worker")])
+    def test_org_task_assigneed_to_see_task_preview(
+        self, org, project_id, role, users, tasks, find_users, is_task_staff
+    ):
+        users = find_users(org=org["id"], role=role)
+        tasks = list(filter(lambda x: x["project_id"] == project_id and x["assignee"], tasks))
+        assert len(tasks)
+
+        self._test_assigned_users_to_see_task_preview(tasks, users, is_task_staff, org=org["slug"])
+
+    @pytest.mark.parametrize("project_id, groups", [(1, "user")])
+    def test_task_unassigned_cannot_see_task_preview(
+        self, project_id, groups, users, tasks, find_users, is_task_staff
+    ):
+        users = find_users(privilege=groups)
+        tasks = list(filter(lambda x: x["project_id"] == project_id and x["assignee"], tasks))
+        assert len(tasks)
+
+        self._test_assigned_users_cannot_see_task_preview(tasks, users, is_task_staff)

tests/python/sdk/test_jobs.py

@@ -118,15 +118,17 @@ class TestJobUsecases:
 
     def test_can_download_preview(self, fxt_new_task: Task):
         frame_encoded = fxt_new_task.get_jobs()[0].get_preview()
+        (width, height) = Image.open(frame_encoded).size
 
-        assert Image.open(frame_encoded).size != 0
+        assert width > 0 and height > 0
         assert self.stdout.getvalue() == ""
 
     @pytest.mark.parametrize("quality", ("compressed", "original"))
     def test_can_download_frame(self, fxt_new_task: Task, quality: str):
         frame_encoded = fxt_new_task.get_jobs()[0].get_frame(0, quality=quality)
+        (width, height) = Image.open(frame_encoded).size
 
-        assert Image.open(frame_encoded).size != 0
+        assert width > 0 and height > 0
         assert self.stdout.getvalue() == ""
 
     @pytest.mark.parametrize("quality", ("compressed", "original"))

tests/python/sdk/test_projects.py

@@ -13,6 +13,7 @@ from cvat_sdk.api_client import exceptions
 from cvat_sdk.core.proxies.projects import Project
 from cvat_sdk.core.proxies.tasks import ResourceType, Task
 from cvat_sdk.core.utils import filter_dict
+from PIL import Image
 
 from .util import make_pbar
 
@@ -187,3 +188,10 @@ class TestProjectUsecases:
         assert restored_project.get_tasks()[0].size == 1
         assert "100%" in pbar_out.getvalue().strip("\r").split("\r")[-1]
         assert self.stdout.getvalue() == ""
+
+    def test_can_download_preview(self, fxt_project_with_shapes: Project):
+        frame_encoded = fxt_project_with_shapes.get_preview()
+        (width, height) = Image.open(frame_encoded).size
+
+        assert width > 0 and height > 0
+        assert self.stdout.getvalue() == ""

tests/python/sdk/test_tasks.py

@@ -279,15 +279,17 @@ class TestTaskUsecases:
 
     def test_can_download_preview(self, fxt_new_task: Task):
         frame_encoded = fxt_new_task.get_preview()
+        (width, height) = Image.open(frame_encoded).size
 
-        assert Image.open(frame_encoded).size != 0
+        assert width > 0 and height > 0
         assert self.stdout.getvalue() == ""
 
     @pytest.mark.parametrize("quality", ("compressed", "original"))
     def test_can_download_frame(self, fxt_new_task: Task, quality: str):
         frame_encoded = fxt_new_task.get_frame(0, quality=quality)
+        (width, height) = Image.open(frame_encoded).size
 
-        assert Image.open(frame_encoded).size != 0
+        assert width > 0 and height > 0
         assert self.stdout.getvalue() == ""
 
     @pytest.mark.parametrize("quality", ("compressed", "original"))