# Copyright (C) 2022 Intel Corporation
# Copyright (C) 2022-2023 CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

import io
import json
import os
import os.path as osp
import zipfile
from copy import deepcopy
from functools import partial
from http import HTTPStatus
from itertools import chain, product
from math import ceil
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory
from time import sleep, time
from typing import List, Optional

import pytest
from cvat_sdk import Client, Config, exceptions
from cvat_sdk.api_client import models
from cvat_sdk.api_client.api_client import ApiClient, Endpoint
from cvat_sdk.core.helpers import get_paginated_collection
from cvat_sdk.core.proxies.tasks import ResourceType, Task
from cvat_sdk.core.uploading import Uploader
from deepdiff import DeepDiff
from PIL import Image

import shared.utils.s3 as s3
from shared.fixtures.init import docker_exec_cvat, kube_exec_cvat
from shared.utils.config import (
    BASE_URL,
    USER_PASS,
    get_method,
    make_api_client,
    patch_method,
    post_method,
)
from shared.utils.helpers import generate_image_files, generate_manifest

from .utils import (
    CollectionSimpleFilterTestBase,
    compare_annotations,
    create_task,
    export_dataset,
    wait_until_task_is_created,
)


def get_cloud_storage_content(
    api_version: int, username: str, cloud_storage_id: int, manifest: Optional[str] = None
):
    with make_api_client(username) as api_client:
        kwargs = {"manifest_path": manifest} if manifest else {}

        if api_version == 1:
            (data, _) = api_client.cloudstorages_api.retrieve_content(cloud_storage_id, **kwargs)
            return data
        (data, _) = api_client.cloudstorages_api.retrieve_content_v2(cloud_storage_id, **kwargs)
        return [f"{f['name']}{'/' if str(f['type']) == 'DIR' else ''}" for f in data["content"]]


@pytest.mark.usefixtures("restore_db_per_class")
class TestGetTasks:
    def _test_task_list_200(self, user, project_id, data, exclude_paths="", **kwargs):
        with make_api_client(user) as api_client:
            results = get_paginated_collection(
                api_client.tasks_api.list_endpoint,
                return_json=True,
                project_id=project_id,
                **kwargs,
            )
            assert DeepDiff(data, results, ignore_order=True, exclude_paths=exclude_paths) == {}

    def _test_users_to_see_task_list(
        self, project_id, tasks, users, is_staff, is_allow, is_project_staff, **kwargs
    ):
        if is_staff:
            users = [user for user in users if is_project_staff(user["id"], project_id)]
        else:
            users = [user for user in users if not is_project_staff(user["id"], project_id)]
        assert len(users)

        for user in users:
            if not is_allow:
                # Users outside project or org should not know if one exists.
                # Thus, no error should be produced on a list request.
                tasks = []

            self._test_task_list_200(user["username"], project_id, tasks, **kwargs)

    def _test_assigned_users_to_see_task_data(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:
                with make_api_client(user["username"]) as api_client:
                    (_, response) = api_client.tasks_api.list(**kwargs)
                    assert response.status == HTTPStatus.OK
                    response_data = json.loads(response.data)

                assert any(_task["id"] == task["id"] for _task in response_data["results"])

    @pytest.mark.parametrize("project_id", [1])
    @pytest.mark.parametrize(
        "groups, is_staff, is_allow",
        [
            ("admin", False, True),
            ("business", False, False),
        ],
    )
    def test_project_tasks_visibility(
        self, project_id, groups, users, tasks, is_staff, is_allow, find_users, is_project_staff
    ):
        users = find_users(privilege=groups)
        tasks = list(filter(lambda x: x["project_id"] == project_id, tasks))
        assert len(tasks)

        self._test_users_to_see_task_list(
            project_id, tasks, users, is_staff, is_allow, is_project_staff
        )

    @pytest.mark.parametrize("project_id, groups", [(1, "user")])
    def test_task_assigned_to_see_task(
        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_data(tasks, users, is_task_staff)

    @pytest.mark.parametrize("org, project_id", [({"id": 2, "slug": "org2"}, 2)])
    @pytest.mark.parametrize(
        "role, is_staff, is_allow",
        [
            ("maintainer", False, True),
            ("supervisor", False, False),
        ],
    )
    def test_org_project_tasks_visibility(
        self,
        org,
        project_id,
        role,
        is_staff,
        is_allow,
        tasks,
        is_task_staff,
        is_project_staff,
        find_users,
    ):
        users = find_users(org=org["id"], role=role)
        tasks = list(filter(lambda x: x["project_id"] == project_id, tasks))
        assert len(tasks)

        self._test_users_to_see_task_list(
            project_id, tasks, users, is_staff, is_allow, is_project_staff, org=org["slug"]
        )

    @pytest.mark.parametrize("org, project_id, role", [({"id": 2, "slug": "org2"}, 2, "worker")])
    def test_org_task_assigneed_to_see_task(
        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_data(tasks, users, is_task_staff, org=org["slug"])

    @pytest.mark.usefixtures("restore_db_per_function")
    def test_can_get_job_validation_summary(self, admin_user, tasks, jobs):
        task = next(t for t in tasks if t["jobs"]["count"] > 0 if t["jobs"]["validation"] == 0)
        job = next(j for j in jobs if j["task_id"] == task["id"])

        with make_api_client(admin_user) as api_client:
            api_client.jobs_api.partial_update(
                job["id"],
                patched_job_write_request=models.PatchedJobWriteRequest(stage="validation"),
            )

            (server_task, _) = api_client.tasks_api.retrieve(task["id"])

        assert server_task.jobs.validation == 1

    @pytest.mark.usefixtures("restore_db_per_function")
    def test_can_get_job_completed_summary(self, admin_user, tasks, jobs):
        task = next(t for t in tasks if t["jobs"]["count"] > 0 if t["jobs"]["completed"] == 0)
        job = next(j for j in jobs if j["task_id"] == task["id"])

        with make_api_client(admin_user) as api_client:
            api_client.jobs_api.partial_update(
                job["id"],
                patched_job_write_request=models.PatchedJobWriteRequest(
                    state="completed", stage="acceptance"
                ),
            )

            (server_task, _) = api_client.tasks_api.retrieve(task["id"])

        assert server_task.jobs.completed == 1

    def test_can_remove_owner_and_fetch_with_sdk(self, admin_user, tasks):
        # test for API schema regressions
        source_task = next(
            t for t in tasks if t.get("owner") and t["owner"]["username"] != admin_user
        ).copy()

        with make_api_client(admin_user) as api_client:
            api_client.users_api.destroy(source_task["owner"]["id"])

            (_, response) = api_client.tasks_api.retrieve(source_task["id"])
            fetched_task = json.loads(response.data)

        source_task["owner"] = None
        assert DeepDiff(source_task, fetched_task, ignore_order=True) == {}


class TestListTasksFilters(CollectionSimpleFilterTestBase):
    field_lookups = {
        "owner": ["owner", "username"],
        "assignee": ["assignee", "username"],
        "tracker_link": ["bug_tracker"],
    }

    @pytest.fixture(autouse=True)
    def setup(self, restore_db_per_class, admin_user, tasks):
        self.user = admin_user
        self.samples = tasks

    def _get_endpoint(self, api_client: ApiClient) -> Endpoint:
        return api_client.tasks_api.list_endpoint

    @pytest.mark.parametrize(
        "field",
        (
            "name",
            "owner",
            "status",
            "assignee",
            "subset",
            "mode",
            "dimension",
            "project_id",
            "tracker_link",
        ),
    )
    def test_can_use_simple_filter_for_object_list(self, field):
        return super().test_can_use_simple_filter_for_object_list(field)


@pytest.mark.usefixtures("restore_db_per_function")
class TestPostTasks:
    def _test_create_task_201(self, user, spec, **kwargs):
        with make_api_client(user) as api_client:
            (_, response) = api_client.tasks_api.create(spec, **kwargs)
            assert response.status == HTTPStatus.CREATED

        return response

    def _test_create_task_403(self, user, spec, **kwargs):
        with make_api_client(user) as api_client:
            (_, response) = api_client.tasks_api.create(
                spec, **kwargs, _parse_response=False, _check_status=False
            )
            assert response.status == HTTPStatus.FORBIDDEN

        return response

    def _test_users_to_create_task_in_project(
        self, project_id, users, is_staff, is_allow, is_project_staff, **kwargs
    ):
        if is_staff:
            users = [user for user in users if is_project_staff(user["id"], project_id)]
        else:
            users = [user for user in users if not is_project_staff(user["id"], project_id)]
        assert len(users)

        for user in users:
            username = user["username"]
            spec = {
                "name": f"test {username} to create a task within a project",
                "project_id": project_id,
            }

            if is_allow:
                self._test_create_task_201(username, spec, **kwargs)
            else:
                self._test_create_task_403(username, spec, **kwargs)

    @pytest.mark.parametrize("project_id", [1])
    @pytest.mark.parametrize(
        "groups, is_staff, is_allow",
        [
            ("admin", False, True),
            ("business", False, False),
            ("user", True, True),
        ],
    )
    def test_users_to_create_task_in_project(
        self, project_id, groups, is_staff, is_allow, is_project_staff, find_users
    ):
        users = find_users(privilege=groups)
        self._test_users_to_create_task_in_project(
            project_id, users, is_staff, is_allow, is_project_staff
        )

    @pytest.mark.parametrize("org, project_id", [({"id": 2, "slug": "org2"}, 2)])
    @pytest.mark.parametrize(
        "role, is_staff, is_allow",
        [
            ("worker", False, False),
        ],
    )
    def test_worker_cannot_create_task_in_project_without_ownership(
        self, org, project_id, role, is_staff, is_allow, is_project_staff, find_users
    ):
        users = find_users(org=org["id"], role=role)
        self._test_users_to_create_task_in_project(
            project_id, users, is_staff, is_allow, is_project_staff, org=org["slug"]
        )

    def test_create_response_matches_get(self, admin_user):
        username = admin_user

        spec = {"name": "test create task", "labels": [{"name": "a"}]}

        response = self._test_create_task_201(username, spec)
        task = json.loads(response.data)

        with make_api_client(username) as api_client:
            (_, response) = api_client.tasks_api.retrieve(task["id"])
            assert DeepDiff(task, json.loads(response.data), ignore_order=True) == {}

    def test_can_create_task_with_skeleton(self, admin_user):
        username = admin_user

        spec = {
            "name": f"test admin1 to create a task with skeleton",
            "labels": [
                {
                    "name": "s1",
                    "color": "#5c5eba",
                    "attributes": [
                        {
                            "name": "color",
                            "mutable": False,
                            "input_type": "select",
                            "default_value": "white",
                            "values": ["white", "black"],
                        }
                    ],
                    "type": "skeleton",
                    "sublabels": [
                        {
                            "name": "1",
                            "color": "#d53957",
                            "attributes": [
                                {
                                    "id": 23,
                                    "name": "attr",
                                    "mutable": False,
                                    "input_type": "select",
                                    "default_value": "val1",
                                    "values": ["val1", "val2"],
                                }
                            ],
                            "type": "points",
                        },
                        {"name": "2", "color": "#4925ec", "attributes": [], "type": "points"},
                        {"name": "3", "color": "#59a8fe", "attributes": [], "type": "points"},
                    ],
                    "svg": '<line x1="36.329429626464844" y1="45.98662185668945" x2="59.07190704345703" y2="23.076923370361328" '
                    'stroke="black" data-type="edge" data-node-from="2" stroke-width="0.5" data-node-to="3"></line>'
                    '<line x1="22.61705780029297" y1="25.75250816345215" x2="36.329429626464844" y2="45.98662185668945" '
                    'stroke="black" data-type="edge" data-node-from="1" stroke-width="0.5" data-node-to="2"></line>'
                    '<circle r="1.5" stroke="black" fill="#b3b3b3" cx="22.61705780029297" cy="25.75250816345215" '
                    'stroke-width="0.1" data-type="element node" data-element-id="1" data-node-id="1" data-label-name="1">'
                    '</circle><circle r="1.5" stroke="black" fill="#b3b3b3" cx="36.329429626464844" cy="45.98662185668945" '
                    'stroke-width="0.1" data-type="element node" data-element-id="2" data-node-id="2" data-label-name="2"></circle>'
                    '<circle r="1.5" stroke="black" fill="#b3b3b3" cx="59.07190704345703" cy="23.076923370361328" '
                    'stroke-width="0.1" data-type="element node" data-element-id="3" data-node-id="3" data-label-name="3"></circle>',
                }
            ],
        }

        self._test_create_task_201(username, spec)


@pytest.mark.usefixtures("restore_db_per_class")
class TestGetData:
    _USERNAME = "user1"

    @pytest.mark.parametrize(
        "content_type, task_id",
        [
            ("image/png", 8),
            ("image/png", 5),
            ("image/x.point-cloud-data", 6),
        ],
    )
    def test_frame_content_type(self, content_type, task_id):
        with make_api_client(self._USERNAME) as api_client:
            (_, response) = api_client.tasks_api.retrieve_data(
                task_id, type="frame", quality="original", number=0
            )
            assert response.status == HTTPStatus.OK
            assert response.headers["Content-Type"] == content_type


@pytest.mark.usefixtures("restore_db_per_function")
class TestPatchTaskAnnotations:
    def _test_check_response(self, is_allow, response, data=None):
        if is_allow:
            assert response.status == HTTPStatus.OK
            assert compare_annotations(data, json.loads(response.data)) == {}
        else:
            assert response.status == HTTPStatus.FORBIDDEN

    @pytest.fixture(scope="class")
    def request_data(self, annotations):
        def get_data(tid):
            data = deepcopy(annotations["task"][str(tid)])
            if data["shapes"][0]["type"] == "skeleton":
                data["shapes"][0]["elements"][0].update({"points": [2.0, 3.0, 4.0, 5.0]})
            else:
                data["shapes"][0].update({"points": [2.0, 3.0, 4.0, 5.0, 6.0, 7.0]})
            data["version"] += 1
            return data

        return get_data

    @pytest.mark.parametrize("org", [""])
    @pytest.mark.parametrize(
        "privilege, task_staff, is_allow",
        [
            ("admin", True, True),
            ("admin", False, True),
            ("business", True, True),
            ("business", False, False),
            ("worker", True, True),
            ("worker", False, False),
            ("user", True, True),
            ("user", False, False),
        ],
    )
    def test_user_update_task_annotations(
        self,
        org,
        privilege,
        task_staff,
        is_allow,
        find_task_staff_user,
        find_users,
        request_data,
        tasks_by_org,
        filter_tasks_with_shapes,
    ):
        users = find_users(privilege=privilege)
        tasks = tasks_by_org[org]
        filtered_tasks = filter_tasks_with_shapes(tasks)
        username, tid = find_task_staff_user(filtered_tasks, users, task_staff, [21])

        data = request_data(tid)
        with make_api_client(username) as api_client:
            (_, response) = api_client.tasks_api.partial_update_annotations(
                id=tid,
                action="update",
                patched_labeled_data_request=deepcopy(data),
                _parse_response=False,
                _check_status=False,
            )

        self._test_check_response(is_allow, response, data)

    @pytest.mark.parametrize("org", [2])
    @pytest.mark.parametrize(
        "role, task_staff, is_allow",
        [
            ("maintainer", False, True),
            ("owner", False, True),
            ("supervisor", False, False),
            ("worker", False, False),
            ("maintainer", True, True),
            ("owner", True, True),
            ("supervisor", True, True),
            ("worker", True, True),
        ],
    )
    def test_member_update_task_annotation(
        self,
        org,
        role,
        task_staff,
        is_allow,
        find_task_staff_user,
        find_users,
        tasks_by_org,
        request_data,
    ):
        users = find_users(role=role, org=org)
        tasks = tasks_by_org[org]
        username, tid = find_task_staff_user(tasks, users, task_staff)

        data = request_data(tid)
        with make_api_client(username) as api_client:
            (_, response) = api_client.tasks_api.partial_update_annotations(
                id=tid,
                action="update",
                patched_labeled_data_request=deepcopy(data),
                _parse_response=False,
                _check_status=False,
            )

        self._test_check_response(is_allow, response, data)


@pytest.mark.usefixtures("restore_db_per_class")
class TestGetTaskDataset:
    def _test_export_task(self, username, tid, **kwargs):
        with make_api_client(username) as api_client:
            return export_dataset(api_client.tasks_api.retrieve_dataset_endpoint, id=tid, **kwargs)

    def test_can_export_task_dataset(self, admin_user, tasks_with_shapes):
        task = tasks_with_shapes[0]
        response = self._test_export_task(admin_user, task["id"], format="CVAT for images 1.1")
        assert response.data

    @pytest.mark.parametrize("tid", [21])
    @pytest.mark.parametrize(
        "format_name", ["CVAT for images 1.1", "CVAT for video 1.1", "COCO Keypoints 1.0"]
    )
    def test_can_export_task_with_several_jobs(self, admin_user, tid, format_name):
        response = self._test_export_task(admin_user, tid, format=format_name)
        assert response.data

    @pytest.mark.parametrize("tid", [8])
    def test_can_export_task_to_coco_format(self, admin_user, tid):
        # these annotations contains incorrect frame numbers
        # in order to check that server handle such cases
        annotations = {
            "version": 0,
            "tags": [],
            "shapes": [],
            "tracks": [
                {
                    "label_id": 63,
                    "frame": 1,
                    "group": 0,
                    "source": "manual",
                    "shapes": [
                        {
                            "type": "skeleton",
                            "frame": 1,
                            "occluded": False,
                            "outside": False,
                            "z_order": 0,
                            "rotation": 0,
                            "points": [],
                            "attributes": [],
                        }
                    ],
                    "attributes": [],
                    "elements": [
                        {
                            "label_id": 64,
                            "frame": 0,
                            "group": 0,
                            "source": "manual",
                            "shapes": [
                                {
                                    "type": "points",
                                    "frame": 1,
                                    "occluded": False,
                                    "outside": True,
                                    "z_order": 0,
                                    "rotation": 0,
                                    "points": [74.14935096036425, 79.09960455479086],
                                    "attributes": [],
                                },
                                {
                                    "type": "points",
                                    "frame": 7,
                                    "occluded": False,
                                    "outside": False,
                                    "z_order": 0,
                                    "rotation": 0,
                                    "points": [74.14935096036425, 79.09960455479086],
                                    "attributes": [],
                                },
                            ],
                            "attributes": [],
                        },
                        {
                            "label_id": 65,
                            "frame": 0,
                            "group": 0,
                            "source": "manual",
                            "shapes": [
                                {
                                    "type": "points",
                                    "frame": 0,
                                    "occluded": False,
                                    "outside": False,
                                    "z_order": 0,
                                    "rotation": 0,
                                    "points": [285.07319976630424, 353.51583641966175],
                                    "attributes": [],
                                }
                            ],
                            "attributes": [],
                        },
                    ],
                }
            ],
        }
        response = patch_method(
            admin_user, f"tasks/{tid}/annotations", annotations, action="update"
        )
        assert response.status_code == HTTPStatus.OK

        # check that we can export task
        response = self._test_export_task(admin_user, tid, format="COCO Keypoints 1.0")
        assert response.status == HTTPStatus.OK

        # check that server saved track annotations correctly
        response = get_method(admin_user, f"tasks/{tid}/annotations")
        assert response.status_code == HTTPStatus.OK

        annotations = response.json()
        assert annotations["tracks"][0]["frame"] == 0
        assert annotations["tracks"][0]["shapes"][0]["frame"] == 0
        assert annotations["tracks"][0]["elements"][0]["shapes"][0]["frame"] == 0


@pytest.mark.usefixtures("restore_db_per_function")
@pytest.mark.usefixtures("restore_cvat_data")
class TestPostTaskData:
    _USERNAME = "admin1"

    def _test_cannot_create_task(self, username, spec, data, **kwargs):
        with make_api_client(username) as api_client:
            (task, response) = api_client.tasks_api.create(spec, **kwargs)
            assert response.status == HTTPStatus.CREATED

            (_, response) = api_client.tasks_api.create_data(
                task.id, data_request=deepcopy(data), _content_type="application/json", **kwargs
            )
            assert response.status == HTTPStatus.ACCEPTED

            status = wait_until_task_is_created(api_client.tasks_api, task.id)
            assert status.state.value == "Failed"

        return status

    def test_can_create_task_with_defined_start_and_stop_frames(self):
        task_spec = {
            "name": f"test {self._USERNAME} to create a task with defined start and stop frames",
            "labels": [
                {
                    "name": "car",
                    "color": "#ff00ff",
                    "attributes": [
                        {
                            "name": "a",
                            "mutable": True,
                            "input_type": "number",
                            "default_value": "5",
                            "values": ["4", "5", "6"],
                        }
                    ],
                }
            ],
        }

        task_data = {
            "image_quality": 75,
            "start_frame": 2,
            "stop_frame": 5,
            "client_files": generate_image_files(7),
        }

        task_id, _ = create_task(self._USERNAME, task_spec, task_data)

        # check task size
        with make_api_client(self._USERNAME) as api_client:
            (task, _) = api_client.tasks_api.retrieve(task_id)
            assert task.size == 4

    def test_can_create_task_with_exif_rotated_images(self):
        task_spec = {
            "name": f"test {self._USERNAME} to create a task with exif rotated images",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }

        image_files = ["images/exif_rotated/left.jpg", "images/exif_rotated/right.jpg"]
        task_data = {
            "server_files": image_files,
            "image_quality": 70,
            "segment_size": 500,
            "use_cache": True,
            "sorting_method": "natural",
        }

        task_id, _ = create_task(self._USERNAME, task_spec, task_data)

        # check that the frames have correct width and height
        with make_api_client(self._USERNAME) as api_client:
            _, response = api_client.tasks_api.retrieve_data(
                task_id, number=0, type="chunk", quality="original"
            )
            with zipfile.ZipFile(io.BytesIO(response.data)) as zip_file:
                for name in zip_file.namelist():
                    with zip_file.open(name) as zipped_img:
                        im = Image.open(zipped_img)
                        # original is 480x640 with 90/-90 degrees rotation
                        assert im.height == 640 and im.width == 480

    def test_can_create_task_with_sorting_method_natural(self):
        task_spec = {
            "name": f"test {self._USERNAME} to create a task with a custom sorting method",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }

        image_files = generate_image_files(15)

        task_data = {
            "client_files": image_files[5:] + image_files[:5],  # perturb the order
            "image_quality": 70,
            "sorting_method": "natural",
        }

        task_id, _ = create_task(self._USERNAME, task_spec, task_data)

        # check that the frames were sorted again
        with make_api_client(self._USERNAME) as api_client:
            data_meta, _ = api_client.tasks_api.retrieve_data_meta(task_id)

            # generate_image_files produces files that are already naturally sorted
            for image_file, frame in zip(image_files, data_meta.frames):
                assert image_file.name == frame.name

    @pytest.mark.parametrize("data_source", ["client_files", "server_files"])
    def test_can_create_task_with_sorting_method_predefined(self, data_source):
        task_spec = {
            "name": f"test {self._USERNAME} to create a task with a custom sorting method",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }

        if data_source == "client_files":
            image_files = generate_image_files(15)

            # shuffle to check for occasional sorting, e.g. in the DB
            image_files = image_files[7:] + image_files[5:7] + image_files[:5]
        elif data_source == "server_files":
            # Files from the test file share
            image_files = ["images/image_3.jpg", "images/image_1.jpg", "images/image_2.jpg"]
        else:
            assert False

        task_data = {
            data_source: image_files,
            "image_quality": 70,
            "sorting_method": "predefined",
        }

        (task_id, _) = create_task(self._USERNAME, task_spec, task_data)

        # check that the frames were sorted again
        with make_api_client(self._USERNAME) as api_client:
            (data_meta, _) = api_client.tasks_api.retrieve_data_meta(task_id)

            for image_file, frame in zip(image_files, data_meta.frames):
                if isinstance(image_file, str):
                    image_name = image_file
                else:
                    image_name = image_file.name

                assert image_name == frame.name

    def test_can_get_annotations_from_new_task_with_skeletons(self):
        spec = {
            "name": f"test admin1 to create a task with skeleton",
            "labels": [
                {
                    "name": "s1",
                    "color": "#5c5eba",
                    "attributes": [],
                    "type": "skeleton",
                    "sublabels": [
                        {"name": "1", "color": "#d12345", "attributes": [], "type": "points"},
                        {"name": "2", "color": "#350dea", "attributes": [], "type": "points"},
                    ],
                    "svg": '<line x1="19.464284896850586" y1="21.922269821166992" x2="54.08613586425781" y2="43.60293960571289" '
                    'stroke="black" data-type="edge" data-node-from="1" stroke-width="0.5" data-node-to="2"></line>'
                    '<circle r="1.5" stroke="black" fill="#b3b3b3" cx="19.464284896850586" cy="21.922269821166992" '
                    'stroke-width="0.1" data-type="element node" data-element-id="1" data-node-id="1" data-label-id="103"></circle>'
                    '<circle r="1.5" stroke="black" fill="#b3b3b3" cx="54.08613586425781" cy="43.60293960571289" '
                    'stroke-width="0.1" data-type="element node" data-element-id="2" data-node-id="2" data-label-id="104"></circle>',
                }
            ],
        }

        task_data = {
            "image_quality": 75,
            "client_files": generate_image_files(3),
        }

        task_id, _ = create_task(self._USERNAME, spec, task_data)

        response = get_method(self._USERNAME, "labels", task_id=f"{task_id}")
        label_ids = {}
        for root_label in response.json()["results"]:
            for label in [root_label] + root_label["sublabels"]:
                label_ids.setdefault(label["type"], []).append(label["id"])

        response = get_method(self._USERNAME, "jobs", task_id=f"{task_id}")
        job_id = response.json()["results"][0]["id"]
        patch_data = {
            "shapes": [
                {
                    "type": "skeleton",
                    "occluded": False,
                    "outside": False,
                    "z_order": 0,
                    "rotation": 0,
                    "points": [],
                    "frame": 0,
                    "label_id": label_ids["skeleton"][0],
                    "group": 0,
                    "source": "manual",
                    "attributes": [],
                    "elements": [
                        {
                            "type": "points",
                            "occluded": False,
                            "outside": False,
                            "z_order": 0,
                            "rotation": 0,
                            "points": [131.63947368421032, 165.0868421052637],
                            "frame": 0,
                            "label_id": label_ids["points"][0],
                            "group": 0,
                            "source": "manual",
                            "attributes": [],
                        },
                        {
                            "type": "points",
                            "occluded": False,
                            "outside": False,
                            "z_order": 0,
                            "rotation": 0,
                            "points": [354.98157894736823, 304.2710526315795],
                            "frame": 0,
                            "label_id": label_ids["points"][1],
                            "group": 0,
                            "source": "manual",
                            "attributes": [],
                        },
                    ],
                }
            ],
            "tracks": [
                {
                    "frame": 0,
                    "label_id": label_ids["skeleton"][0],
                    "group": 0,
                    "source": "manual",
                    "shapes": [
                        {
                            "type": "skeleton",
                            "occluded": False,
                            "outside": False,
                            "z_order": 0,
                            "rotation": 0,
                            "points": [],
                            "frame": 0,
                            "attributes": [],
                        }
                    ],
                    "attributes": [],
                    "elements": [
                        {
                            "frame": 0,
                            "label_id": label_ids["points"][0],
                            "group": 0,
                            "source": "manual",
                            "shapes": [
                                {
                                    "type": "points",
                                    "occluded": False,
                                    "outside": False,
                                    "z_order": 0,
                                    "rotation": 0,
                                    "points": [295.6394736842103, 472.5868421052637],
                                    "frame": 0,
                                    "attributes": [],
                                }
                            ],
                            "attributes": [],
                        },
                        {
                            "frame": 0,
                            "label_id": label_ids["points"][1],
                            "group": 0,
                            "source": "manual",
                            "shapes": [
                                {
                                    "type": "points",
                                    "occluded": False,
                                    "outside": False,
                                    "z_order": 0,
                                    "rotation": 0,
                                    "points": [619.3236842105262, 846.9815789473689],
                                    "frame": 0,
                                    "attributes": [],
                                }
                            ],
                            "attributes": [],
                        },
                    ],
                }
            ],
            "tags": [],
            "version": 0,
        }

        response = patch_method(
            self._USERNAME, f"jobs/{job_id}/annotations", patch_data, action="create"
        )
        response = get_method(self._USERNAME, f"jobs/{job_id}/annotations")
        assert response.status_code == HTTPStatus.OK

    @pytest.mark.with_external_services
    @pytest.mark.parametrize(
        "use_cache, cloud_storage_id, manifest, use_bucket_content, content_api_version, org",
        [
            (True, 1, "manifest.jsonl", False, None, ""),  # public bucket
            (True, 2, "sub/manifest.jsonl", True, 1, "org2"),  # private bucket
            (True, 2, "sub/manifest.jsonl", True, 2, "org2"),  # private bucket
            (True, 1, None, False, None, ""),
            (True, 2, None, True, 1, "org2"),
            (True, 2, None, True, 2, "org2"),
            (False, 1, None, False, None, ""),
            (False, 2, None, True, 1, "org2"),
            (False, 2, None, True, 2, "org2"),
        ],
    )
    def test_create_task_with_cloud_storage_files(
        self,
        use_cache: bool,
        cloud_storage_id: int,
        manifest: str,
        use_bucket_content: bool,
        content_api_version: Optional[int],
        org: str,
    ):
        if use_bucket_content:
            cloud_storage_content = get_cloud_storage_content(
                content_api_version, self._USERNAME, cloud_storage_id, manifest
            )
        else:
            cloud_storage_content = ["image_case_65_1.png", "image_case_65_2.png"]
        if manifest:
            cloud_storage_content.append(manifest)

        task_spec = {
            "name": f"Task with files from cloud storage {cloud_storage_id}",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }

        data_spec = {
            "image_quality": 75,
            "use_cache": use_cache,
            "cloud_storage_id": cloud_storage_id,
            "server_files": cloud_storage_content,
        }

        kwargs = {"org": org} if org else {}
        create_task(self._USERNAME, task_spec, data_spec, **kwargs)

    @pytest.mark.with_external_services
    @pytest.mark.parametrize("cloud_storage_id", [2])
    @pytest.mark.parametrize(
        "use_cache, use_manifest, server_files, server_files_exclude, task_size",
        [
            (True, False, ["test/"], None, 6),
            (True, False, ["test/sub_0/", "test/sub_1/"], None, 6),
            (True, False, ["test/"], ["test/sub_0/", "test/sub_1/img_1.jpeg"], 2),
            (True, True, ["test/"], None, 6),
            (True, True, ["test/sub_0/", "test/sub_1/"], None, 6),
            (True, True, ["test/"], ["test/sub_0/", "test/sub_1/img_1.jpeg"], 2),
            (False, False, ["test/"], None, 6),
            (False, False, ["test/sub_0/", "test/sub_1/"], None, 6),
            (False, False, ["test/"], ["test/sub_0/", "test/sub_1/img_1.jpeg"], 2),
        ],
    )
    @pytest.mark.parametrize("org", [""])
    def test_create_task_with_cloud_storage_directories_and_excluded_files(
        self,
        cloud_storage_id: int,
        use_cache: bool,
        use_manifest: bool,
        server_files: List[str],
        server_files_exclude: Optional[List[str]],
        task_size: int,
        org: str,
        cloud_storages,
        request,
    ):
        s3_client = s3.make_client()
        images = generate_image_files(3, prefixes=["img_"] * 3)

        cloud_storage = cloud_storages[cloud_storage_id]

        for image in images:
            for i in range(2):
                image.seek(0)
                s3_client.create_file(
                    data=image,
                    bucket=cloud_storage["resource"],
                    filename=f"test/sub_{i}/{image.name}",
                )
                request.addfinalizer(
                    partial(
                        s3_client.remove_file,
                        bucket=cloud_storage["resource"],
                        filename=f"test/sub_{i}/{image.name}",
                    )
                )

        if use_manifest:
            with TemporaryDirectory() as tmp_dir:
                manifest_root_path = f"{tmp_dir}/test/"
                for i in range(2):
                    path_with_sub_folders = f"{tmp_dir}/test/sub_{i}/"
                    os.makedirs(path_with_sub_folders)
                    for image in images:
                        with open(osp.join(path_with_sub_folders, image.name), "wb") as f:
                            f.write(image.getvalue())

                generate_manifest(manifest_root_path)

                with open(osp.join(manifest_root_path, "manifest.jsonl"), mode="rb") as m_file:
                    s3_client.create_file(
                        data=m_file.read(),
                        bucket=cloud_storage["resource"],
                        filename="test/manifest.jsonl",
                    )
                    request.addfinalizer(
                        partial(
                            s3_client.remove_file,
                            bucket=cloud_storage["resource"],
                            filename="test/manifest.jsonl",
                        )
                    )
                server_files.append("test/manifest.jsonl")

        task_spec = {
            "name": f"Task created from directories from cloud storage {cloud_storage_id}",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }

        data_spec = {
            "image_quality": 75,
            "use_cache": use_cache,
            "cloud_storage_id": cloud_storage_id,
            "server_files": server_files,
        }
        if server_files_exclude:
            data_spec["server_files_exclude"] = server_files_exclude

        task_id, _ = create_task(self._USERNAME, task_spec, data_spec, org=org)

        with make_api_client(self._USERNAME) as api_client:
            (task, response) = api_client.tasks_api.retrieve(task_id)
            assert response.status == HTTPStatus.OK
            assert task.size == task_size

    @pytest.mark.with_external_services
    @pytest.mark.parametrize(
        "storage_id, manifest",
        [
            (1, "manifest.jsonl"),  # public bucket
            (2, "sub/manifest.jsonl"),  # private bucket
        ],
    )
    @pytest.mark.parametrize(
        "spec, field",
        [
            ("spec", "source_storage"),
            ("spec", "target_storage"),
            ("data", "cloud_storage_id"),
        ],
    )
    def test_user_cannot_create_task_with_cloud_storage_without_access(
        self, storage_id, spec, field, manifest, regular_lonely_user
    ):
        user = regular_lonely_user

        task_spec = {
            "name": f"Task with files from foreign cloud storage {storage_id}",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }

        data_spec = {
            "image_quality": 75,
            "use_cache": True,
        }

        if spec == "spec":
            task_spec.update(
                {
                    field: {
                        "location": "cloud_storage",
                        "cloud_storage_id": storage_id,
                    }
                }
            )
            data_spec["server_files"] = ["images/image_1.jpg"]

        elif spec == "data":
            data_spec.update(
                {
                    field: storage_id,
                    "filename_pattern": "*",
                    "server_files": [manifest],
                }
            )
        else:
            assert False

        with pytest.raises(exceptions.ApiException) as capture:
            create_task(user, task_spec, data_spec)

        assert capture.value.status == HTTPStatus.FORBIDDEN

    @pytest.mark.with_external_services
    @pytest.mark.parametrize("cloud_storage_id", [1])
    @pytest.mark.parametrize(
        "manifest, filename_pattern, sub_dir, task_size",
        [
            ("manifest.jsonl", "*", True, 3),  # public bucket
            ("manifest.jsonl", "test/*", True, 3),
            ("manifest.jsonl", "test/sub*1.jpeg", True, 1),
            ("manifest.jsonl", "*image*.jpeg", True, 3),
            ("manifest.jsonl", "wrong_pattern", True, 0),
            ("abc_manifest.jsonl", "[a-c]*.jpeg", False, 2),
            ("abc_manifest.jsonl", "[d]*.jpeg", False, 1),
            ("abc_manifest.jsonl", "[e-z]*.jpeg", False, 0),
            (None, "*", True, 5),
            (None, "test/*", True, 3),
            (None, "test/sub*1.jpeg", True, 1),
            (None, "*image*.jpeg", True, 3),
            (None, "wrong_pattern", True, 0),
            (None, "[a-c]*.jpeg", False, 2),
            (None, "[d]*.jpeg", False, 1),
            (None, "[e-z]*.jpeg", False, 0),
        ],
    )
    def test_create_task_with_file_pattern(
        self,
        cloud_storage_id,
        manifest,
        filename_pattern,
        sub_dir,
        task_size,
        cloud_storages,
        request,
    ):
        # prepare dataset on the bucket
        prefixes = ("test_image_",) * 3 if sub_dir else ("a_", "b_", "d_")
        images = generate_image_files(3, prefixes=prefixes)
        s3_client = s3.make_client()

        cloud_storage = cloud_storages[cloud_storage_id]

        for image in images:
            s3_client.create_file(
                data=image,
                bucket=cloud_storage["resource"],
                filename=f"{'test/sub/' if sub_dir else ''}{image.name}",
            )
            request.addfinalizer(
                partial(
                    s3_client.remove_file,
                    bucket=cloud_storage["resource"],
                    filename=f"{'test/sub/' if sub_dir else ''}{image.name}",
                )
            )

        if manifest:
            with TemporaryDirectory() as tmp_dir:
                for image in images:
                    with open(osp.join(tmp_dir, image.name), "wb") as f:
                        f.write(image.getvalue())

                generate_manifest(tmp_dir)

                with open(osp.join(tmp_dir, "manifest.jsonl"), mode="rb") as m_file:
                    s3_client.create_file(
                        data=m_file.read(),
                        bucket=cloud_storage["resource"],
                        filename=f"test/sub/{manifest}" if sub_dir else manifest,
                    )
                    request.addfinalizer(
                        partial(
                            s3_client.remove_file,
                            bucket=cloud_storage["resource"],
                            filename=f"test/sub/{manifest}" if sub_dir else manifest,
                        )
                    )

        task_spec = {
            "name": f"Task with files from cloud storage {cloud_storage_id}",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }

        data_spec = {
            "image_quality": 75,
            "use_cache": True,
            "cloud_storage_id": cloud_storage_id,
            "filename_pattern": filename_pattern,
        }
        if manifest:
            data_spec["server_files"] = [f"test/sub/{manifest}" if sub_dir else manifest]

        if task_size:
            task_id, _ = create_task(self._USERNAME, task_spec, data_spec)

            with make_api_client(self._USERNAME) as api_client:
                (task, response) = api_client.tasks_api.retrieve(task_id)
                assert response.status == HTTPStatus.OK
                assert task.size == task_size
        else:
            status = self._test_cannot_create_task(self._USERNAME, task_spec, data_spec)
            assert "No media data found" in status.message

    def test_can_specify_file_job_mapping(self):
        task_spec = {
            "name": f"test file-job mapping",
            "labels": [{"name": "car"}],
        }

        files = generate_image_files(7)
        filenames = [osp.basename(f.name) for f in files]
        expected_segments = [
            filenames[0:1],
            filenames[1:5][::-1],  # a reversed fragment
            filenames[5:7],
        ]

        data_spec = {
            "image_quality": 75,
            "client_files": files,
            "job_file_mapping": expected_segments,
        }

        task_id, _ = create_task(
            self._USERNAME, task_spec, data_spec, content_type="application/json"
        )

        with make_api_client(self._USERNAME) as api_client:
            jobs: List[models.JobRead] = get_paginated_collection(
                api_client.jobs_api.list_endpoint, task_id=task_id, sort="id"
            )
            (task_meta, _) = api_client.tasks_api.retrieve_data_meta(id=task_id)

            assert [f.name for f in task_meta.frames] == list(
                chain.from_iterable(expected_segments)
            )

            start_frame = 0
            for i, job in enumerate(jobs):
                expected_size = len(expected_segments[i])
                stop_frame = start_frame + expected_size - 1
                assert job.start_frame == start_frame
                assert job.stop_frame == stop_frame

                start_frame = stop_frame + 1

    def test_cannot_create_task_with_same_labels(self):
        task_spec = {
            "name": "test cannot create task with same labels",
            "labels": [{"name": "l1"}, {"name": "l1"}],
        }
        response = post_method(self._USERNAME, "tasks", task_spec)
        assert response.status_code == HTTPStatus.BAD_REQUEST

        response = get_method(self._USERNAME, "tasks")
        assert response.status_code == HTTPStatus.OK

    def test_cannot_create_task_with_same_skeleton_sublabels(self):
        task_spec = {
            "name": "test cannot create task with same skeleton sublabels",
            "labels": [
                {"name": "s1", "type": "skeleton", "sublabels": [{"name": "1"}, {"name": "1"}]}
            ],
        }
        response = post_method(self._USERNAME, "tasks", task_spec)
        assert response.status_code == HTTPStatus.BAD_REQUEST

        response = get_method(self._USERNAME, "tasks")
        assert response.status_code == HTTPStatus.OK


@pytest.mark.usefixtures("restore_db_per_function")
class TestPatchTaskLabel:
    def _get_task_labels(self, pid, user, **kwargs) -> List[models.Label]:
        kwargs.setdefault("return_json", True)
        with make_api_client(user) as api_client:
            return get_paginated_collection(
                api_client.labels_api.list_endpoint, task_id=pid, **kwargs
            )

    def test_can_delete_label(self, tasks, labels, admin_user):
        task = [t for t in tasks if t["project_id"] is None and t["labels"]["count"] > 0][0]
        label = deepcopy([l for l in labels if l.get("task_id") == task["id"]][0])
        label_payload = {"id": label["id"], "deleted": True}

        response = patch_method(admin_user, f'tasks/{task["id"]}', {"labels": [label_payload]})
        assert response.status_code == HTTPStatus.OK, response.content
        assert response.json()["labels"]["count"] == task["labels"]["count"] - 1

    def test_can_delete_skeleton_label(self, tasks, labels, admin_user):
        task = next(
            t
            for t in tasks
            if any(
                label
                for label in labels
                if label.get("task_id") == t["id"]
                if label["type"] == "skeleton"
            )
        )
        task_labels = deepcopy([l for l in labels if l.get("task_id") == task["id"]])
        label = next(l for l in task_labels if l["type"] == "skeleton")
        task_labels.remove(label)
        label_payload = {"id": label["id"], "deleted": True}

        response = patch_method(admin_user, f'tasks/{task["id"]}', {"labels": [label_payload]})
        assert response.status_code == HTTPStatus.OK
        assert response.json()["labels"]["count"] == task["labels"]["count"] - 1

        resulting_labels = self._get_task_labels(task["id"], admin_user)
        assert DeepDiff(resulting_labels, task_labels, ignore_order=True) == {}

    def test_can_rename_label(self, tasks, labels, admin_user):
        task = [t for t in tasks if t["project_id"] is None and t["labels"]["count"] > 0][0]
        task_labels = deepcopy([l for l in labels if l.get("task_id") == task["id"]])
        task_labels[0].update({"name": "new name"})

        response = patch_method(admin_user, f'tasks/{task["id"]}', {"labels": [task_labels[0]]})
        assert response.status_code == HTTPStatus.OK

        resulting_labels = self._get_task_labels(task["id"], admin_user)
        assert DeepDiff(resulting_labels, task_labels, ignore_order=True) == {}

    def test_cannot_rename_label_to_duplicate_name(self, tasks, labels, admin_user):
        task = [t for t in tasks if t["project_id"] is None and t["labels"]["count"] > 1][0]
        task_labels = deepcopy([l for l in labels if l.get("task_id") == task["id"]])
        task_labels[0].update({"name": task_labels[1]["name"]})

        label_payload = {"id": task_labels[0]["id"], "name": task_labels[0]["name"]}

        response = patch_method(admin_user, f'tasks/{task["id"]}', {"labels": [label_payload]})
        assert response.status_code == HTTPStatus.BAD_REQUEST
        assert "All label names must be unique" in response.text

    def test_cannot_add_foreign_label(self, tasks, labels, admin_user):
        task = [t for t in tasks if t["project_id"] is None][0]
        new_label = deepcopy(
            [
                l
                for l in labels
                if l.get("task_id") != task["id"]
                if not l.get("project_id") or l.get("project_id") != task.get("project_id")
            ][0]
        )

        response = patch_method(admin_user, f'tasks/{task["id"]}', {"labels": [new_label]})
        assert response.status_code == HTTPStatus.NOT_FOUND
        assert f"Not found label with id #{new_label['id']} to change" in response.text

    def test_admin_can_add_label(self, tasks, admin_user):
        task = [t for t in tasks if t["project_id"] is None][0]
        new_label = {"name": "new name"}

        response = patch_method(admin_user, f'tasks/{task["id"]}', {"labels": [new_label]})
        assert response.status_code == HTTPStatus.OK
        assert response.json()["labels"]["count"] == task["labels"]["count"] + 1

    @pytest.mark.parametrize("role", ["maintainer", "owner"])
    def test_non_task_staff_privileged_org_members_can_add_label(
        self,
        find_users,
        tasks,
        is_task_staff,
        is_org_member,
        role,
    ):
        users = find_users(role=role, exclude_privilege="admin")

        user, task = next(
            (user, task)
            for user, task in product(users, tasks)
            if not is_task_staff(user["id"], task["id"])
            and task["organization"]
            and is_org_member(user["id"], task["organization"] and task["project_id"] is None)
        )

        new_label = {"name": "new name"}
        response = patch_method(
            user["username"],
            f'tasks/{task["id"]}',
            {"labels": [new_label]},
        )
        assert response.status_code == HTTPStatus.OK
        assert response.json()["labels"]["count"] == task["labels"]["count"] + 1

    @pytest.mark.parametrize("role", ["supervisor", "worker"])
    def test_non_task_staff_org_members_cannot_add_label(
        self,
        find_users,
        tasks,
        is_task_staff,
        is_org_member,
        role,
    ):
        users = find_users(role=role, exclude_privilege="admin")

        user, task = next(
            (user, task)
            for user, task in product(users, tasks)
            if not is_task_staff(user["id"], task["id"])
            and task["organization"]
            and is_org_member(user["id"], task["organization"])
        )

        new_label = {"name": "new name"}
        response = patch_method(
            user["username"],
            f'tasks/{task["id"]}',
            {"labels": [new_label]},
        )
        assert response.status_code == HTTPStatus.FORBIDDEN

    # TODO: add supervisor too, but this leads to a test-side problem with DB restoring
    @pytest.mark.parametrize("role", ["worker"])
    def test_task_staff_org_members_can_add_label(
        self, find_users, tasks, is_task_staff, is_org_member, labels, role
    ):
        users = find_users(role=role, exclude_privilege="admin")

        user, task = next(
            (user, task)
            for user, task in product(users, tasks)
            if is_task_staff(user["id"], task["id"])
            and task["organization"]
            and is_org_member(user["id"], task["organization"])
            and any(label.get("task_id") == task["id"] for label in labels)
        )

        new_label = {"name": "new name"}
        response = patch_method(
            user["username"],
            f'tasks/{task["id"]}',
            {"labels": [new_label]},
        )
        assert response.status_code == HTTPStatus.OK
        assert response.json()["labels"]["count"] == task["labels"]["count"] + 1

    def test_admin_can_add_skeleton(self, tasks, admin_user):
        task = [t for t in tasks if t["project_id"] is None][0]
        new_skeleton = {
            "name": "new skeleton",
            "type": "skeleton",
            "sublabels": [
                {
                    "name": "1",
                    "type": "points",
                }
            ],
            "svg": '<circle r="1.5" stroke="black" fill="#b3b3b3" cx="48.794559478759766" '
            'cy="36.98698806762695" stroke-width="0.1" data-type="element node" '
            'data-element-id="1" data-node-id="1" data-label-name="597501"></circle>',
        }

        response = patch_method(admin_user, f'tasks/{task["id"]}', {"labels": [new_skeleton]})
        assert response.status_code == HTTPStatus.OK
        assert response.json()["labels"]["count"] == task["labels"]["count"] + 1


@pytest.mark.usefixtures("restore_db_per_function")
@pytest.mark.usefixtures("restore_cvat_data")
class TestWorkWithTask:
    _USERNAME = "admin1"

    @pytest.mark.with_external_services
    @pytest.mark.parametrize(
        "cloud_storage_id, manifest",
        [(1, "manifest.jsonl")],  # public bucket
    )
    def test_work_with_task_containing_non_stable_cloud_storage_files(
        self, cloud_storage_id, manifest, cloud_storages, request
    ):
        image_name = "image_case_65_1.png"
        cloud_storage_content = [image_name, manifest]

        task_spec = {
            "name": f"Task with mythical file from cloud storage {cloud_storage_id}",
            "labels": [{"name": "car"}],
        }

        data_spec = {
            "image_quality": 75,
            "use_cache": True,
            "cloud_storage_id": cloud_storage_id,
            "server_files": cloud_storage_content,
        }

        task_id, _ = create_task(self._USERNAME, task_spec, data_spec)

        # save image from the "public" bucket and remove it temporary

        s3_client = s3.make_client()
        bucket_name = cloud_storages[cloud_storage_id]["resource"]

        image = s3_client.download_fileobj(bucket_name, image_name)
        s3_client.remove_file(bucket_name, image_name)
        request.addfinalizer(
            partial(s3_client.create_file, bucket=bucket_name, filename=image_name, data=image)
        )

        with make_api_client(self._USERNAME) as api_client:
            try:
                api_client.tasks_api.retrieve_data(
                    task_id, number=0, quality="original", type="frame"
                )
                raise AssertionError("Frame should not exist")
            except AssertionError:
                raise
            except Exception as ex:
                assert ex.status == HTTPStatus.NOT_FOUND
                assert image_name in ex.body


class TestTaskBackups:
    def _make_client(self) -> Client:
        return Client(BASE_URL, config=Config(status_check_period=0.01))

    @pytest.fixture(autouse=True)
    def setup(self, restore_db_per_function, restore_cvat_data, tmp_path: Path, admin_user: str):
        self.tmp_dir = tmp_path

        self.client = self._make_client()
        self.user = admin_user

        with self.client:
            self.client.login((self.user, USER_PASS))

    @pytest.mark.parametrize("mode", ["annotation", "interpolation"])
    def test_can_export_backup(self, tasks, mode):
        task_id = next(t for t in tasks if t["mode"] == mode)["id"]
        task = self.client.tasks.retrieve(task_id)

        filename = self.tmp_dir / f"task_{task.id}_backup.zip"
        task.download_backup(filename)

        assert filename.is_file()
        assert filename.stat().st_size > 0

    @pytest.mark.parametrize("mode", ["annotation", "interpolation"])
    def test_can_import_backup(self, tasks, mode):
        task_json = next(t for t in tasks if t["mode"] == mode)
        self._test_can_restore_backup_task(task_json["id"])

    @pytest.mark.parametrize("mode", ["annotation", "interpolation"])
    def test_can_import_backup_for_task_in_nondefault_state(self, tasks, mode):
        # Reproduces the problem with empty 'mode' in a restored task,
        # described in the reproduction steps https://github.com/opencv/cvat/issues/5668

        task_json = next(t for t in tasks if t["mode"] == mode and t["jobs"]["count"])

        task = self.client.tasks.retrieve(task_json["id"])
        jobs = task.get_jobs()
        for j in jobs:
            j.update({"stage": "validation"})

        self._test_can_restore_backup_task(task_json["id"])

    def _test_can_restore_backup_task(self, task_id: int):
        task = self.client.tasks.retrieve(task_id)
        (_, response) = self.client.api_client.tasks_api.retrieve(task_id)
        task_json = json.loads(response.data)

        filename = self.tmp_dir / f"task_{task.id}_backup.zip"
        task.download_backup(filename)

        restored_task = self.client.tasks.create_from_backup(filename)

        old_jobs = task.get_jobs()
        new_jobs = restored_task.get_jobs()
        assert len(old_jobs) == len(new_jobs)

        for old_job, new_job in zip(old_jobs, new_jobs):
            assert old_job.status == new_job.status
            assert old_job.start_frame == new_job.start_frame
            assert old_job.stop_frame == new_job.stop_frame

        (_, response) = self.client.api_client.tasks_api.retrieve(restored_task.id)
        restored_task_json = json.loads(response.data)

        assert restored_task_json["assignee"] is None
        assert restored_task_json["owner"]["username"] == self.user
        assert restored_task_json["id"] != task_json["id"]
        assert restored_task_json["data"] != task_json["data"]
        assert restored_task_json["organization"] is None
        assert restored_task_json["data_compressed_chunk_type"] in ["imageset", "video"]
        if task_json["jobs"]["count"] == 1:
            assert restored_task_json["overlap"] == 0
        else:
            assert restored_task_json["overlap"] == task_json["overlap"]
        assert restored_task_json["jobs"]["completed"] == 0
        assert restored_task_json["jobs"]["validation"] == 0
        assert restored_task_json["source_storage"] is None
        assert restored_task_json["target_storage"] is None
        assert restored_task_json["project_id"] is None

        assert (
            DeepDiff(
                task_json,
                restored_task_json,
                ignore_order=True,
                exclude_regex_paths=[
                    r"root\['id'\]",  # id, must be different
                    r"root\['created_date'\]",  # must be different
                    r"root\['updated_date'\]",  # must be different
                    r"root\['assignee'\]",  # id, depends on the situation
                    r"root\['owner'\]",  # id, depends on the situation
                    r"root\['data'\]",  # id, must be different
                    r"root\['organization'\]",  # depends on the task setup
                    r"root\['project_id'\]",  # should be dropped
                    r"root(\['.*'\])*\['url'\]",  # depends on the task id
                    r"root\['data_compressed_chunk_type'\]",  # depends on the server configuration
                    r"root\['source_storage'\]",  # should be dropped
                    r"root\['target_storage'\]",  # should be dropped
                    r"root\['jobs'\]\['completed'\]",  # job statuses should be renewed
                    r"root\['jobs'\]\['validation'\]",  # job statuses should be renewed
                    # depends on the actual job configuration,
                    # unlike to what is obtained from the regular task creation,
                    # where the requested number is recorded
                    r"root\['overlap'\]",
                ],
            )
            == {}
        )


@pytest.mark.usefixtures("restore_db_per_function")
class TestWorkWithGtJobs:
    def test_normal_and_gt_job_annotations_are_not_merged(
        self, tmp_path, admin_user, tasks, jobs, annotations
    ):
        gt_job = next(j for j in jobs if j["type"] == "ground_truth")
        task = tasks[gt_job["task_id"]]
        task_jobs = [j for j in jobs if j["task_id"] == task["id"]]

        gt_job_source_annotations = annotations["job"][str(gt_job["id"])]
        assert (
            gt_job_source_annotations["tags"]
            or gt_job_source_annotations["shapes"]
            or gt_job_source_annotations["tracks"]
        )

        with Client(BASE_URL) as client:
            client.config.status_check_period = 0.01
            client.login((admin_user, USER_PASS))

            for j in task_jobs:
                if j["type"] != "ground_truth":
                    client.jobs.retrieve(j["id"]).remove_annotations()

            task_obj = client.tasks.retrieve(task["id"])
            task_raw_annotations = task_obj.get_annotations()

            # It's quite hard to parse the dataset files, just import the data back instead
            dataset_format = "CVAT for images 1.1"

            dataset_file = tmp_path / "dataset.zip"
            task_obj.export_dataset(dataset_format, dataset_file, include_images=True)
            task_obj.import_annotations("CVAT 1.1", dataset_file)
            task_dataset_file_annotations = task_obj.get_annotations()

            annotations_file = tmp_path / "annotations.zip"
            task_obj.export_dataset(dataset_format, annotations_file, include_images=False)
            task_obj.import_annotations("CVAT 1.1", annotations_file)
            task_annotations_file_annotations = task_obj.get_annotations()

        for annotation_source in [
            task_raw_annotations,
            task_dataset_file_annotations,
            task_annotations_file_annotations,
        ]:
            assert not annotation_source.tags
            assert not annotation_source.shapes
            assert not annotation_source.tracks

    def test_can_backup_task_with_gt_jobs(self, tmp_path, admin_user, tasks, jobs, annotations):
        gt_job = next(
            j
            for j in jobs
            if j["type"] == "ground_truth" and tasks[j["task_id"]]["jobs"]["count"] == 2
        )
        task = tasks[gt_job["task_id"]]
        annotation_job = next(
            j for j in jobs if j["task_id"] == task["id"] and j["type"] == "annotation"
        )

        gt_job_source_annotations = annotations["job"][str(gt_job["id"])]
        assert (
            gt_job_source_annotations["tags"]
            or gt_job_source_annotations["shapes"]
            or gt_job_source_annotations["tracks"]
        )

        annotation_job_source_annotations = annotations["job"][str(annotation_job["id"])]

        with Client(BASE_URL) as client:
            client.config.status_check_period = 0.01
            client.login((admin_user, USER_PASS))

            backup_file: Path = tmp_path / "dataset.zip"
            client.tasks.retrieve(task["id"]).download_backup(backup_file)

            new_task = client.tasks.create_from_backup(backup_file)
            updated_job_annotations = {
                j.type: json.loads(j.api.retrieve_annotations(j.id)[1].data)
                for j in new_task.get_jobs()
            }

        for job_type, source_annotations in {
            gt_job["type"]: gt_job_source_annotations,
            annotation_job["type"]: annotation_job_source_annotations,
        }.items():
            assert (
                DeepDiff(
                    source_annotations,
                    updated_job_annotations[job_type],
                    ignore_order=True,
                    exclude_regex_paths=[
                        r"root(\['\w+'\]\[\d+\])+\['id'\]",
                        r"root(\['\w+'\]\[\d+\])+\['label_id'\]",
                        r"root(\['\w+'\]\[\d+\])+\['attributes'\]\[\d+\]\['spec_id'\]",
                    ],
                )
                == {}
            )


@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)

    @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)


class TestUnequalJobs:
    def _make_client(self) -> Client:
        return Client(BASE_URL, config=Config(status_check_period=0.01))

    @pytest.fixture(autouse=True)
    def setup(self, restore_db_per_function, tmp_path: Path, admin_user: str):
        self.tmp_dir = tmp_path

        self.client = self._make_client()
        self.user = admin_user

        with self.client:
            self.client.login((self.user, USER_PASS))

    @pytest.fixture
    def fxt_task_with_unequal_jobs(self):
        task_spec = {
            "name": f"test file-job mapping",
            "labels": [{"name": "car"}],
        }

        files = generate_image_files(7)
        filenames = [osp.basename(f.name) for f in files]
        for file_data in files:
            with open(self.tmp_dir / file_data.name, "wb") as f:
                f.write(file_data.getvalue())

        expected_segments = [
            filenames[0:1],
            filenames[1:5][::-1],  # a reversed fragment
            filenames[5:7],
        ]

        data_spec = {
            "job_file_mapping": expected_segments,
        }

        return self.client.tasks.create_from_data(
            spec=task_spec,
            resource_type=ResourceType.LOCAL,
            resources=[self.tmp_dir / fn for fn in filenames],
            data_params=data_spec,
        )

    def test_can_export(self, fxt_task_with_unequal_jobs: Task):
        task = fxt_task_with_unequal_jobs

        filename = self.tmp_dir / f"task_{task.id}_coco.zip"
        task.export_dataset("COCO 1.0", filename)

        assert filename.is_file()
        assert filename.stat().st_size > 0

    def test_can_import_annotations(self, fxt_task_with_unequal_jobs: Task):
        task = fxt_task_with_unequal_jobs

        format_name = "COCO 1.0"
        filename = self.tmp_dir / f"task_{task.id}_coco.zip"
        task.export_dataset(format_name, filename)

        task.import_annotations(format_name, filename)

    def test_can_dump_backup(self, fxt_task_with_unequal_jobs: Task):
        task = fxt_task_with_unequal_jobs

        filename = self.tmp_dir / f"task_{task.id}_backup.zip"
        task.download_backup(filename)

        assert filename.is_file()
        assert filename.stat().st_size > 0

    def test_can_import_backup(self, fxt_task_with_unequal_jobs: Task):
        task = fxt_task_with_unequal_jobs

        filename = self.tmp_dir / f"task_{task.id}_backup.zip"
        task.download_backup(filename)

        restored_task = self.client.tasks.create_from_backup(filename)

        old_jobs = task.get_jobs()
        new_jobs = restored_task.get_jobs()
        assert len(old_jobs) == len(new_jobs)

        for old_job, new_job in zip(old_jobs, new_jobs):
            assert old_job.start_frame == new_job.start_frame
            assert old_job.stop_frame == new_job.stop_frame


@pytest.mark.usefixtures("restore_db_per_function")
class TestPatchTask:
    @pytest.mark.parametrize("task_id, project_id, user", [(19, 12, "admin1")])
    def test_move_task_to_project_with_attributes(self, task_id, project_id, user):
        response = get_method(user, f"tasks/{task_id}/annotations")
        assert response.status_code == HTTPStatus.OK
        annotations = response.json()

        response = patch_method(user, f"tasks/{task_id}", {"project_id": project_id})
        assert response.status_code == HTTPStatus.OK

        response = get_method(user, f"tasks/{task_id}")
        assert response.status_code == HTTPStatus.OK
        assert response.json().get("project_id") == project_id

        response = get_method(user, f"tasks/{task_id}/annotations")
        assert response.status_code == HTTPStatus.OK
        assert (
            DeepDiff(
                annotations,
                response.json(),
                ignore_order=True,
                exclude_regex_paths=[
                    r"root\['\w+'\]\[\d+\]\['label_id'\]",
                    r"root\['\w+'\]\[\d+\]\['attributes'\]\[\d+\]\['spec_id'\]",
                ],
            )
            == {}
        )

    @pytest.mark.parametrize("task_id, project_id, user", [(20, 13, "admin1")])
    def test_move_task_from_one_project_to_another_with_attributes(self, task_id, project_id, user):
        response = get_method(user, f"tasks/{task_id}/annotations")
        assert response.status_code == HTTPStatus.OK
        annotations = response.json()

        response = patch_method(user, f"tasks/{task_id}", {"project_id": project_id})
        assert response.status_code == HTTPStatus.OK

        response = get_method(user, f"tasks/{task_id}")
        assert response.status_code == HTTPStatus.OK
        assert response.json().get("project_id") == project_id

        response = get_method(user, f"tasks/{task_id}/annotations")
        assert response.status_code == HTTPStatus.OK
        assert (
            DeepDiff(
                annotations,
                response.json(),
                ignore_order=True,
                exclude_regex_paths=[
                    r"root\['\w+'\]\[\d+\]\['label_id'\]",
                    r"root\['\w+'\]\[\d+\]\['attributes'\]\[\d+\]\['spec_id'\]",
                ],
            )
            == {}
        )

    @pytest.mark.with_external_services
    @pytest.mark.parametrize(
        "storage_id",
        [
            1,  # public bucket
            2,  # private bucket
        ],
    )
    @pytest.mark.parametrize("field", ["source_storage", "target_storage"])
    def test_user_cannot_update_task_with_cloud_storage_without_access(
        self, storage_id, field, regular_lonely_user
    ):
        user = regular_lonely_user

        task_spec = {
            "name": f"Task with files from foreign cloud storage {storage_id}",
            "labels": [
                {
                    "name": "car",
                }
            ],
        }
        data_spec = {
            "image_quality": 75,
            "use_cache": True,
            "server_files": ["images/image_1.jpg"],
        }
        (task_id, _) = create_task(user, task_spec, data_spec)

        updated_fields = {
            field: {
                "location": "cloud_storage",
                "cloud_storage_id": storage_id,
            }
        }

        with make_api_client(user) as api_client:
            (_, response) = api_client.tasks_api.partial_update(
                task_id,
                patched_task_write_request=updated_fields,
                _parse_response=False,
                _check_status=False,
            )
        assert response.status == HTTPStatus.FORBIDDEN


@pytest.mark.usefixtures("restore_db_per_function")
def test_can_report_correct_completed_jobs_count(tasks, jobs, admin_user):
    # Reproduces https://github.com/opencv/cvat/issues/6098
    task = next(
        t
        for t in tasks
        if t["jobs"]["count"] > 1 and t["jobs"]["completed"] == 0 and t["labels"]["count"] > 1
    )
    task_jobs = [j for j in jobs if j["task_id"] == task["id"]]

    with make_api_client(admin_user) as api_client:
        api_client.jobs_api.partial_update(
            task_jobs[0]["id"],
            patched_job_write_request=dict(stage="acceptance", state="completed"),
        )

        task, _ = api_client.tasks_api.retrieve(task["id"])
        assert task.jobs.completed == 1


class TestImportTaskAnnotations:
    def _make_client(self) -> Client:
        return Client(BASE_URL, config=Config(status_check_period=0.01))

    @pytest.fixture(autouse=True)
    def setup(self, restore_db_per_function, tmp_path: Path, admin_user: str):
        self.tmp_dir = tmp_path
        self.client = self._make_client()
        self.user = admin_user
        self.format = "COCO 1.0"

        with self.client:
            self.client.login((self.user, USER_PASS))

    def _check_annotations(self, task_id):
        with make_api_client(self.user) as api_client:
            (_, response) = api_client.tasks_api.retrieve_annotations(id=task_id)
            assert response.status == HTTPStatus.OK
            annotations = json.loads(response.data)["shapes"]
            assert len(annotations) > 0

    def _delete_annotations(self, task_id):
        with make_api_client(self.user) as api_client:
            (_, response) = api_client.tasks_api.destroy_annotations(id=task_id)
            assert response.status == HTTPStatus.NO_CONTENT

    @pytest.mark.skip("Fails sometimes, needs to be fixed")
    @pytest.mark.timeout(70)
    @pytest.mark.parametrize("successful_upload", [True, False])
    def test_can_import_annotations_after_previous_unclear_import(
        self, successful_upload: bool, tasks_with_shapes
    ):
        task_id = tasks_with_shapes[0]["id"]
        self._check_annotations(task_id)

        with NamedTemporaryFile() as f:
            filename = self.tmp_dir / f"task_{task_id}_{Path(f.name).name}_coco.zip"

        task = self.client.tasks.retrieve(task_id)
        task.export_dataset(self.format, filename, include_images=False)

        self._delete_annotations(task_id)

        params = {"format": self.format, "filename": filename.name}
        url = self.client.api_map.make_endpoint_url(
            self.client.api_client.tasks_api.create_annotations_endpoint.path
        ).format(id=task_id)
        uploader = Uploader(self.client)

        if successful_upload:
            # define time required to upload file with annotations
            start_time = time()
            task.import_annotations(self.format, filename)
            required_time = ceil(time() - start_time) * 2
            self._delete_annotations(task_id)

            response = uploader.upload_file(
                url, filename, meta=params, query_params=params, logger=self.client.logger.debug
            )
            rq_id = json.loads(response.data)["rq_id"]
            assert rq_id
        else:
            required_time = 60
            uploader._tus_start_upload(url, query_params=params)
            uploader._upload_file_data_with_tus(
                url, filename, meta=params, logger=self.client.logger.debug
            )

        sleep(required_time)
        if successful_upload:
            self._check_annotations(task_id)
            self._delete_annotations(task_id)
        task.import_annotations(self.format, filename)
        self._check_annotations(task_id)

    @pytest.mark.skip("Fails sometimes, needs to be fixed")
    @pytest.mark.timeout(70)
    def test_check_import_cache_after_previous_interrupted_upload(self, tasks_with_shapes, request):
        task_id = tasks_with_shapes[0]["id"]
        with NamedTemporaryFile() as f:
            filename = self.tmp_dir / f"task_{task_id}_{Path(f.name).name}_coco.zip"
        task = self.client.tasks.retrieve(task_id)
        task.export_dataset(self.format, filename, include_images=False)

        params = {"format": self.format, "filename": filename.name}
        url = self.client.api_map.make_endpoint_url(
            self.client.api_client.tasks_api.create_annotations_endpoint.path
        ).format(id=task_id)

        uploader = Uploader(self.client)
        uploader._tus_start_upload(url, query_params=params)
        uploader._upload_file_data_with_tus(
            url, filename, meta=params, logger=self.client.logger.debug
        )
        number_of_files = 1
        sleep(30)  # wait when the cleaning job from rq worker will be started
        command = ["/bin/bash", "-c", f"ls data/tasks/{task_id}/tmp | wc -l"]
        platform = request.config.getoption("--platform")
        assert platform in ("kube", "local")
        func = docker_exec_cvat if platform == "local" else kube_exec_cvat
        for _ in range(12):
            sleep(2)
            result, _ = func(command)
            number_of_files = int(result)
            if not number_of_files:
                break
        assert not number_of_files


class TestImportWithComplexFilenames:
    @staticmethod
    def _make_client() -> Client:
        return Client(BASE_URL, config=Config(status_check_period=0.01))

    @pytest.fixture(
        autouse=True,
        scope="class",
        # classmethod way may not work in some versions
        # https://github.com/opencv/cvat/actions/runs/5336023573/jobs/9670573955?pr=6350
        name="TestImportWithComplexFilenames.setup_class",
    )
    @classmethod
    def setup_class(
        cls, restore_db_per_class, tmp_path_factory: pytest.TempPathFactory, admin_user: str
    ):
        cls.tmp_dir = tmp_path_factory.mktemp(cls.__class__.__name__)
        cls.client = cls._make_client()
        cls.user = admin_user
        cls.format_name = "PASCAL VOC 1.1"

        with cls.client:
            cls.client.login((cls.user, USER_PASS))

        cls._init_tasks()

    @classmethod
    def _create_task_with_annotations(cls, filenames: List[str]):
        images = generate_image_files(len(filenames), filenames=filenames)

        source_archive_path = cls.tmp_dir / "source_data.zip"
        with zipfile.ZipFile(source_archive_path, "w") as zip_file:
            for image in images:
                zip_file.writestr(image.name, image.getvalue())

        task = cls.client.tasks.create_from_data(
            {
                "name": "test_images_with_dots",
                "labels": [{"name": "cat"}, {"name": "dog"}],
            },
            resources=[source_archive_path],
        )

        labels = task.get_labels()
        task.set_annotations(
            models.LabeledDataRequest(
                shapes=[
                    models.LabeledShapeRequest(
                        frame=frame_id,
                        label_id=labels[0].id,
                        type="rectangle",
                        points=[1, 1, 2, 2],
                    )
                    for frame_id in range(len(filenames))
                ],
            )
        )

        return task

    @classmethod
    def _init_tasks(cls):
        cls.flat_filenames = [
            "filename0.jpg",
            "file.name1.jpg",
            "fi.le.na.me.2.jpg",
            ".filename3.jpg",
            "..filename..4.jpg",
            "..filename..5.png..jpg",
        ]

        cls.nested_filenames = [
            f"{prefix}/{fn}"
            for prefix, fn in zip(
                [
                    "ab/cd",
                    "ab/cd",
                    "ab",
                    "ab",
                    "cd/ef",
                    "cd/ef",
                    "cd",
                    "",
                ],
                cls.flat_filenames,
            )
        ]

        cls.data = {}
        for (kind, filenames), prefix in product(
            [("flat", cls.flat_filenames), ("nested", cls.nested_filenames)], ["", "pre/fix"]
        ):
            key = kind
            if prefix:
                key += "_prefixed"

            task = cls._create_task_with_annotations(
                [f"{prefix}/{fn}" if prefix else fn for fn in filenames]
            )

            dataset_file = cls.tmp_dir / f"{key}_dataset.zip"
            task.export_dataset(cls.format_name, dataset_file, include_images=False)

            cls.data[key] = (task, dataset_file)

    @pytest.mark.parametrize(
        "task_kind, annotation_kind, expect_success",
        [
            ("flat", "flat", True),
            ("flat", "flat_prefixed", False),
            ("flat", "nested", False),
            ("flat", "nested_prefixed", False),
            ("flat_prefixed", "flat", True),  # allow this for better UX
            ("flat_prefixed", "flat_prefixed", True),
            ("flat_prefixed", "nested", False),
            ("flat_prefixed", "nested_prefixed", False),
            ("nested", "flat", False),
            ("nested", "flat_prefixed", False),
            ("nested", "nested", True),
            ("nested", "nested_prefixed", False),
            ("nested_prefixed", "flat", False),
            ("nested_prefixed", "flat_prefixed", False),
            ("nested_prefixed", "nested", True),  # allow this for better UX
            ("nested_prefixed", "nested_prefixed", True),
        ],
    )
    def test_import_annotations(self, task_kind, annotation_kind, expect_success):
        # Tests for regressions about https://github.com/opencv/cvat/issues/6319
        #
        # X annotations must be importable to X prefixed cases
        # with and without dots in filenames.
        #
        # Nested structures can potentially be matched to flat ones and vise-versa,
        # but it's not supported now, as it may lead to some errors in matching.

        task: Task = self.data[task_kind][0]
        dataset_file = self.data[annotation_kind][1]

        if expect_success:
            task.import_annotations(self.format_name, dataset_file)

            assert set(s.frame for s in task.get_annotations().shapes) == set(
                range(len(self.flat_filenames))
            )
        else:
            with pytest.raises(exceptions.ApiException) as capture:
                task.import_annotations(self.format_name, dataset_file)

            assert b"Could not match item id" in capture.value.body
