mirror of
https://github.com/alerta/alerta.git
synced 2024-11-05 08:37:12 +00:00
591 lines
23 KiB
Python
591 lines
23 KiB
Python
import base64
|
|
import json
|
|
import unittest
|
|
|
|
from mohawk import Sender
|
|
|
|
from alerta.app import create_app, db, plugins
|
|
from alerta.models.enums import Scope
|
|
from alerta.models.key import ApiKey
|
|
from alerta.models.token import Jwt
|
|
|
|
|
|
class AuthTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
|
|
self.access_key = 'cc3b7f30-360e-47bc-8abb-c0a27625e134'
|
|
self.secret_key = 'MjM0ODU4NGI1YWQxZWMyYzcxNjAxZDA4MzczNGQ1M2IK'
|
|
|
|
test_config = {
|
|
'DEBUG': True,
|
|
'TESTING': True,
|
|
'AUTH_REQUIRED': True,
|
|
'CUSTOMER_VIEWS': True,
|
|
'ADMIN_USERS': ['admin@alerta.io'],
|
|
'DELETE_SCOPES': ['delete:alerts'],
|
|
'ALLOWED_EMAIL_DOMAINS': ['bonaparte.fr', 'debeauharnais.fr', 'manorfarm.ru'],
|
|
'HMAC_AUTH_CREDENTIALS': [
|
|
{
|
|
'key': self.access_key,
|
|
'secret': self.secret_key,
|
|
'algorithm': 'sha256'
|
|
}
|
|
]
|
|
}
|
|
self.app = create_app(test_config)
|
|
self.client = self.app.test_client()
|
|
|
|
self.alert = {
|
|
'event': 'Foo',
|
|
'resource': 'Bar',
|
|
'environment': 'Production',
|
|
'service': ['Quux']
|
|
}
|
|
|
|
with self.app.test_request_context('/'):
|
|
self.app.preprocess_request()
|
|
self.api_key = ApiKey(
|
|
user='admin@alerta.io',
|
|
scopes=[Scope.admin, Scope.read, Scope.write],
|
|
text='Demo API key - not for production use',
|
|
key='demo-key'
|
|
)
|
|
self.api_key.create()
|
|
|
|
self.headers = {
|
|
'Authorization': f'Key {self.api_key.key}',
|
|
'Content-type': 'application/json'
|
|
}
|
|
|
|
def tearDown(self):
|
|
plugins.plugins.clear()
|
|
db.destroy()
|
|
|
|
def test_401_error(self):
|
|
|
|
response = self.client.get('/alerts')
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
def test_user_defined_key(self):
|
|
|
|
self.assertEqual(self.api_key.key, 'demo-key')
|
|
|
|
def test_admin_key(self):
|
|
|
|
payload = {
|
|
'user': 'rw-demo-key-user',
|
|
'scopes': ['admin']
|
|
}
|
|
|
|
response = self.client.post('/key', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIsNotNone(data['key'], 'Failed to create read-write key')
|
|
|
|
admin_api_key = data['key']
|
|
|
|
response = self.client.post('/alert', data=json.dumps(self.alert),
|
|
content_type='application/json', headers={'Authorization': 'Key ' + admin_api_key})
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['alert']['history'][0]['user'], 'rw-demo-key-user')
|
|
|
|
alert_id = data['id']
|
|
|
|
response = self.client.get('/alerts', headers={'Authorization': 'Key ' + admin_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn('total', data)
|
|
|
|
response = self.client.get('/key/' + admin_api_key, headers={'Authorization': 'Key ' + admin_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['key']['scopes'], ['admin'])
|
|
|
|
# delete alert
|
|
response = self.client.delete('/alert/' + alert_id, headers={'Authorization': 'Key ' + admin_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response = self.client.delete('/key/' + admin_api_key, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_readwrite_key(self):
|
|
|
|
payload = {
|
|
'user': 'rw-demo-key-user',
|
|
'type': 'read-write'
|
|
}
|
|
|
|
response = self.client.post('/key', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIsNotNone(data['key'], 'Failed to create read-write key')
|
|
|
|
rw_api_key = data['key']
|
|
|
|
response = self.client.post('/alert', data=json.dumps(self.alert),
|
|
content_type='application/json', headers={'Authorization': 'Key ' + rw_api_key})
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['alert']['history'][0]['user'], 'rw-demo-key-user')
|
|
|
|
alert_id = data['id']
|
|
|
|
response = self.client.get('/alerts', headers={'Authorization': 'Key ' + rw_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn('total', data)
|
|
|
|
response = self.client.get('/key/' + rw_api_key, headers={'Authorization': 'Key ' + rw_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['key']['scopes'], ['read', 'write'])
|
|
|
|
# delete alert
|
|
response = self.client.delete('/alert/' + alert_id, headers={'Authorization': 'Key ' + rw_api_key})
|
|
self.assertEqual(response.status_code, 403)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['message'], 'Missing required scope: delete:alerts')
|
|
|
|
response = self.client.delete('/key/' + rw_api_key, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_rw_delete_key(self):
|
|
|
|
payload = {
|
|
'user': 'rw-delete-demo-key-user',
|
|
'scopes': ['read', 'write', 'delete:alerts']
|
|
}
|
|
|
|
response = self.client.post('/key', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIsNotNone(data['key'], 'Failed to create read-write-delete key')
|
|
self.assertEqual(data['data']['scopes'], ['read', 'write', 'delete:alerts'])
|
|
|
|
rwd_api_key = data['key']
|
|
|
|
response = self.client.post('/alert', data=json.dumps(self.alert),
|
|
content_type='application/json', headers={'Authorization': 'Key ' + rwd_api_key})
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['alert']['history'][0]['user'], 'rw-delete-demo-key-user')
|
|
|
|
alert_id = data['id']
|
|
|
|
response = self.client.get('/alerts', headers={'Authorization': 'Key ' + rwd_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn('total', data)
|
|
|
|
response = self.client.get('/key/' + rwd_api_key, headers={'Authorization': 'Key ' + rwd_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['key']['scopes'], ['read', 'write', 'delete:alerts'])
|
|
|
|
# delete alert
|
|
response = self.client.delete('/alert/' + alert_id, headers={'Authorization': 'Key ' + rwd_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response = self.client.delete('/key/' + rwd_api_key, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_readonly_key(self):
|
|
|
|
payload = {
|
|
'user': 'ro-demo-key-user',
|
|
'type': 'read-only'
|
|
}
|
|
|
|
response = self.client.post('/key', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIsNotNone(data['key'], 'Failed to create read-only key')
|
|
|
|
ro_api_key = data['key']
|
|
|
|
response = self.client.post('/alert', data=json.dumps(self.alert),
|
|
content_type='application/json', headers={'Authorization': 'Key ' + ro_api_key})
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
response = self.client.get('/alerts', headers={'Authorization': 'Key ' + ro_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn('total', data)
|
|
|
|
response = self.client.delete('/key/' + ro_api_key, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_users(self):
|
|
|
|
# add customer mapping
|
|
payload = {
|
|
'customer': 'Bonaparte Industries',
|
|
'match': 'bonaparte.fr'
|
|
}
|
|
response = self.client.post('/customer', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
payload = {
|
|
'name': 'Napoleon Bonaparte',
|
|
'email': 'napoleon@bonaparte.fr',
|
|
'password': 'blackforest',
|
|
'text': 'added to circle of trust'
|
|
}
|
|
|
|
# create user
|
|
response = self.client.post('/auth/signup', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIsNotNone(data, 'Failed to create user')
|
|
|
|
with self.app.test_request_context():
|
|
jwt = Jwt.parse(data['token'])
|
|
user_id = jwt.subject
|
|
|
|
# get user
|
|
response = self.client.get('/users', headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn(user_id, [u['id'] for u in data['users']])
|
|
|
|
# create duplicate user
|
|
response = self.client.post('/auth/signup', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 409)
|
|
|
|
# delete user
|
|
response = self.client.delete('/user/' + user_id, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_login(self):
|
|
|
|
name = 'Josephine de Beauharnais'
|
|
|
|
payload = {
|
|
'name': name,
|
|
'email': 'josephine@debeauharnais.fr',
|
|
'password': 'jojo',
|
|
'text': 'Test login'
|
|
}
|
|
|
|
# sign-up user with no customer mapping
|
|
response = self.client.post('/auth/signup', data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
# add customer mapping
|
|
payload = {
|
|
'customer': 'Bonaparte Industries',
|
|
'match': 'debeauharnais.fr'
|
|
}
|
|
response = self.client.post('/customer', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
|
|
customer_id = data['id']
|
|
|
|
payload = {
|
|
'email': 'josephine@debeauharnais.fr',
|
|
'password': 'jojo'
|
|
}
|
|
|
|
# login now that customer mapping exists
|
|
response = self.client.post('/auth/login', data=json.dumps(payload), content_type='application/json')
|
|
# self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn('token', data)
|
|
|
|
token = data['token']
|
|
|
|
headers = {
|
|
'Authorization': 'Bearer ' + token,
|
|
'Content-type': 'application/json'
|
|
}
|
|
|
|
# create a customer demo key
|
|
payload = {
|
|
'user': 'customer-demo-key-user',
|
|
'type': 'read-only'
|
|
}
|
|
|
|
response = self.client.post('/key', data=json.dumps(payload), content_type='application/json', headers=headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIsNotNone(data['key'], 'Failed to create read-only key')
|
|
|
|
customer_api_key = data['key']
|
|
|
|
response = self.client.get('/alerts', headers={'Authorization': 'Key ' + customer_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn('total', data)
|
|
|
|
response = self.client.get('/key/' + customer_api_key, headers=headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['key']['scopes'], ['read'])
|
|
|
|
response = self.client.delete('/key/' + customer_api_key, headers={'Authorization': 'Key ' + customer_api_key})
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
response = self.client.delete('/key/' + customer_api_key, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# get user
|
|
response = self.client.get('/users', headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn(name, [u['name'] for u in data['users']])
|
|
|
|
user_id = [u['id'] for u in data['users'] if u['name'] == name][0]
|
|
|
|
# delete user
|
|
response = self.client.delete('/user/' + user_id, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# delete customer mapping
|
|
response = self.client.delete('/customer/' + customer_id, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_x_api_key(self):
|
|
|
|
self.headers = {
|
|
'X-API-Key': self.api_key.key,
|
|
'Content-type': 'application/json'
|
|
}
|
|
|
|
payload = {
|
|
'user': 'rw-demo-key-user',
|
|
'type': 'read-write'
|
|
}
|
|
|
|
response = self.client.post('/key', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIsNotNone(data['key'], 'Failed to create read-write key')
|
|
|
|
rw_api_key = data['key']
|
|
|
|
response = self.client.post('/alert', data=json.dumps(self.alert),
|
|
content_type='application/json', headers={'X-API-Key': rw_api_key})
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
response = self.client.get('/alerts', headers={'X-API-Key': rw_api_key})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertIn('total', data)
|
|
|
|
response = self.client.delete('/key/' + rw_api_key, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_edit_api_keys(self):
|
|
|
|
self.headers = {
|
|
'Authorization': f'Key {self.api_key.key}',
|
|
'Content-type': 'application/json'
|
|
}
|
|
|
|
# create api key
|
|
payload = {
|
|
'scopes': [Scope.read, Scope.write_alerts, Scope.write_blackouts],
|
|
'text': 'devops automation key',
|
|
'expireTime': '2099-12-31T23:59:59.999Z'
|
|
}
|
|
response = self.client.post('/key', data=json.dumps(payload), headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
|
|
api_key_id = data['key']
|
|
|
|
# extend blackout period & change environment
|
|
update = {
|
|
'scopes': [Scope.read, Scope.write_blackouts, Scope.write_webhooks],
|
|
'expireTime': '2022-12-31T23:59:59.999Z'
|
|
}
|
|
response = self.client.put('/key/' + api_key_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('/key/' + api_key_id, headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['key']['scopes'], [Scope.read, Scope.write_blackouts, Scope.write_webhooks])
|
|
self.assertEqual(data['key']['user'], 'admin@alerta.io')
|
|
self.assertEqual(data['key']['text'], 'devops automation key')
|
|
self.assertEqual(data['key']['expireTime'], '2022-12-31T23:59:59.999Z')
|
|
|
|
def test_basic_auth(self):
|
|
|
|
# add customer mapping
|
|
payload = {
|
|
'customer': 'Bonaparte Industries',
|
|
'match': 'bonaparte.fr'
|
|
}
|
|
response = self.client.post('/customer', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
payload = {
|
|
'name': 'Napoleon Bonaparte',
|
|
'email': 'napoleon@bonaparte.fr',
|
|
'password': 'blackforest',
|
|
'text': 'added to circle of trust'
|
|
}
|
|
|
|
# create user
|
|
response = self.client.post('/auth/signup', data=json.dumps(payload),
|
|
content_type='application/json', headers=self.headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# authenticate using basic auth
|
|
headers = {
|
|
'Authorization': 'Basic ' + base64.b64encode(b'napoleon@bonaparte.fr:blackforest').decode(),
|
|
'Content-type': 'application/json'
|
|
}
|
|
|
|
response = self.client.get('/users', headers=headers)
|
|
self.assertEqual(response.status_code, 403)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['message'], 'Missing required scope: admin:users')
|
|
|
|
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
|
|
})
|
|
|
|
def test_hmac_auth(self):
|
|
|
|
credentials = {
|
|
'id': self.access_key,
|
|
'key': self.secret_key,
|
|
'algorithm': 'sha256'
|
|
}
|
|
|
|
sender = Sender(
|
|
url='http://localhost/alerts',
|
|
method='GET',
|
|
content='',
|
|
content_type='application/json',
|
|
credentials=credentials
|
|
)
|
|
headers = {
|
|
'Authorization': sender.request_header,
|
|
'Content-Type': 'application/json'
|
|
}
|
|
response = self.client.get('/alerts', headers=headers)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
|
|
sender = Sender(
|
|
url='http://localhost/alert',
|
|
method='POST',
|
|
content=json.dumps(self.alert),
|
|
content_type='application/json',
|
|
credentials=credentials
|
|
)
|
|
headers = {
|
|
'Authorization': sender.request_header,
|
|
'Content-Type': 'application/json'
|
|
}
|
|
response = self.client.post('/alert', data=json.dumps(self.alert),
|
|
content_type='application/json', headers=headers)
|
|
self.assertEqual(response.status_code, 201)
|
|
data = json.loads(response.data.decode('utf-8'))
|
|
self.assertEqual(data['alert']['event'], 'Foo')
|