mirror of
https://github.com/GSA/notifications-api.git
synced 2026-07-27 19:29:24 -04:00
Updated service DAO and API end points
This commit is contained in:
@@ -11,6 +11,7 @@ from app.dao.services_dao import (
|
||||
dao_create_service,
|
||||
dao_add_user_to_service,
|
||||
dao_remove_user_from_service,
|
||||
dao_remove_service_permission as dao_services_remove_service_permission,
|
||||
dao_fetch_all_services,
|
||||
dao_fetch_service_by_id,
|
||||
dao_fetch_all_services_by_user,
|
||||
@@ -45,7 +46,10 @@ from app.models import (
|
||||
InvitedUser,
|
||||
Service,
|
||||
ServicePermission,
|
||||
<<<<<<< HEAD
|
||||
ServicePermissionTypes,
|
||||
=======
|
||||
>>>>>>> Updated service DAO and API end points
|
||||
BRANDING_GOVUK,
|
||||
DVLA_ORG_HM_GOVERNMENT,
|
||||
KEY_TYPE_NORMAL,
|
||||
@@ -372,6 +376,40 @@ def test_update_service_creates_a_history_record_with_current_data(sample_user):
|
||||
assert Service.get_history_model().query.filter_by(name='updated_service_name').one().version == 2
|
||||
|
||||
|
||||
def test_update_service_permission_creates_a_history_record_with_current_data(sample_user):
|
||||
assert Service.query.count() == 0
|
||||
assert Service.get_history_model().query.count() == 0
|
||||
service = Service(name="service_name",
|
||||
email_from="email_from",
|
||||
message_limit=1000,
|
||||
restricted=False,
|
||||
created_by=sample_user)
|
||||
dao_create_service(service, sample_user)
|
||||
|
||||
service.permissions.append(ServicePermission(service_id=service.id, permission='letter'))
|
||||
dao_update_service(service)
|
||||
|
||||
assert Service.query.count() == 1
|
||||
assert Service.get_history_model().query.count() == 2
|
||||
|
||||
service_from_db = Service.query.first()
|
||||
|
||||
assert service_from_db.version == 2
|
||||
assert LETTER_TYPE in [p.permission for p in service_from_db.permissions]
|
||||
|
||||
dao_services_remove_service_permission(service, permission='sms')
|
||||
|
||||
assert Service.query.count() == 1
|
||||
assert Service.get_history_model().query.count() == 3
|
||||
|
||||
service_from_db = Service.query.first()
|
||||
assert service_from_db.version == 3
|
||||
assert SMS_TYPE not in [p.permission for p in service_from_db.permissions]
|
||||
|
||||
assert len(Service.get_history_model().query.filter_by(name='service_name').all()) == 3
|
||||
assert Service.get_history_model().query.filter_by(name='service_name').all()[2].version == 3
|
||||
|
||||
|
||||
def test_create_service_and_history_is_transactional(sample_user):
|
||||
assert Service.query.count() == 0
|
||||
assert Service.get_history_model().query.count() == 0
|
||||
|
||||
@@ -10,7 +10,7 @@ from freezegun import freeze_time
|
||||
|
||||
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, Organisation, DVLA_ORG_LAND_REGISTRY, Rate
|
||||
from app.models import User, Organisation, DVLA_ORG_LAND_REGISTRY, Rate, ServicePermission
|
||||
from tests import create_authorization_header
|
||||
from tests.app.db import create_template
|
||||
from tests.app.conftest import (
|
||||
@@ -20,28 +20,26 @@ from tests.app.conftest import (
|
||||
sample_notification_history as create_notification_history,
|
||||
sample_notification_with_job
|
||||
)
|
||||
from app.models import Service, KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST
|
||||
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST, EMAIL_TYPE, SMS_TYPE, LETTER_TYPE
|
||||
|
||||
from tests.app.db import create_user
|
||||
|
||||
|
||||
def test_get_service_list(notify_api, service_factory):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
service_factory.get('one')
|
||||
service_factory.get('two')
|
||||
service_factory.get('three')
|
||||
auth_header = create_authorization_header()
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers=[auth_header]
|
||||
)
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert len(json_resp['data']) == 3
|
||||
assert json_resp['data'][0]['name'] == 'one'
|
||||
assert json_resp['data'][1]['name'] == 'two'
|
||||
assert json_resp['data'][2]['name'] == 'three'
|
||||
def test_get_service_list(client, service_factory):
|
||||
service_factory.get('one')
|
||||
service_factory.get('two')
|
||||
service_factory.get('three')
|
||||
auth_header = create_authorization_header()
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers=[auth_header]
|
||||
)
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert len(json_resp['data']) == 3
|
||||
assert json_resp['data'][0]['name'] == 'one'
|
||||
assert json_resp['data'][1]['name'] == 'two'
|
||||
assert json_resp['data'][2]['name'] == 'three'
|
||||
|
||||
|
||||
def test_get_service_list_with_only_active_flag(client, service_factory):
|
||||
@@ -117,17 +115,15 @@ def test_get_service_list_by_user_should_return_empty_list_if_no_services(client
|
||||
assert len(json_resp['data']) == 0
|
||||
|
||||
|
||||
def test_get_service_list_should_return_empty_list_if_no_services(notify_api, notify_db, notify_db_session):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
auth_header = create_authorization_header()
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers=[auth_header]
|
||||
)
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert len(json_resp['data']) == 0
|
||||
def test_get_service_list_should_return_empty_list_if_no_services(client):
|
||||
auth_header = create_authorization_header()
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers=[auth_header]
|
||||
)
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert len(json_resp['data']) == 0
|
||||
|
||||
|
||||
def test_get_service_by_id(client, sample_service):
|
||||
@@ -147,6 +143,32 @@ def test_get_service_by_id(client, sample_service):
|
||||
assert json_resp['data']['sms_sender'] == current_app.config['FROM_NUMBER']
|
||||
|
||||
|
||||
def test_get_service_list_has_default_permissions(client, service_factory):
|
||||
service_factory.get('one')
|
||||
service_factory.get('two')
|
||||
service_factory.get('three')
|
||||
auth_header = create_authorization_header()
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers=[auth_header]
|
||||
)
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert len(json_resp['data']) == 3
|
||||
assert all([set(json['permissions']) == set([EMAIL_TYPE, SMS_TYPE]) for json in json_resp['data']])
|
||||
|
||||
|
||||
def test_get_service_by_id_has_default_permissions(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
resp = client.get(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
headers=[auth_header]
|
||||
)
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
|
||||
assert set(json_resp['data']['permissions']) == set([EMAIL_TYPE, SMS_TYPE])
|
||||
|
||||
|
||||
def test_get_service_by_id_should_404_if_no_service(notify_api, notify_db):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -414,39 +436,140 @@ def test_update_service(client, notify_db, sample_service):
|
||||
assert result['data']['dvla_organisation'] == DVLA_ORG_LAND_REGISTRY
|
||||
|
||||
|
||||
def test_update_service_flags(notify_api, sample_service):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
auth_header = create_authorization_header()
|
||||
resp = client.get(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
headers=[auth_header]
|
||||
)
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 200
|
||||
assert json_resp['data']['name'] == sample_service.name
|
||||
assert json_resp['data']['research_mode'] is False
|
||||
assert json_resp['data']['can_send_letters'] is False
|
||||
assert json_resp['data']['can_send_international_sms'] is False
|
||||
def test_update_service_flags(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
resp = client.get(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
headers=[auth_header]
|
||||
)
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 200
|
||||
assert json_resp['data']['name'] == sample_service.name
|
||||
assert json_resp['data']['research_mode'] is False
|
||||
assert json_resp['data']['can_send_letters'] is False
|
||||
assert json_resp['data']['can_send_international_sms'] is False
|
||||
|
||||
data = {
|
||||
'research_mode': True,
|
||||
'can_send_letters': True,
|
||||
'can_send_international_sms': True,
|
||||
}
|
||||
data = {
|
||||
'research_mode': True,
|
||||
'can_send_letters': True,
|
||||
'can_send_international_sms': True,
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header()
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 200
|
||||
assert result['data']['research_mode'] is True
|
||||
assert result['data']['can_send_letters'] is True
|
||||
assert result['data']['can_send_international_sms'] is True
|
||||
resp = client.post(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 200
|
||||
assert result['data']['research_mode'] is True
|
||||
assert result['data']['can_send_letters'] is True
|
||||
assert result['data']['can_send_international_sms'] is True
|
||||
|
||||
|
||||
def test_update_service_flags_will_add_service_permissions(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
data = {
|
||||
'research_mode': True,
|
||||
'can_send_letters': True,
|
||||
'can_send_international_sms': True,
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert result['data']['research_mode'] is True
|
||||
assert result['data']['can_send_letters'] is True
|
||||
assert result['data']['can_send_international_sms'] is True
|
||||
|
||||
|
||||
def test_update_permissions_can_add_service_permissions(client, sample_service):
|
||||
from app.models import ServicePermission
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {
|
||||
'research_mode': True,
|
||||
'can_send_letters': True,
|
||||
'can_send_international_sms': True,
|
||||
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE]
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert result['data']['research_mode'] is True
|
||||
assert result['data']['can_send_letters'] is True
|
||||
assert result['data']['can_send_international_sms'] is True
|
||||
assert all(set(result['data']['permissions']) & set([SMS_TYPE, EMAIL_TYPE, LETTER_TYPE]))
|
||||
|
||||
|
||||
def test_update_permissions_with_an_invalid_permission_will_raise_error(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
invalid_permission = 'invalid_permission'
|
||||
|
||||
data = {
|
||||
'research_mode': True,
|
||||
'can_send_letters': True,
|
||||
'can_send_international_sms': True,
|
||||
'permissions': [EMAIL_TYPE, SMS_TYPE, invalid_permission]
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
|
||||
assert resp.status_code == 400
|
||||
assert result['result'] == 'error'
|
||||
assert "Invalid Service Permission: '{}'".format(invalid_permission) in result['message']['permissions']
|
||||
|
||||
|
||||
def test_update_service_flags_can_remove_service_permissions(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
initial_data = {
|
||||
'research_mode': True,
|
||||
'can_send_letters': True,
|
||||
'can_send_international_sms': True,
|
||||
'permissions': [EMAIL_TYPE, SMS_TYPE]
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
data=json.dumps(initial_data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
|
||||
remove_permission_data = {
|
||||
'can_send_letters': True,
|
||||
'can_send_international_sms': False,
|
||||
'permissions': [EMAIL_TYPE]
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
'/service/{}'.format(sample_service.id),
|
||||
data=json.dumps(remove_permission_data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert result['data']['can_send_letters'] is True
|
||||
assert result['data']['can_send_international_sms'] is False
|
||||
assert result['data']['permissions'] == [EMAIL_TYPE]
|
||||
|
||||
|
||||
def test_update_service_research_mode_throws_validation_error(notify_api, sample_service):
|
||||
|
||||
Reference in New Issue
Block a user