archive/alerta_alerta
0
0
Fork 0
mirror of https://github.com/alerta/alerta.git synced 2025-03-13 04:53:16 +00:00
Code Issues Projects Releases Wiki Activity

Add test for user updates ()

Browse source
This commit is contained in:
Nick Satterly 2019-01-15 10:37:22 +01:00 committed by GitHub
parent 2c1d4ef152
commit 7dd0e20c28
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 113 additions and 14 deletions
alerta
database/backends/mongodb
base.py
views
users.py
tests
test_auth.py

38
alerta/database/backends/mongodb/base.py
View file

@ -378,7 +378,7 @@ class Backend(Database):
def update_attributes(self, id, old_attrs, new_attrs):
"""
Set all attributes (including private attributes) and unset attributes by using a value of 'null'.
Set all attributes and unset attributes by using a value of 'null'.
"""
update = dict()
set_value = {'attributes.' + k: v for k, v in new_attrs.items() if v is not None}
@ -1200,22 +1200,34 @@ class Backend(Database):
).matched_count == 1
def update_user(self, id, **kwargs):
kwargs['updateTime'] = datetime.utcnow()
kwargs = {k: v for k, v in kwargs.items() if v is not None} # backward compatibility for alerta client
update = dict()
update['$set'] = {k: v for k, v in kwargs.items() if k != 'attributes'}
set_value = {'attributes.' + k: v for k, v in kwargs['attributes'].items() if v is not None}
if set_value:
update['$set'].update(set_value)
unset_value = {'attributes.' + k: v for k, v in kwargs['attributes'].items() if v is None}
if unset_value:
update['$unset'] = unset_value
return self.get_db().users.find_one_and_update(
{'_id': id},
update={'$set': kwargs},
return_document=ReturnDocument.AFTER
{'_id': {'$regex': '^' + id}}, update=update, return_document=ReturnDocument.AFTER
)
def update_user_attributes(self, id, old_attrs, new_attrs):
from alerta.utils.collections import merge
merge(old_attrs, new_attrs)
attrs = {k: v for k, v in old_attrs.items() if v is not None}
return self.get_db().users.update_one(
{'_id': {'$regex': '^' + id}},
update={'$set': {'attributes': attrs, 'updateTime': datetime.utcnow()}}
).matched_count == 1
"""
Set all attributes and unset attributes by using a value of 'null'.
"""
update = dict()
set_value = {'attributes.' + k: v for k, v in new_attrs.items() if v is not None}
if set_value:
update['$set'] = set_value
unset_value = {'attributes.' + k: v for k, v in new_attrs.items() if v is None}
if unset_value:
update['$unset'] = unset_value
response = self.get_db().users.update_one({'_id': {'$regex': '^' + id}}, update=update)
return response.matched_count > 0
def delete_user(self, id):
response = self.get_db().users.delete_one({'_id': id})

1
alerta/views/users.py
View file

@ -187,6 +187,7 @@ def update_user_attributes(user_id):
admin_audit_trail.send(current_app._get_current_object(), event='user-attributes-updated', message='', user=g.user,
customers=g.customers, scopes=g.scopes, resource_id=user.id, type='user', request=request)
print(request.json)
if user.update_attributes(request.json['attributes']):
return jsonify(status='ok')
else:

88
tests/test_auth.py
View file

@ -17,7 +17,7 @@ class AuthTestCase(unittest.TestCase):
'AUTH_REQUIRED': True,
'CUSTOMER_VIEWS': True,
'ADMIN_USERS': ['admin@alerta.io'],
'ALLOWED_EMAIL_DOMAINS': ['bonaparte.fr', 'debeauharnais.fr']
'ALLOWED_EMAIL_DOMAINS': ['bonaparte.fr', 'debeauharnais.fr', 'manorfarm.ru']
}
self.app = create_app(test_config)
self.client = self.app.test_client()
@ -351,3 +351,89 @@ class AuthTestCase(unittest.TestCase):
response = self.client.get('/alerts', headers=headers)
data = json.loads(response.data.decode('utf-8'))
self.assertEqual(data['status'], 'ok', response.data)
def test_edit_user(self):
# add customer mapping
payload = {
'customer': 'Manor Farm',
'match': 'manorfarm.ru'
}
response = self.client.post('/customer', data=json.dumps(payload),
content_type='application/json', headers=self.headers)
self.assertEqual(response.status_code, 201)
payload = {
'name': 'Snowball',
'email': 'snowball@manorfarm.ru',
'password': 'Postoronii',
'text': 'Can you not understand that liberty is worth more than ribbons?',
'attributes': {'two-legs': 'bad', 'hasFourLegs': True, 'isEvil': False}
}
# create user
response = self.client.post('/auth/signup', data=json.dumps(payload), content_type='application/json')
self.assertEqual(response.status_code, 200)
data = json.loads(response.data.decode('utf-8'))
with self.app.test_request_context():
jwt = Jwt.parse(data['token'])
user_id = jwt.subject
# get user
response = self.client.get('/user/' + user_id, headers=self.headers)
self.assertEqual(response.status_code, 200)
data = json.loads(response.data.decode('utf-8'))
self.assertEqual(data['status'], 'ok')
self.assertEqual(data['user']['name'], 'Snowball')
self.assertEqual(data['user']['email'], 'snowball@manorfarm.ru')
self.assertEqual(data['user']['text'], 'Can you not understand that liberty is worth more than ribbons?')
# FIXME: attribute keys with None (null) values aren't deleted in postgres
# change user details
update = {
'name': 'Squealer',
'text': 'Four legs good, two legs bad.',
'attributes': {'four-legs': 'good', 'isEvil': True}
}
response = self.client.put('/user/' + user_id, data=json.dumps(update), headers=self.headers)
self.assertEqual(response.status_code, 200)
data = json.loads(response.data.decode('utf-8'))
self.assertEqual(data['status'], 'ok')
# check updates worked and didn't change anything else
response = self.client.get('/user/' + user_id, headers=self.headers)
self.assertEqual(response.status_code, 200)
data = json.loads(response.data.decode('utf-8'))
self.assertEqual(data['user']['name'], 'Squealer')
self.assertEqual(data['user']['email'], 'snowball@manorfarm.ru')
self.assertEqual(data['user']['text'], 'Four legs good, two legs bad.')
self.assertEqual(data['user']['attributes'], {
'four-legs': 'good',
'two-legs': 'bad',
'hasFourLegs': True,
'isEvil': True
})
# just update attributes
update = {
'attributes': {'four-legs': 'double good', 'isEvil': False, 'hasFourLegs': None}
}
response = self.client.put('/user/' + user_id + '/attributes', data=json.dumps(update), headers=self.headers)
self.assertEqual(response.status_code, 200)
data = json.loads(response.data.decode('utf-8'))
self.assertEqual(data['status'], 'ok')
# check updates worked and didn't change anything else
response = self.client.get('/user/' + user_id, headers=self.headers)
self.assertEqual(response.status_code, 200)
data = json.loads(response.data.decode('utf-8'))
self.assertEqual(data['user']['name'], 'Squealer')
self.assertEqual(data['user']['email'], 'snowball@manorfarm.ru')
self.assertEqual(data['user']['text'], 'Four legs good, two legs bad.')
self.assertEqual(data['user']['attributes'], {
'four-legs': 'double good',
'two-legs': 'bad',
'isEvil': False
})