200 lines
7.2 KiB
Python
200 lines
7.2 KiB
Python
import random
|
|
import uuid
|
|
|
|
import pytest
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.urls import reverse
|
|
|
|
from authentic2.a2_rbac.models import ADD_OP, SEARCH_OP, VIEW_OP, Role
|
|
from authentic2.a2_rbac.utils import get_default_ou
|
|
from authentic2.models import APIClient, Attribute
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
@pytest.fixture
|
|
def api_client(db):
|
|
return APIClient.objects.create(name='foo', identifier='id', password='pass')
|
|
|
|
|
|
def test_has_perm(api_client):
|
|
role_ct = ContentType.objects.get_for_model(Role)
|
|
role_admin_role = Role.objects.get_admin_role(role_ct, 'admin %s' % role_ct, 'admin-role')
|
|
api_client = APIClient.objects.create(name='foo')
|
|
assert not api_client.has_perm('a2_rbac.change_role')
|
|
assert not api_client.has_perm('a2_rbac.view_role')
|
|
assert not api_client.has_perm('a2_rbac.delete_role')
|
|
assert not api_client.has_perm('a2_rbac.add_role')
|
|
role_admin_role.apiclients.add(api_client)
|
|
del api_client._rbac_perms_cache
|
|
assert api_client.has_perm('a2_rbac.change_role')
|
|
assert api_client.has_perm('a2_rbac.view_role')
|
|
assert api_client.has_perm('a2_rbac.delete_role')
|
|
assert api_client.has_perm('a2_rbac.add_role')
|
|
|
|
|
|
def test_has_perm_ou(api_client, ou1):
|
|
role_ct = ContentType.objects.get_for_model(Role)
|
|
role_admin_role = Role.objects.get_admin_role(role_ct, 'admin %s' % role_ct, 'admin-role')
|
|
api_client = APIClient.objects.create(name='foo', ou=ou1)
|
|
assert not api_client.has_ou_perm('a2_rbac.change_role', ou1)
|
|
assert not api_client.has_ou_perm('a2_rbac.view_role', ou1)
|
|
assert not api_client.has_ou_perm('a2_rbac.delete_role', ou1)
|
|
assert not api_client.has_ou_perm('a2_rbac.add_role', ou1)
|
|
role_admin_role.apiclients.add(api_client)
|
|
del api_client._rbac_perms_cache
|
|
assert api_client.has_ou_perm('a2_rbac.change_role', ou1)
|
|
assert api_client.has_ou_perm('a2_rbac.view_role', ou1)
|
|
assert api_client.has_ou_perm('a2_rbac.delete_role', ou1)
|
|
assert api_client.has_ou_perm('a2_rbac.add_role', ou1)
|
|
|
|
|
|
def test_api_users_list(app, api_client):
|
|
user = User.objects.create(username='user1')
|
|
|
|
app.authorization = ('Basic', ('foo', 'bar'))
|
|
resp = app.get('/api/users/', status=401)
|
|
|
|
app.authorization = ('Basic', (api_client.identifier, api_client.password))
|
|
resp = app.get('/api/users/')
|
|
assert len(resp.json['results']) == 0
|
|
|
|
# give permissions
|
|
r = Role.objects.get_admin_role(
|
|
ContentType.objects.get_for_model(User), name='role', slug='role', operation=VIEW_OP
|
|
)
|
|
api_client.apiclient_roles.add(r)
|
|
resp = app.get('/api/users/')
|
|
assert len(resp.json['results']) == 1
|
|
|
|
preferred_color = Attribute.objects.create(
|
|
name='preferred_color',
|
|
label='Preferred color',
|
|
kind='string',
|
|
disabled=False,
|
|
multiple=False,
|
|
)
|
|
phone2 = Attribute.objects.create(
|
|
name='phone2',
|
|
label='Second phone number',
|
|
kind='phone_number',
|
|
disabled=False,
|
|
multiple=False,
|
|
)
|
|
user.attributes.preferred_color = 'blue'
|
|
user.attributes.phone2 = '+33122334455'
|
|
user.save()
|
|
|
|
api_client.allowed_user_attributes.add(preferred_color, phone2)
|
|
api_client.save()
|
|
resp = app.get('/api/users/')
|
|
assert len(resp.json['results']) == 1
|
|
assert resp.json['results'][0]['preferred_color'] == 'blue'
|
|
assert resp.json['results'][0]['phone2'] == '+33122334455'
|
|
|
|
api_client.allowed_user_attributes.remove(preferred_color)
|
|
api_client.save()
|
|
resp = app.get('/api/users/')
|
|
assert len(resp.json['results']) == 1
|
|
assert 'preferred_color' not in resp.json['results'][0]
|
|
assert resp.json['results'][0]['phone2'] == '+33122334455'
|
|
|
|
api_client.allowed_user_attributes.remove(phone2)
|
|
api_client.save()
|
|
resp = app.get('/api/users/')
|
|
assert len(resp.json['results']) == 1
|
|
assert resp.json['results'][0]['preferred_color'] == 'blue'
|
|
assert resp.json['results'][0]['phone2'] == '+33122334455'
|
|
|
|
api_client.allowed_user_attributes.add(preferred_color)
|
|
api_client.save()
|
|
resp = app.get('/api/users/')
|
|
assert len(resp.json['results']) == 1
|
|
assert resp.json['results'][0]['preferred_color'] == 'blue'
|
|
assert 'phone2' not in resp.json['results'][0]
|
|
|
|
|
|
def test_api_user_synchronization(app, api_client):
|
|
uuids = []
|
|
for _ in range(100):
|
|
user = User.objects.create(first_name='ben', last_name='dauve')
|
|
uuids.append(user.uuid)
|
|
unknown_uuids = [uuid.uuid4().hex for i in range(100)]
|
|
url = reverse('a2-api-users-synchronization')
|
|
content = {
|
|
'known_uuids': uuids + unknown_uuids,
|
|
}
|
|
random.shuffle(content['known_uuids'])
|
|
|
|
app.authorization = ('Basic', ('foo', 'bar'))
|
|
response = app.post_json(url, params=content, status=401)
|
|
|
|
app.authorization = ('Basic', (api_client.identifier, api_client.password))
|
|
response = app.post_json(url, params=content, status=403)
|
|
|
|
# give custom_user.search_user permission to user
|
|
r = Role.objects.get_admin_role(
|
|
ContentType.objects.get_for_model(User), name='role', slug='role', operation=SEARCH_OP
|
|
)
|
|
api_client.apiclient_roles.add(r)
|
|
response = app.post_json(url, params=content)
|
|
assert response.json['result'] == 1
|
|
assert set(response.json['unknown_uuids']) == set(unknown_uuids)
|
|
|
|
|
|
def test_api_user_synchronization_ou(app, api_client, ou1):
|
|
uuids = []
|
|
authorized_uuids = []
|
|
for index in range(100):
|
|
ou = ou1 if index % 2 else get_default_ou()
|
|
user = User.objects.create(first_name='ben', last_name='dauve', ou=ou)
|
|
uuids.append(user.uuid)
|
|
if index % 2:
|
|
authorized_uuids.append(user.uuid)
|
|
unknown_uuids = [uuid.uuid4().hex for i in range(100)]
|
|
url = reverse('a2-api-users-synchronization')
|
|
content = {
|
|
'known_uuids': uuids + unknown_uuids,
|
|
}
|
|
random.shuffle(content['known_uuids'])
|
|
|
|
app.authorization = ('Basic', ('foo', 'bar'))
|
|
response = app.post_json(url, params=content, status=401)
|
|
|
|
app.authorization = ('Basic', (api_client.identifier, api_client.password))
|
|
response = app.post_json(url, params=content, status=403)
|
|
|
|
# give custom_user.search_user permission to user
|
|
r = Role.objects.get_admin_role(
|
|
ContentType.objects.get_for_model(User),
|
|
name='role',
|
|
slug='role',
|
|
ou=ou1,
|
|
operation=SEARCH_OP,
|
|
)
|
|
api_client.apiclient_roles.add(r)
|
|
response = app.post_json(url, params=content)
|
|
assert response.json['result'] == 1
|
|
assert set(response.json['unknown_uuids']) != set(unknown_uuids)
|
|
assert set(unknown_uuids).issubset(set(response.json['unknown_uuids']))
|
|
|
|
|
|
def test_api_users_create(app, api_client):
|
|
payload = {
|
|
'username': 'janedoe',
|
|
'email': 'jane.doe@nowhere.null',
|
|
'first_name': 'Jane',
|
|
'last_name': 'Doe',
|
|
}
|
|
app.authorization = ('Basic', (api_client.identifier, api_client.password))
|
|
resp = app.post_json('/api/users/', params=payload, status=403)
|
|
|
|
# give permissions
|
|
r = Role.objects.get_admin_role(
|
|
ContentType.objects.get_for_model(User), name='role', slug='role', operation=ADD_OP
|
|
)
|
|
api_client.apiclient_roles.add(r)
|
|
resp = app.post_json('/api/users/', params=payload)
|
|
assert User.objects.get(uuid=resp.json['uuid'])
|