Skip to content

Commit

Permalink
Basic user information (#761)
Browse files Browse the repository at this point in the history
* Fix #750
* Updated CHANGELOG
* Added more tests for /api/v1/users* REST API.
  • Loading branch information
nmanovic authored Oct 8, 2019
1 parent a18fb6d commit f5c5624
Show file tree
Hide file tree
Showing 4 changed files with 166 additions and 98 deletions.
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [1.0.0.alpha] - 2020-02-XX
### Added
- Server only support for projects. Extend REST API v1 (/api/v1/projects*).
- Ability to get basic information about users without admin permissions (#750).
Changed REST API: removed PUT and added DELETE methods for /api/v1/users/ID.

### Changed
-
Expand Down
20 changes: 19 additions & 1 deletion cvat/apps/engine/serializers.py
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,24 @@ class Meta:
read_only_fields = ('created_date', 'updated_date', 'status')
ordering = ['-id']

class BasicUserSerializer(serializers.ModelSerializer):
def validate(self, data):
if hasattr(self, 'initial_data'):
unknown_keys = set(self.initial_data.keys()) - set(self.fields.keys())
if unknown_keys:
if set(['is_staff', 'is_superuser', 'groups']) & unknown_keys:
message = 'You do not have permissions to access some of' + \
' these fields: {}'.format(unknown_keys)
else:
message = 'Got unknown fields: {}'.format(unknown_keys)
raise serializers.ValidationError(message)
return data

class Meta:
model = User
fields = ('url', 'id', 'username', 'first_name', 'last_name', 'email')
ordering = ['-id']

class UserSerializer(serializers.ModelSerializer):
groups = serializers.SlugRelatedField(many=True,
slug_field='name', queryset=Group.objects.all())
Expand All @@ -294,7 +312,7 @@ class Meta:
model = User
fields = ('url', 'id', 'username', 'first_name', 'last_name', 'email',
'groups', 'is_staff', 'is_superuser', 'is_active', 'last_login',
'date_joined', 'groups')
'date_joined')
read_only_fields = ('last_login', 'date_joined')
write_only_fields = ('password', )
ordering = ['-id']
Expand Down
208 changes: 121 additions & 87 deletions cvat/apps/engine/tests/test_rest_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,11 @@ def create_db_users(cls):
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
cls.owner = cls.user1 = user_owner
cls.assignee = cls.user2 = user_assignee
cls.annotator = cls.user3 = user_annotator
cls.observer = cls.user4 = user_observer
cls.user = cls.user5 = user_dummy

def create_db_task(data):
db_task = Task.objects.create(**data)
Expand Down Expand Up @@ -462,53 +462,69 @@ def test_api_v1_server_logs_no_auth(self):
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)


class UserListAPITestCase(APITestCase):
class UserAPITestCase(APITestCase):
def setUp(self):
self.client = APIClient()
create_db_users(self)

@classmethod
def setUpTestData(cls):
create_db_users(cls)

def _check_response(self, user, response, is_full=True):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self._check_data(user, response.data, is_full)

def _check_data(self, user, data, is_full):
self.assertEqual(data["id"], user.id)
self.assertEqual(data["username"], user.username)
self.assertEqual(data["first_name"], user.first_name)
self.assertEqual(data["last_name"], user.last_name)
self.assertEqual(data["email"], user.email)
extra_check = self.assertIn if is_full else self.assertNotIn
extra_check("groups", data)
extra_check("is_staff", data)
extra_check("is_superuser", data)
extra_check("is_active", data)
extra_check("last_login", data)
extra_check("date_joined", data)

class UserListAPITestCase(UserAPITestCase):
def _run_api_v1_users(self, user):
with ForceLogin(user, self.client):
response = self.client.get('/api/v1/users')

return response

def _check_response(self, user, response, is_full):
self.assertEqual(response.status_code, status.HTTP_200_OK)
for user_info in response.data['results']:
db_user = getattr(self, user_info['username'])
self._check_data(db_user, user_info, is_full)

def test_api_v1_users_admin(self):
response = self._run_api_v1_users(self.admin)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertListEqual(
["admin", "user1", "user2", "user3", "user4", "user5"],
[res["username"] for res in response.data["results"]])
self._check_response(self.admin, response, True)

def test_api_v1_users_user(self):
response = self._run_api_v1_users(self.user)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self._check_response(self.user, response, False)

def test_api_v1_users_annotator(self):
response = self._run_api_v1_users(self.annotator)
self._check_response(self.annotator, response, False)

def test_api_v1_users_observer(self):
response = self._run_api_v1_users(self.observer)
self._check_response(self.observer, response, False)

def test_api_v1_users_no_auth(self):
response = self._run_api_v1_users(None)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

class UserSelfAPITestCase(APITestCase):
def setUp(self):
self.client = APIClient()

@classmethod
def setUpTestData(cls):
create_db_users(cls)

class UserSelfAPITestCase(UserAPITestCase):
def _run_api_v1_users_self(self, user):
with ForceLogin(user, self.client):
response = self.client.get('/api/v1/users/self')

return response

def _check_response(self, user, response):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["username"], user.username)

def test_api_v1_users_self_admin(self):
response = self._run_api_v1_users_self(self.admin)
self._check_response(self.admin, response)
Expand All @@ -521,121 +537,139 @@ def test_api_v1_users_self_annotator(self):
response = self._run_api_v1_users_self(self.annotator)
self._check_response(self.annotator, response)

def test_api_v1_users_self_observer(self):
response = self._run_api_v1_users_self(self.observer)
self._check_response(self.observer, response)

def test_api_v1_users_self_no_auth(self):
response = self._run_api_v1_users_self(None)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

class UserGetAPITestCase(APITestCase):
def setUp(self):
self.client = APIClient()

@classmethod
def setUpTestData(cls):
create_db_users(cls)

class UserGetAPITestCase(UserAPITestCase):
def _run_api_v1_users_id(self, user, user_id):
with ForceLogin(user, self.client):
response = self.client.get('/api/v1/users/{}'.format(user_id))

return response

def _check_response(self, user, response):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["id"], user.id)
self.assertEqual(response.data["username"], user.username)

def test_api_v1_users_id_admin(self):
response = self._run_api_v1_users_id(self.admin, self.user.id)
self._check_response(self.user, response)
self._check_response(self.user, response, True)

response = self._run_api_v1_users_id(self.admin, self.admin.id)
self._check_response(self.admin, response)
self._check_response(self.admin, response, True)

response = self._run_api_v1_users_id(self.admin, self.owner.id)
self._check_response(self.owner, response)
self._check_response(self.owner, response, True)

def test_api_v1_users_id_user(self):
response = self._run_api_v1_users_id(self.user, self.user.id)
self._check_response(self.user, response)
self._check_response(self.user, response, True)

response = self._run_api_v1_users_id(self.user, self.owner.id)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self._check_response(self.owner, response, False)

def test_api_v1_users_id_annotator(self):
response = self._run_api_v1_users_id(self.annotator, self.annotator.id)
self._check_response(self.annotator, response)
self._check_response(self.annotator, response, True)

response = self._run_api_v1_users_id(self.annotator, self.user.id)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self._check_response(self.user, response, False)

def test_api_v1_users_id_observer(self):
response = self._run_api_v1_users_id(self.observer, self.observer.id)
self._check_response(self.observer, response, True)

response = self._run_api_v1_users_id(self.observer, self.user.id)
self._check_response(self.user, response, False)

def test_api_v1_users_id_no_auth(self):
response = self._run_api_v1_users_id(None, self.user.id)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

class UserUpdateAPITestCase(APITestCase):
def setUp(self):
self.client = APIClient()
create_db_users(self)

class UserPartialUpdateAPITestCase(UserAPITestCase):
def _run_api_v1_users_id(self, user, user_id, data):
with ForceLogin(user, self.client):
response = self.client.put('/api/v1/users/{}'.format(user_id), data=data)
response = self.client.patch('/api/v1/users/{}'.format(user_id), data=data)

return response

def test_api_v1_users_id_admin(self):
data = {"username": "user09", "groups": ["user", "admin"],
"first_name": "my name"}
def _check_response_with_data(self, user, response, data, is_full):
# refresh information about the user from DB
user = User.objects.get(id=user.id)
for k,v in data.items():
self.assertEqual(response.data[k], v)
self._check_response(user, response, is_full)

def test_api_v1_users_id_admin_partial(self):
data = {"username": "user09", "last_name": "my last name"}
response = self._run_api_v1_users_id(self.admin, self.user.id, data)

self.assertEqual(response.status_code, status.HTTP_200_OK)
user09 = User.objects.get(id=self.user.id)
self.assertEqual(user09.username, data["username"])
self.assertEqual(user09.first_name, data["first_name"])
self._check_response_with_data(self.user, response, data, True)

def test_api_v1_users_id_user(self):
data = {"username": "user10", "groups": ["user", "annotator"],
"first_name": "my name"}
def test_api_v1_users_id_user_partial(self):
data = {"username": "user10", "first_name": "my name"}
response = self._run_api_v1_users_id(self.user, self.user.id, data)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self._check_response_with_data(self.user, response, data, False)

def test_api_v1_users_id_annotator(self):
data = {"username": "user11", "groups": ["annotator"],
"first_name": "my name"}
response = self._run_api_v1_users_id(self.annotator, self.user.id, data)
data = {"is_staff": True}
response = self._run_api_v1_users_id(self.user, self.user.id, data)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

data = {"username": "admin", "is_superuser": True}
response = self._run_api_v1_users_id(self.user, self.user.id, data)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

data = {"username": "non_active", "is_active": False}
response = self._run_api_v1_users_id(self.user, self.user.id, data)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

data = {"username": "annotator01", "first_name": "slave"}
response = self._run_api_v1_users_id(self.user, self.annotator.id, data)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

def test_api_v1_users_id_no_auth(self):
data = {"username": "user12", "groups": ["user", "observer"],
"first_name": "my name"}
def test_api_v1_users_id_no_auth_partial(self):
data = {"username": "user12"}
response = self._run_api_v1_users_id(None, self.user.id, data)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

class UserPartialUpdateAPITestCase(UserUpdateAPITestCase):
def _run_api_v1_users_id(self, user, user_id, data):
class UserDeleteAPITestCase(UserAPITestCase):
def _run_api_v1_users_id(self, user, user_id):
with ForceLogin(user, self.client):
response = self.client.patch('/api/v1/users/{}'.format(user_id), data=data)
response = self.client.delete('/api/v1/users/{}'.format(user_id))

return response

def test_api_v1_users_id_admin_partial(self):
data = {"username": "user09", "last_name": "my last name"}
response = self._run_api_v1_users_id(self.admin, self.user.id, data)
def test_api_v1_users_id_admin(self):
response = self._run_api_v1_users_id(self.admin, self.user.id)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

self.assertEqual(response.status_code, status.HTTP_200_OK)
user09 = User.objects.get(id=self.user.id)
self.assertEqual(user09.username, data["username"])
self.assertEqual(user09.last_name, data["last_name"])
response = self._run_api_v1_users_id(self.admin, self.admin.id)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

def test_api_v1_users_id_user_partial(self):
data = {"username": "user10", "first_name": "my name"}
response = self._run_api_v1_users_id(self.user, self.user.id, data)
def test_api_v1_users_id_user(self):
response = self._run_api_v1_users_id(self.user, self.owner.id)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

def test_api_v1_users_id_no_auth_partial(self):
data = {"username": "user12"}
response = self._run_api_v1_users_id(None, self.user.id, data)
response = self._run_api_v1_users_id(self.user, self.user.id)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

def test_api_v1_users_id_annotator(self):
response = self._run_api_v1_users_id(self.annotator, self.user.id)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

response = self._run_api_v1_users_id(self.annotator, self.annotator.id)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

def test_api_v1_users_id_observer(self):
response = self._run_api_v1_users_id(self.observer, self.user.id)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

response = self._run_api_v1_users_id(self.observer, self.observer.id)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

def test_api_v1_users_id_no_auth(self):
response = self._run_api_v1_users_id(None, self.user.id)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

class ProjectListAPITestCase(APITestCase):
Expand Down
Loading

0 comments on commit f5c5624

Please sign in to comment.