From 52e3f3c28aa80cfc4b3646b8c25d50dd8172c15c Mon Sep 17 00:00:00 2001 From: Andrey Zhavoronkov Date: Thu, 29 Dec 2022 17:48:19 +0200 Subject: [PATCH] Added tests for preview endpoints (#5529) --- cvat-sdk/cvat_sdk/core/proxies/projects.py | 7 ++ tests/python/rest_api/test_cloud_storages.py | 79 ++++++++++++ tests/python/rest_api/test_jobs.py | 47 +++++++ tests/python/rest_api/test_projects.py | 122 +++++++++++++++++++ tests/python/rest_api/test_tasks.py | 66 ++++++++++ tests/python/sdk/test_jobs.py | 6 +- tests/python/sdk/test_projects.py | 8 ++ tests/python/sdk/test_tasks.py | 6 +- 8 files changed, 337 insertions(+), 4 deletions(-) diff --git a/cvat-sdk/cvat_sdk/core/proxies/projects.py b/cvat-sdk/cvat_sdk/core/proxies/projects.py index 05646ee8..0053c15b 100644 --- a/cvat-sdk/cvat_sdk/core/proxies/projects.py +++ b/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, diff --git a/tests/python/rest_api/test_cloud_storages.py b/tests/python/rest_api/test_cloud_storages.py index 5f304326..a392c2a5 100644 --- a/tests/python/rest_api/test_cloud_storages.py +++ b/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) diff --git a/tests/python/rest_api/test_jobs.py b/tests/python/rest_api/test_jobs.py index 0a8bb53c..31ec6104 100644 --- a/tests/python/rest_api/test_jobs.py +++ b/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) diff --git a/tests/python/rest_api/test_projects.py b/tests/python/rest_api/test_projects.py index fdfbf1ac..b21ff6c1 100644 --- a/tests/python/rest_api/test_projects.py +++ b/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"]) diff --git a/tests/python/rest_api/test_tasks.py b/tests/python/rest_api/test_tasks.py index 08800aef..a02899b8 100644 --- a/tests/python/rest_api/test_tasks.py +++ b/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) diff --git a/tests/python/sdk/test_jobs.py b/tests/python/sdk/test_jobs.py index bf876807..6be70fa4 100644 --- a/tests/python/sdk/test_jobs.py +++ b/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")) diff --git a/tests/python/sdk/test_projects.py b/tests/python/sdk/test_projects.py index 06af0ead..14de38b4 100644 --- a/tests/python/sdk/test_projects.py +++ b/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() == "" diff --git a/tests/python/sdk/test_tasks.py b/tests/python/sdk/test_tasks.py index 4ca82a5b..4398d74d 100644 --- a/tests/python/sdk/test_tasks.py +++ b/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"))