mirror of
https://github.com/GSA/notifications-api.git
synced 2026-02-04 10:21:14 -05:00
Add permissions validation in service schema
This commit is contained in:
@@ -25,8 +25,9 @@ from notifications_utils.recipients import (
|
||||
|
||||
from app import ma
|
||||
from app import models
|
||||
from app.models import ServicePermission
|
||||
from app.models import ServicePermission, INTERNATIONAL_SMS_TYPE, SMS_TYPE
|
||||
from app.dao.permissions_dao import permission_dao
|
||||
from app.dao.service_permissions_dao import dao_fetch_service_permissions
|
||||
from app.utils import get_template_instance
|
||||
|
||||
|
||||
@@ -180,11 +181,10 @@ class ServiceSchema(BaseSchema):
|
||||
branding = field_for(models.Service, 'branding')
|
||||
dvla_organisation = field_for(models.Service, 'dvla_organisation')
|
||||
permissions = fields.Method("service_permissions")
|
||||
|
||||
|
||||
def service_permissions(self, service):
|
||||
permissions = []
|
||||
|
||||
from app.dao.service_permissions_dao import dao_fetch_service_permissions
|
||||
perms = dao_fetch_service_permissions(service.id)
|
||||
for p in perms:
|
||||
permission = {
|
||||
@@ -216,9 +216,17 @@ class ServiceSchema(BaseSchema):
|
||||
|
||||
@validates('permissions')
|
||||
def validate_permissions(self, value):
|
||||
for v in [val.permission for val in value]:
|
||||
if v not in models.SERVICE_PERMISSION_TYPES:
|
||||
raise ValidationError("Invalid Service Permission: '{}'".format(v))
|
||||
permissions = [v.permission for v in value]
|
||||
for p in permissions:
|
||||
if p not in models.SERVICE_PERMISSION_TYPES:
|
||||
raise ValidationError("Invalid Service Permission: '{}'".format(p))
|
||||
|
||||
if len(set(permissions)) != len(permissions):
|
||||
duplicates = list(set([x for x in permissions if permissions.count(x) > 1]))
|
||||
raise ValueError('Service Permission duplicated: {}'.format(duplicates))
|
||||
|
||||
if INTERNATIONAL_SMS_TYPE in permissions and SMS_TYPE not in permissions:
|
||||
raise ValueError('International SMS must have SMS enabled')
|
||||
|
||||
@pre_load()
|
||||
def format_permissions_for_data_model(self, in_data):
|
||||
|
||||
@@ -129,7 +129,10 @@ def update_service(service_id):
|
||||
|
||||
current_data = dict(service_schema.dump(fetched_service).data.items())
|
||||
current_data.update(request.get_json())
|
||||
update_dict = service_schema.load(current_data).data
|
||||
try:
|
||||
update_dict = service_schema.load(current_data).data
|
||||
except ValueError as e:
|
||||
raise InvalidRequest(str(e), status_code=400)
|
||||
dao_update_service(update_dict)
|
||||
|
||||
if service_going_live:
|
||||
|
||||
@@ -541,6 +541,44 @@ def test_update_permissions_with_an_invalid_permission_will_raise_error(client,
|
||||
assert "Invalid Service Permission: '{}'".format(invalid_permission) in result['message']['permissions']
|
||||
|
||||
|
||||
def test_update_permissions_with_duplicate_permissions_will_raise_error(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {
|
||||
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_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 == 400
|
||||
assert result['result'] == 'error'
|
||||
assert "Service Permission duplicated: ['{}']".format(LETTER_TYPE) in result['message']
|
||||
|
||||
|
||||
def test_update_permissions_with_international_sms_without_sms_permissions_will_raise_error(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
|
||||
data = {
|
||||
'permissions': [EMAIL_TYPE, INTERNATIONAL_SMS_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 == 400
|
||||
assert result['result'] == 'error'
|
||||
assert "International SMS must have SMS enabled" == result['message']
|
||||
|
||||
|
||||
def test_update_service_flags_can_remove_service_permissions(client, sample_service):
|
||||
auth_header = create_authorization_header()
|
||||
initial_data = {
|
||||
|
||||
Reference in New Issue
Block a user