mirror of
https://github.com/GSA/notifications-api.git
synced 2026-08-11 09:27:56 -04:00
Merge branch 'master' into invitation
Conflicts: scripts/run_celery.sh
This commit is contained in:
@@ -316,7 +316,8 @@ def sample_invited_user(notify_db,
|
||||
data = {
|
||||
'service': service,
|
||||
'email_address': to_email_address,
|
||||
'from_user': from_user
|
||||
'from_user': from_user,
|
||||
'permissions': 'send_messages,manage_service,manage_api_keys'
|
||||
}
|
||||
invited_user = InvitedUser(**data)
|
||||
save_invited_user(invited_user)
|
||||
@@ -341,3 +342,24 @@ def sample_permission(notify_db,
|
||||
db.session.add(p_model)
|
||||
db.session.commit()
|
||||
return p_model
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def sample_service_permission(notify_db,
|
||||
notify_db_session,
|
||||
service=None,
|
||||
user=None,
|
||||
permission="sample permission"):
|
||||
if user is None:
|
||||
user = sample_user(notify_db, notify_db_session)
|
||||
if service is None:
|
||||
service = sample_service(notify_db, notify_db_session)
|
||||
data = {
|
||||
'user': user,
|
||||
'service': service,
|
||||
'permission': permission
|
||||
}
|
||||
p_model = Permission(**data)
|
||||
db.session.add(p_model)
|
||||
db.session.commit()
|
||||
return p_model
|
||||
|
||||
@@ -17,7 +17,8 @@ def test_create_invited_user(notify_db, notify_db_session, sample_service):
|
||||
data = {
|
||||
'service': sample_service,
|
||||
'email_address': email_address,
|
||||
'from_user': invite_from
|
||||
'from_user': invite_from,
|
||||
'permissions': 'send_messages,manage_service'
|
||||
}
|
||||
|
||||
invited_user = InvitedUser(**data)
|
||||
@@ -26,6 +27,10 @@ def test_create_invited_user(notify_db, notify_db_session, sample_service):
|
||||
assert InvitedUser.query.count() == 1
|
||||
assert invited_user.email_address == email_address
|
||||
assert invited_user.from_user == invite_from
|
||||
permissions = invited_user.get_permissions()
|
||||
assert len(permissions) == 2
|
||||
assert 'send_messages' in permissions
|
||||
assert 'manage_service' in permissions
|
||||
|
||||
|
||||
def test_get_invited_user(notify_db, notify_db_session, sample_invited_user):
|
||||
|
||||
@@ -19,7 +19,8 @@ def test_create_invited_user(notify_api, sample_service, mocker):
|
||||
data = {
|
||||
'service': str(sample_service.id),
|
||||
'email_address': email_address,
|
||||
'from_user': invite_from.id
|
||||
'from_user': invite_from.id,
|
||||
'permissions': 'send_messages,manage_service,manage_api_keys'
|
||||
}
|
||||
|
||||
data = json.dumps(data)
|
||||
@@ -41,6 +42,7 @@ def test_create_invited_user(notify_api, sample_service, mocker):
|
||||
assert json_resp['data']['service'] == str(sample_service.id)
|
||||
assert json_resp['data']['email_address'] == email_address
|
||||
assert json_resp['data']['from_user'] == invite_from.id
|
||||
assert json_resp['data']['permissions'] == 'send_messages,manage_service,manage_api_keys'
|
||||
assert json_resp['data']['id']
|
||||
invitation_expiration_days = notify_api.config['INVITATION_EXPIRATION_DAYS']
|
||||
expiry_date = (datetime.now() + timedelta(days=invitation_expiration_days)).replace(hour=0, minute=0,
|
||||
@@ -68,7 +70,8 @@ def test_create_invited_user_invalid_email(notify_api, sample_service, mocker):
|
||||
data = {
|
||||
'service': str(sample_service.id),
|
||||
'email_address': email_address,
|
||||
'from_user': invite_from.id
|
||||
'from_user': invite_from.id,
|
||||
'permissions': 'send_messages,manage_service,manage_api_keys'
|
||||
}
|
||||
|
||||
data = json.dumps(data)
|
||||
|
||||
@@ -50,7 +50,6 @@ def test_get_job_with_invalid_job_id_returns404(notify_api, sample_template):
|
||||
response = client.get(path, headers=[auth_header])
|
||||
assert response.status_code == 404
|
||||
resp_json = json.loads(response.get_data(as_text=True))
|
||||
print(resp_json)
|
||||
assert resp_json['result'] == 'error'
|
||||
assert resp_json['message'] == 'No result found'
|
||||
|
||||
@@ -218,7 +217,6 @@ def test_get_update_job(notify_api, sample_job):
|
||||
|
||||
def _setup_jobs(notify_db, notify_db_session, template, number_of_jobs=5):
|
||||
for i in range(number_of_jobs):
|
||||
print(i)
|
||||
create_job(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
|
||||
@@ -53,7 +53,6 @@ def test_get_permission_filter(notify_api,
|
||||
headers=[header])
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert len(json_resp['data']) == 1
|
||||
expected = {
|
||||
"permission": another_permission.permission,
|
||||
"user": sample_user.id,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import json
|
||||
import uuid
|
||||
from flask import url_for
|
||||
from app.dao.users_dao import save_model_user
|
||||
from app.dao.services_dao import dao_remove_user_from_service
|
||||
from app.models import User
|
||||
@@ -403,3 +404,55 @@ def test_get_users_for_service_returns_404_when_service_does_not_exist(notify_ap
|
||||
result = json.loads(response.get_data(as_text=True))
|
||||
assert result['result'] == 'error'
|
||||
assert result['message'] == 'Service not found for id: {}'.format(service_id)
|
||||
|
||||
|
||||
def test_default_permissions_are_added_for_user_service(notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service,
|
||||
sample_user):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
data = {
|
||||
'name': 'created service',
|
||||
'user_id': sample_user.id,
|
||||
'limit': 1000,
|
||||
'restricted': False,
|
||||
'active': False}
|
||||
auth_header = create_authorization_header(
|
||||
path='/service',
|
||||
method='POST',
|
||||
request_body=json.dumps(data)
|
||||
)
|
||||
headers = [('Content-Type', 'application/json'), auth_header]
|
||||
resp = client.post(
|
||||
'/service',
|
||||
data=json.dumps(data),
|
||||
headers=headers)
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 201
|
||||
assert json_resp['data']['id']
|
||||
assert json_resp['data']['name'] == 'created service'
|
||||
assert json_resp['data']['email_from'] == 'created.service'
|
||||
|
||||
auth_header_fetch = create_authorization_header(
|
||||
path='/service/{}'.format(json_resp['data']['id']),
|
||||
method='GET'
|
||||
)
|
||||
|
||||
resp = client.get(
|
||||
'/service/{}?user_id={}'.format(json_resp['data']['id'], sample_user.id),
|
||||
headers=[auth_header_fetch]
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
header = create_authorization_header(
|
||||
path=url_for('user.get_user', user_id=sample_user.id),
|
||||
method='GET')
|
||||
response = client.get(
|
||||
url_for('user.get_user', user_id=sample_user.id),
|
||||
headers=[header])
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
service_permissions = json_resp['data']['permissions'][str(sample_service.id)]
|
||||
from app.dao.permissions_dao import default_service_permissions
|
||||
assert sorted(default_service_permissions) == sorted(service_permissions)
|
||||
|
||||
@@ -28,8 +28,12 @@ def test_get_user_list(notify_api, notify_db, notify_db_session, sample_user, sa
|
||||
"password_changed_at": None,
|
||||
"logged_in_at": None,
|
||||
"state": "active",
|
||||
"failed_login_count": 0
|
||||
"failed_login_count": 0,
|
||||
"permissions": {
|
||||
str(sample_admin_service_id): [
|
||||
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']}
|
||||
}
|
||||
print(json_resp['data'])
|
||||
assert expected in json_resp['data']
|
||||
|
||||
|
||||
@@ -55,7 +59,10 @@ def test_get_user(notify_api, notify_db, notify_db_session, sample_user, sample_
|
||||
"password_changed_at": None,
|
||||
"logged_in_at": None,
|
||||
"state": "active",
|
||||
"failed_login_count": 0
|
||||
"failed_login_count": 0,
|
||||
"permissions": {
|
||||
str(sample_admin_service_id): [
|
||||
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']}
|
||||
}
|
||||
assert json_resp['data'] == expected
|
||||
|
||||
@@ -75,7 +82,8 @@ def test_post_user(notify_api, notify_db, notify_db_session, sample_admin_servic
|
||||
"password_changed_at": None,
|
||||
"logged_in_at": None,
|
||||
"state": "active",
|
||||
"failed_login_count": 0
|
||||
"failed_login_count": 0,
|
||||
"permissions": {}
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_admin_service_id,
|
||||
path=url_for('user.create_user'),
|
||||
@@ -108,7 +116,8 @@ def test_post_user_missing_attribute_email(notify_api, notify_db, notify_db_sess
|
||||
"password_changed_at": None,
|
||||
"logged_in_at": None,
|
||||
"state": "active",
|
||||
"failed_login_count": 0
|
||||
"failed_login_count": 0,
|
||||
"permissions": {}
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_admin_service_id,
|
||||
path=url_for('user.create_user'),
|
||||
@@ -139,7 +148,8 @@ def test_post_user_missing_attribute_password(notify_api, notify_db, notify_db_s
|
||||
"password_changed_at": None,
|
||||
"logged_in_at": None,
|
||||
"state": "active",
|
||||
"failed_login_count": 0
|
||||
"failed_login_count": 0,
|
||||
"permissions": {}
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_admin_service_id,
|
||||
path=url_for('user.create_user'),
|
||||
@@ -190,7 +200,10 @@ def test_put_user(notify_api, notify_db, notify_db_session, sample_user, sample_
|
||||
"id": user.id,
|
||||
"logged_in_at": None,
|
||||
"state": "active",
|
||||
"failed_login_count": 0
|
||||
"failed_login_count": 0,
|
||||
"permissions": {
|
||||
str(sample_admin_service_id): [
|
||||
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']}
|
||||
}
|
||||
assert json_resp['data'] == expected
|
||||
assert json_resp['data']['email_address'] == new_email
|
||||
@@ -287,7 +300,10 @@ def test_get_user_by_email(notify_api, notify_db, notify_db_session, sample_user
|
||||
"password_changed_at": None,
|
||||
"logged_in_at": None,
|
||||
"state": "active",
|
||||
"failed_login_count": 0
|
||||
"failed_login_count": 0,
|
||||
"permissions": {
|
||||
str(sample_admin_service_id): [
|
||||
'manage_service', 'send_messages', 'manage_api_keys', 'manage_templates']}
|
||||
}
|
||||
|
||||
assert json_resp['data'] == expected
|
||||
@@ -327,3 +343,19 @@ def test_get_user_by_email_bad_url_returns_404(notify_api,
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert json_resp['result'] == 'error'
|
||||
assert json_resp['message'] == 'invalid request'
|
||||
|
||||
|
||||
def test_get_user_with_permissions(notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service_permission):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
header = create_authorization_header(
|
||||
path=url_for('user.get_user', user_id=sample_service_permission.user.id),
|
||||
method='GET')
|
||||
response = client.get(url_for('user.get_user', user_id=sample_service_permission.user.id),
|
||||
headers=[header])
|
||||
assert response.status_code == 200
|
||||
permissions = json.loads(response.get_data(as_text=True))['data']['permissions']
|
||||
assert sample_service_permission.permission in permissions[str(sample_service_permission.service.id)]
|
||||
|
||||
@@ -358,7 +358,6 @@ def test_send_user_sms_code(notify_api,
|
||||
url_for('user.send_user_sms_code', user_id=sample_sms_code.user.id),
|
||||
data=data,
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
print(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 204
|
||||
app.celery.tasks.send_sms_code.apply_async.assert_called_once_with(['something_encrypted'],
|
||||
queue='sms-code')
|
||||
@@ -427,7 +426,6 @@ def test_send_user_email_code(notify_api,
|
||||
url_for('user.send_user_email_code', user_id=sample_email_code.user.id),
|
||||
data=data,
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
print(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 204
|
||||
app.celery.tasks.send_email_code.apply_async.assert_called_once_with(['something_encrypted'],
|
||||
queue='email-code')
|
||||
|
||||
Reference in New Issue
Block a user