diff --git a/tests/test_api.py b/tests/test_api.py
index 81f02480fdf52221bfa1ad7b2bed695f0b06a5ee..4d1a0bf7bc56df8cd0031e77b6c4f09c64355566 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -109,3 +109,125 @@ class TestAPICheckPassword(UffdTestCase):
r = self.client.post(path=url_for('api.checkpassword'), data={'loginname': 'testuser', 'password': 'wrongpassword'}, headers=[basic_auth('test', 'test')])
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json, None)
+
+class TestAPIGetusers(UffdTestCase):
+ def setUpDB(self):
+ db.session.add(APIClient(service=Service(name='test'), auth_username='test', auth_password='test', perm_users=True))
+
+ def fix_result(self, result):
+ result.sort(key=lambda user: user['id'])
+ for user in result:
+ user['groups'].sort()
+ return result
+
+ def test_all(self):
+ r = self.client.get(path=url_for('api.getusers'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'displayname': 'Test User', 'email': 'test@example.com', 'id': 10000, 'loginname': 'testuser', 'groups': ['uffd_access', 'users']},
+ {'displayname': 'Test Admin', 'email': 'admin@example.com', 'id': 10001, 'loginname': 'testadmin', 'groups': ['uffd_access', 'uffd_admin', 'users']}
+ ])
+
+ def test_id(self):
+ r = self.client.get(path=url_for('api.getusers', id=10000), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'displayname': 'Test User', 'email': 'test@example.com', 'id': 10000, 'loginname': 'testuser', 'groups': ['uffd_access', 'users']},
+ ])
+
+ def test_id_empty(self):
+ r = self.client.get(path=url_for('api.getusers', id=0), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json, [])
+
+ def test_loginname(self):
+ r = self.client.get(path=url_for('api.getusers', loginname='testuser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'displayname': 'Test User', 'email': 'test@example.com', 'id': 10000, 'loginname': 'testuser', 'groups': ['uffd_access', 'users']},
+ ])
+
+ def test_loginname_empty(self):
+ r = self.client.get(path=url_for('api.getusers', loginname='notauser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json, [])
+
+ def test_email(self):
+ r = self.client.get(path=url_for('api.getusers', email='admin@example.com'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'displayname': 'Test Admin', 'email': 'admin@example.com', 'id': 10001, 'loginname': 'testadmin', 'groups': ['uffd_access', 'uffd_admin', 'users']}
+ ])
+
+ def test_email_empty(self):
+ r = self.client.get(path=url_for('api.getusers', email='foo@bar'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json, [])
+
+ def test_group(self):
+ r = self.client.get(path=url_for('api.getusers', group='uffd_admin'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'displayname': 'Test Admin', 'email': 'admin@example.com', 'id': 10001, 'loginname': 'testadmin', 'groups': ['uffd_access', 'uffd_admin', 'users']}
+ ])
+
+ def test_group_empty(self):
+ r = self.client.get(path=url_for('api.getusers', group='notagroup'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json, [])
+
+class TestAPIGetgroups(UffdTestCase):
+ def setUpDB(self):
+ db.session.add(APIClient(service=Service(name='test'), auth_username='test', auth_password='test', perm_users=True))
+
+ def fix_result(self, result):
+ result.sort(key=lambda group: group['id'])
+ for group in result:
+ group['members'].sort()
+ return result
+
+ def test_all(self):
+ r = self.client.get(path=url_for('api.getgroups'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'id': 20001, 'members': ['testadmin', 'testuser'], 'name': 'users'},
+ {'id': 20002, 'members': ['testadmin', 'testuser'], 'name': 'uffd_access'},
+ {'id': 20003, 'members': ['testadmin'], 'name': 'uffd_admin'}
+ ])
+
+ def test_id(self):
+ r = self.client.get(path=url_for('api.getgroups', id=20002), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'id': 20002, 'members': ['testadmin', 'testuser'], 'name': 'uffd_access'},
+ ])
+
+ def test_id_empty(self):
+ r = self.client.get(path=url_for('api.getgroups', id=0), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json, [])
+
+ def test_name(self):
+ r = self.client.get(path=url_for('api.getgroups', name='uffd_admin'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'id': 20003, 'members': ['testadmin'], 'name': 'uffd_admin'}
+ ])
+
+ def test_name_empty(self):
+ r = self.client.get(path=url_for('api.getgroups', name='notagroup'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json, [])
+
+ def test_member(self):
+ r = self.client.get(path=url_for('api.getgroups', member='testuser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.fix_result(r.json), [
+ {'id': 20001, 'members': ['testadmin', 'testuser'], 'name': 'users'},
+ {'id': 20002, 'members': ['testadmin', 'testuser'], 'name': 'uffd_access'},
+ ])
+
+ def test_member_empty(self):
+ r = self.client.get(path=url_for('api.getgroups', member='notauser'), headers=[basic_auth('test', 'test')], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json, [])
diff --git a/uffd/api/views.py b/uffd/api/views.py
index 208b7ff77b5c3ca13e8699676ef6943985cbc5d4..1f406d2470f500f84e1764efcf7e682fe83c0d56 100644
--- a/uffd/api/views.py
+++ b/uffd/api/views.py
@@ -34,8 +34,11 @@ def apikey_required(permission=None):
return wrapper
def generate_group_dict(group):
- return {'id': group.unix_gid, 'name': group.name,
- 'members': [user.loginname for user in group.members]}
+ return {
+ 'id': group.unix_gid,
+ 'name': group.name,
+ 'members': [user.loginname for user in group.members]
+ }
@bp.route('/getgroups', methods=['GET', 'POST'])
@apikey_required('users')
@@ -44,24 +47,30 @@ def getgroups():
abort(400)
key = (list(request.values.keys()) or [None])[0]
values = request.values.getlist(key)
+ query = Group.query
if key is None:
- groups = Group.query.all()
+ pass
elif key == 'id' and len(values) == 1:
- groups = Group.query.filter_by(unix_gid=values[0]).all()
+ query = query.filter(Group.unix_gid == values[0])
elif key == 'name' and len(values) == 1:
- groups = Group.query.filter_by(name=values[0]).all()
+ query = query.filter(Group.name == values[0])
elif key == 'member' and len(values) == 1:
- user = User.query.filter_by(loginname=values[0]).one_or_none()
- groups = [] if user is None else user.groups
+ query = query.join(Group.members).filter(User.loginname == values[0])
else:
abort(400)
- return jsonify([generate_group_dict(group) for group in groups])
+ # Single-result queries perform better without joinedload
+ if key is None or key == 'member':
+ query = query.options(db.joinedload(Group.members))
+ return jsonify([generate_group_dict(group) for group in query])
-def generate_user_dict(user, all_groups=None):
- if all_groups is None:
- all_groups = user.groups
- return {'id': user.unix_uid, 'loginname': user.loginname, 'email': user.mail, 'displayname': user.displayname,
- 'groups': [group.name for group in all_groups if user in group.members]}
+def generate_user_dict(user):
+ return {
+ 'id': user.unix_uid,
+ 'loginname': user.loginname,
+ 'email': user.mail,
+ 'displayname': user.displayname,
+ 'groups': [group.name for group in user.groups]
+ }
@bp.route('/getusers', methods=['GET', 'POST'])
@apikey_required('users')
@@ -70,23 +79,23 @@ def getusers():
abort(400)
key = (list(request.values.keys()) or [None])[0]
values = request.values.getlist(key)
+ query = User.query
if key is None:
- users = User.query.all()
+ pass
elif key == 'id' and len(values) == 1:
- users = User.query.filter_by(unix_uid=values[0]).all()
+ query = query.filter(User.unix_uid == values[0])
elif key == 'loginname' and len(values) == 1:
- users = User.query.filter_by(loginname=values[0]).all()
+ query = query.filter(User.loginname == values[0])
elif key == 'email' and len(values) == 1:
- users = User.query.filter_by(mail=values[0]).all()
+ query = query.filter(User.mail == values[0])
elif key == 'group' and len(values) == 1:
- group = Group.query.filter_by(name=values[0]).one_or_none()
- users = [] if group is None else group.members
+ query = query.join(User.groups).filter(Group.name == values[0])
else:
abort(400)
- all_groups = None
- if len(users) > 1:
- all_groups = Group.query.all()
- return jsonify([generate_user_dict(user, all_groups) for user in users])
+ # Single-result queries perform better without joinedload
+ if key is None or key == 'group':
+ query = query.options(db.joinedload(User.groups))
+ return jsonify([generate_user_dict(user) for user in query])
@bp.route('/checkpassword', methods=['POST'])
@apikey_required('checkpassword')
@@ -108,8 +117,11 @@ def checkpassword():
return jsonify(generate_user_dict(user))
def generate_mail_dict(mail):
- return {'name': mail.uid, 'receive_addresses': list(mail.receivers),
- 'destination_addresses': list(mail.destinations)}
+ return {
+ 'name': mail.uid,
+ 'receive_addresses': list(mail.receivers),
+ 'destination_addresses': list(mail.destinations)
+ }
@bp.route('/getmails', methods=['GET', 'POST'])
@apikey_required('mail_aliases')