Add tests for Job REST API

.vscode/launch.json

@@ -100,6 +100,22 @@
             "cwd": "${workspaceFolder}",
             "env": {}
         },
+        {
+            "name": "tests",
+            "type": "python",
+            "request": "launch",
+            "debugStdLib": true,
+            "stopOnEntry": false,
+            "pythonPath": "${config:python.pythonPath}",
+            "program": "${workspaceRoot}/manage.py",
+            "args": [
+                "test",
+                "cvat/apps/engine"
+            ],
+            "django": true,
+            "cwd": "${workspaceFolder}",
+            "env": {}
+        },
     ],
     "compounds": [
         {

cvat/apps/engine/migrations/0029_auto_20190211_1853.py

@@ -0,0 +1,29 @@
+# Generated by Django 2.1.5 on 2019-02-11 15:53
+
+import cvat.apps.engine.models
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('engine', '0028_auto_20190206_1620'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='attributespec',
+            name='input_type',
+            field=models.CharField(choices=[('checkbox', 'CHECKBOX'), ('radio', 'RADIO'), ('number', 'NUMBER'), ('text', 'TEXT'), ('select', 'SELECT')], max_length=16),
+        ),
+        migrations.AlterField(
+            model_name='job',
+            name='status',
+            field=models.CharField(choices=[('annotation', 'ANNOTATION'), ('validation', 'VALIDATION'), ('completed', 'COMPLETED')], default=cvat.apps.engine.models.StatusChoice('annotation'), max_length=32),
+        ),
+        migrations.AlterField(
+            model_name='task',
+            name='status',
+            field=models.CharField(choices=[('annotation', 'ANNOTATION'), ('validation', 'VALIDATION'), ('completed', 'COMPLETED')], default=cvat.apps.engine.models.StatusChoice('annotation'), max_length=32),
+        ),
+    ]

cvat/apps/engine/models.py

@@ -23,7 +23,10 @@ class StatusChoice(str, Enum):
 
     @classmethod
     def choices(self):
-        return tuple((x.name, x.value) for x in self)
+        return tuple((x.value, x.name) for x in self)
+
+    def __str__(self):
+        return self.value
 
 class AttributeType(str, Enum):
     CHECKBOX = 'checkbox'
@@ -34,7 +37,10 @@ class AttributeType(str, Enum):
 
     @classmethod
     def choices(self):
-        return tuple((x.name, x.value) for x in self)
+        return tuple((x.value, x.name) for x in self)
+
+    def __str__(self):
+        return self.value
 
 
 class SafeCharField(models.CharField):

cvat/apps/engine/tests/test_rest_api.py

@@ -0,0 +1,225 @@
+# Copyright (C) 2018 Intel Corporation
+#
+# SPDX-License-Identifier: MIT
+
+from rest_framework.test import APITestCase, APIClient
+from rest_framework import status
+from django.contrib.auth.models import User, Group
+from cvat.apps.engine.models import Task, Segment, Job, StatusChoice
+
+def setUpModule():
+    import logging
+    global django_request_logger
+    global django_request_loglevel
+
+    django_request_logger = logging.getLogger('django.request')
+    django_request_loglevel = django_request_logger.getEffectiveLevel()
+    django_request_logger.setLevel(logging.ERROR)
+
+def tearDownModule():
+    django_request_logger.setLevel(django_request_loglevel)
+
+
+def createUsers(cls):
+    (group_admin, _) = Group.objects.get_or_create(name="admin")
+    (group_user, _) = Group.objects.get_or_create(name="user")
+    (group_annotator, _) = Group.objects.get_or_create(name="annotator")
+    (group_observer, _) = Group.objects.get_or_create(name="observer")
+
+    user_admin = User.objects.create_superuser(username="admin", email="",
+        password="admin")
+    user_admin.groups.add(group_admin)
+    user_owner = User.objects.create_user(username="user1", password="user1")
+    user_owner.groups.add(group_user)
+    user_assignee = User.objects.create_user(username="user2", password="user2")
+    user_assignee.groups.add(group_annotator)
+    user_annotator = User.objects.create_user(username="user3", password="user3")
+    user_annotator.groups.add(group_annotator)
+    user_observer = User.objects.create_user(username="user4", password="user4")
+    user_observer.groups.add(group_observer)
+    user_dummy = User.objects.create_user(username="user5", password="user5")
+    user_dummy.groups.add(group_user)
+
+    cls.admin = user_admin
+    cls.owner = user_owner
+    cls.assignee = user_assignee
+    cls.annotator = user_annotator
+    cls.observer = user_observer
+    cls.user = user_dummy
+
+def createTask(cls):
+    task = {
+        "name": "my test task",
+        "owner": cls.owner,
+        "assignee": cls.assignee,
+        "overlap": 0,
+        "segment_size": 100,
+        "z_order": False,
+        "image_quality": 75,
+        "size": 100
+    }
+    cls.task = Task.objects.create(**task)
+
+    segment = {
+        "start_frame": 0,
+        "stop_frame": 100
+    }
+    cls.segment = Segment.objects.create(task=cls.task, **segment)
+    cls.job = Job.objects.create(segment=cls.segment, assignee=cls.annotator)
+
+
+class JobGetAPITestCase(APITestCase):
+    def setUp(self):
+        self.client = APIClient()
+
+    @classmethod
+    def setUpTestData(cls):
+        createUsers(cls)
+        createTask(cls)
+
+    def _run_api_v1_jobs_id(self, jid, user):
+        if user:
+            self.client.force_login(user, backend='django.contrib.auth.backends.ModelBackend')
+
+        response = self.client.get('/api/v1/jobs/{}'.format(jid))
+
+        if user:
+            self.client.logout()
+
+        return response
+
+    def _check_request(self, response):
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["id"], self.job.id)
+        self.assertEqual(response.data["status"], StatusChoice.ANNOTATION)
+        self.assertEqual(response.data["start_frame"], self.job.segment.start_frame)
+        self.assertEqual(response.data["stop_frame"], self.job.segment.stop_frame)
+
+    def test_api_v1_jobs_id_admin(self):
+        response = self._run_api_v1_jobs_id(self.job.id, self.admin)
+        self._check_request(response)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.admin)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_owner(self):
+        response = self._run_api_v1_jobs_id(self.job.id, self.owner)
+        self._check_request(response)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.owner)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_annotator(self):
+        response = self._run_api_v1_jobs_id(self.job.id, self.annotator)
+        self._check_request(response)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.annotator)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_observer(self):
+        response = self._run_api_v1_jobs_id(self.job.id, self.observer)
+        self._check_request(response)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.observer)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_user(self):
+        response = self._run_api_v1_jobs_id(self.job.id, self.user)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.user)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_no_auth(self):
+        response = self._run_api_v1_jobs_id(self.job.id, None)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, None)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+
+class JobUpdateAPITestCase(APITestCase):
+    def setUp(self):
+        self.client = APIClient()
+        createTask(self)
+
+    @classmethod
+    def setUpTestData(cls):
+        createUsers(cls)
+
+    def _run_api_v1_jobs_id(self, jid, user, data):
+        if user:
+            self.client.force_login(user, backend='django.contrib.auth.backends.ModelBackend')
+
+        response = self.client.put('/api/v1/jobs/{}'.format(jid), data=data, format='json')
+
+        if user:
+            self.client.logout()
+
+        return response
+
+    def _check_request(self, response, data):
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["id"], self.job.id)
+        self.assertEqual(response.data["status"], data.get('status', self.job.status))
+        self.assertEqual(response.data["assignee"], data.get('assignee', self.job.assignee.id))
+        self.assertEqual(response.data["start_frame"], self.job.segment.start_frame)
+        self.assertEqual(response.data["stop_frame"], self.job.segment.stop_frame)
+
+    def test_api_v1_jobs_id_admin(self):
+        data = {"status": StatusChoice.COMPLETED, "assignee": self.owner.id}
+        response = self._run_api_v1_jobs_id(self.job.id, self.admin, data)
+        self._check_request(response, data)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.admin, data)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_owner(self):
+        data = {"status": StatusChoice.VALIDATION, "assignee": self.annotator.id}
+        response = self._run_api_v1_jobs_id(self.job.id, self.owner, data)
+        self._check_request(response, data)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.owner, data)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_annotator(self):
+        data = {"status": StatusChoice.ANNOTATION, "assignee": self.user.id}
+        response = self._run_api_v1_jobs_id(self.job.id, self.annotator, data)
+        self._check_request(response, data)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.annotator, data)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_observer(self):
+        data = {"status": StatusChoice.ANNOTATION, "assignee": self.admin.id}
+        response = self._run_api_v1_jobs_id(self.job.id, self.observer, data)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.observer, data)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_user(self):
+        data = {"status": StatusChoice.ANNOTATION, "assignee": self.user.id}
+        response = self._run_api_v1_jobs_id(self.job.id, self.user, data)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, self.user, data)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_api_v1_jobs_id_no_auth(self):
+        data = {"status": StatusChoice.ANNOTATION, "assignee": self.user.id}
+        response = self._run_api_v1_jobs_id(self.job.id, None, data)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        response = self._run_api_v1_jobs_id(self.job.id + 10, None, data)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+class JobPartialUpdateAPITestCase(JobUpdateAPITestCase):
+    def _run_api_v1_jobs_id(self, jid, user, data):
+        if user:
+            self.client.force_login(user, backend='django.contrib.auth.backends.ModelBackend')
+
+        response = self.client.patch('/api/v1/jobs/{}'.format(jid), data=data, format='json')
+
+        if user:
+            self.client.logout()
+
+        return response
+
+    def test_api_v1_jobs_id_annotator_partial(self):
+        data = {"status": StatusChoice.VALIDATION}
+        response = self._run_api_v1_jobs_id(self.job.id, self.owner, data)
+        self._check_request(response, data)
+
+    def test_api_v1_jobs_id_admin_partial(self):
+        data = {"assignee": self.user.id}
+        response = self._run_api_v1_jobs_id(self.job.id, self.owner, data)
+        self._check_request(response, data)

cvat/apps/engine/views.py

@@ -100,7 +100,7 @@ def get_permissions(self):
         http_method = self.request.method
         permissions = [auth.IsAuthenticated]
 
-        if http_method in auth.SAFE_METHODS:
+        if http_method in SAFE_METHODS:
             permissions.append(auth.TaskAccessPermission)
         elif http_method in ["POST"]:
             permissions.append(auth.TaskCreatePermission)

cvat/settings/development.py

@@ -11,6 +11,8 @@
     'django_extensions',
 ]
 
+ALLOWED_HOSTS.append('testserver')
+
 # Django-sendfile:
 # https://github.com/johnsensible/django-sendfile
 SENDFILE_BACKEND = 'sendfile.backends.development'