mirror of
https://github.com/GSA/notifications-api.git
synced 2026-02-04 02:11:11 -05:00
Add default permissions when creating a service
This commit is contained in:
@@ -9,13 +9,14 @@ def dao_fetch_service_permissions(service_id):
|
|||||||
|
|
||||||
|
|
||||||
@transactional
|
@transactional
|
||||||
def dao_create_service_permission(service_id, permission):
|
def dao_add_service_permission(service_id, permission):
|
||||||
service_permission = ServicePermission(service_id=service_id, permission=permission)
|
service_permission = ServicePermission(service_id=service_id, permission=permission)
|
||||||
|
|
||||||
db.session.add(service_permission)
|
db.session.add(service_permission)
|
||||||
|
|
||||||
|
|
||||||
def dao_remove_service_permission(service_id, permission):
|
def dao_remove_service_permission(service_id, permission):
|
||||||
return ServicePermission.query.filter(
|
deleted = ServicePermission.query.filter(
|
||||||
ServicePermission.service_id == service_id,
|
ServicePermission.service_id == service_id,
|
||||||
ServicePermission.permission == permission).delete()
|
ServicePermission.permission == permission).delete()
|
||||||
|
db.session.commit()
|
||||||
|
return deleted
|
||||||
|
|||||||
@@ -25,9 +25,12 @@ from app.models import (
|
|||||||
User,
|
User,
|
||||||
InvitedUser,
|
InvitedUser,
|
||||||
Service,
|
Service,
|
||||||
|
ServicePermission,
|
||||||
KEY_TYPE_TEST,
|
KEY_TYPE_TEST,
|
||||||
NOTIFICATION_STATUS_TYPES,
|
NOTIFICATION_STATUS_TYPES,
|
||||||
TEMPLATE_TYPES,
|
TEMPLATE_TYPES,
|
||||||
|
SMS_TYPE,
|
||||||
|
EMAIL_TYPE
|
||||||
)
|
)
|
||||||
from app.service.statistics import format_monthly_template_notification_stats
|
from app.service.statistics import format_monthly_template_notification_stats
|
||||||
from app.statsd_decorators import statsd
|
from app.statsd_decorators import statsd
|
||||||
@@ -124,13 +127,18 @@ def dao_fetch_service_by_id_and_user(service_id, user_id):
|
|||||||
|
|
||||||
@transactional
|
@transactional
|
||||||
@version_class(Service)
|
@version_class(Service)
|
||||||
def dao_create_service(service, user, service_id=None):
|
def dao_create_service(service, user, service_id=None, service_permissions=[SMS_TYPE, EMAIL_TYPE]):
|
||||||
from app.dao.permissions_dao import permission_dao
|
from app.dao.permissions_dao import permission_dao
|
||||||
service.users.append(user)
|
service.users.append(user)
|
||||||
permission_dao.add_default_service_permissions_for_user(user, service)
|
permission_dao.add_default_service_permissions_for_user(user, service)
|
||||||
service.id = service_id or uuid.uuid4() # must be set now so version history model can use same id
|
service.id = service_id or uuid.uuid4() # must be set now so version history model can use same id
|
||||||
service.active = True
|
service.active = True
|
||||||
service.research_mode = False
|
service.research_mode = False
|
||||||
|
|
||||||
|
for permission in service_permissions:
|
||||||
|
service_permission = ServicePermission(service_id=service.id, permission=permission)
|
||||||
|
db.session.add(service_permission)
|
||||||
|
|
||||||
db.session.add(service)
|
db.session.add(service)
|
||||||
|
|
||||||
|
|
||||||
@@ -185,6 +193,7 @@ def delete_service_and_all_associated_db_objects(service):
|
|||||||
_delete_commit(Job.query.filter_by(service=service))
|
_delete_commit(Job.query.filter_by(service=service))
|
||||||
_delete_commit(Template.query.filter_by(service=service))
|
_delete_commit(Template.query.filter_by(service=service))
|
||||||
_delete_commit(TemplateHistory.query.filter_by(service_id=service.id))
|
_delete_commit(TemplateHistory.query.filter_by(service_id=service.id))
|
||||||
|
_delete_commit(ServicePermission.query.filter_by(service_id=service.id))
|
||||||
|
|
||||||
verify_codes = VerifyCode.query.join(User).filter(User.id.in_([x.id for x in service.users]))
|
verify_codes = VerifyCode.query.join(User).filter(User.id.in_([x.id for x in service.users]))
|
||||||
list(map(db.session.delete, verify_codes))
|
list(map(db.session.delete, verify_codes))
|
||||||
|
|||||||
@@ -143,6 +143,30 @@ class DVLAOrganisation(db.Model):
|
|||||||
name = db.Column(db.String(255), nullable=True)
|
name = db.Column(db.String(255), nullable=True)
|
||||||
|
|
||||||
|
|
||||||
|
INTERNATIONAL_SMS_TYPE = 'international_sms'
|
||||||
|
INCOMING_SMS_TYPE = 'incoming_sms'
|
||||||
|
|
||||||
|
SERVICE_PERMISSION_TYPES = [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INCOMING_SMS_TYPE]
|
||||||
|
|
||||||
|
|
||||||
|
class ServicePermissionTypes(db.Model):
|
||||||
|
__tablename__ = 'service_permission_types'
|
||||||
|
|
||||||
|
name = db.Column(db.String(255), primary_key=True)
|
||||||
|
|
||||||
|
|
||||||
|
class ServicePermission(db.Model):
|
||||||
|
__tablename__ = "service_permissions"
|
||||||
|
|
||||||
|
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'),
|
||||||
|
primary_key=True, index=True, nullable=False)
|
||||||
|
service = db.relationship('Service')
|
||||||
|
permission = db.Column(db.String(255), db.ForeignKey('service_permission_types.name'),
|
||||||
|
index=True, primary_key=True, nullable=False)
|
||||||
|
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, nullable=False)
|
||||||
|
updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.datetime.utcnow)
|
||||||
|
|
||||||
|
|
||||||
class Service(db.Model, Versioned):
|
class Service(db.Model, Versioned):
|
||||||
__tablename__ = 'services'
|
__tablename__ = 'services'
|
||||||
|
|
||||||
@@ -193,30 +217,13 @@ class Service(db.Model, Versioned):
|
|||||||
nullable=False,
|
nullable=False,
|
||||||
default=BRANDING_GOVUK
|
default=BRANDING_GOVUK
|
||||||
)
|
)
|
||||||
|
permissions = db.relationship('ServicePermission')
|
||||||
|
|
||||||
|
# This is only for backward compatibility and will be dropped when the columns are removed from the data model
|
||||||
INTERNATIONAL_SMS_TYPE = 'international_sms'
|
def set_permissions(self):
|
||||||
INCOMING_SMS_TYPE = 'incoming_sms'
|
if self.permissions:
|
||||||
|
self.can_send_letters = LETTER_TYPE in [p.permission for p in self.permissions]
|
||||||
SERVICE_PERMISSION_TYPES = [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INCOMING_SMS_TYPE]
|
self.can_send_international_sms = INTERNATIONAL_SMS_TYPE in [p.permission for p in self.permissions]
|
||||||
|
|
||||||
|
|
||||||
class ServicePermissionTypes(db.Model):
|
|
||||||
__tablename__ = 'service_permission_types'
|
|
||||||
|
|
||||||
name = db.Column(db.String(255), primary_key=True)
|
|
||||||
|
|
||||||
|
|
||||||
class ServicePermission(db.Model):
|
|
||||||
__tablename__ = "service_permissions"
|
|
||||||
|
|
||||||
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'),
|
|
||||||
primary_key=True, index=True, nullable=False)
|
|
||||||
service = db.relationship('Service')
|
|
||||||
permission = db.Column(db.String(255), db.ForeignKey('service_permission_types.name'),
|
|
||||||
index=True, primary_key=True, nullable=False)
|
|
||||||
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, nullable=False)
|
|
||||||
updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.datetime.utcnow)
|
|
||||||
|
|
||||||
|
|
||||||
MOBILE_TYPE = 'mobile'
|
MOBILE_TYPE = 'mobile'
|
||||||
|
|||||||
@@ -1,36 +1,44 @@
|
|||||||
|
import pytest
|
||||||
|
|
||||||
from app.dao.service_permissions_dao import dao_fetch_service_permissions, dao_remove_service_permission
|
from app.dao.service_permissions_dao import dao_fetch_service_permissions, dao_remove_service_permission
|
||||||
from app.models import EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INCOMING_SMS_TYPE
|
from app.models import EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INCOMING_SMS_TYPE
|
||||||
|
|
||||||
from tests.app.db import create_service_permission
|
from tests.app.db import create_service_permission, create_service
|
||||||
|
|
||||||
|
|
||||||
def test_create_service_permission(sample_service):
|
@pytest.fixture(scope='function')
|
||||||
service_permissions = create_service_permission(service_id=sample_service.id, permission=SMS_TYPE)
|
def service_without_permissions(notify_db, notify_db_session):
|
||||||
|
return create_service(service_permissions=[])
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_service_permission(service_without_permissions):
|
||||||
|
service_permissions = create_service_permission(
|
||||||
|
service_id=service_without_permissions.id, permission=SMS_TYPE)
|
||||||
|
|
||||||
assert len(service_permissions) == 1
|
assert len(service_permissions) == 1
|
||||||
assert service_permissions[0].service_id == sample_service.id
|
assert service_permissions[0].service_id == service_without_permissions.id
|
||||||
assert service_permissions[0].permission == SMS_TYPE
|
assert service_permissions[0].permission == SMS_TYPE
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_service_permissions_gets_service_permissions(sample_service):
|
def test_fetch_service_permissions_gets_service_permissions(service_without_permissions):
|
||||||
create_service_permission(service_id=sample_service.id, permission=LETTER_TYPE)
|
create_service_permission(service_id=service_without_permissions.id, permission=LETTER_TYPE)
|
||||||
create_service_permission(service_id=sample_service.id, permission=INTERNATIONAL_SMS_TYPE)
|
create_service_permission(service_id=service_without_permissions.id, permission=INTERNATIONAL_SMS_TYPE)
|
||||||
create_service_permission(service_id=sample_service.id, permission=SMS_TYPE)
|
create_service_permission(service_id=service_without_permissions.id, permission=SMS_TYPE)
|
||||||
|
|
||||||
service_permissions = dao_fetch_service_permissions(sample_service.id)
|
service_permissions = dao_fetch_service_permissions(service_without_permissions.id)
|
||||||
|
|
||||||
assert len(service_permissions) == 3
|
assert len(service_permissions) == 3
|
||||||
assert all(sp.service_id == sample_service.id for sp in service_permissions)
|
assert all(sp.service_id == service_without_permissions.id for sp in service_permissions)
|
||||||
assert all(sp.permission in [LETTER_TYPE, INTERNATIONAL_SMS_TYPE, SMS_TYPE] for sp in service_permissions)
|
assert all(sp.permission in [LETTER_TYPE, INTERNATIONAL_SMS_TYPE, SMS_TYPE] for sp in service_permissions)
|
||||||
|
|
||||||
|
|
||||||
def test_remove_service_permission(sample_service):
|
def test_remove_service_permission(service_without_permissions):
|
||||||
create_service_permission(service_id=sample_service.id, permission=EMAIL_TYPE)
|
create_service_permission(service_id=service_without_permissions.id, permission=EMAIL_TYPE)
|
||||||
create_service_permission(service_id=sample_service.id, permission=INCOMING_SMS_TYPE)
|
create_service_permission(service_id=service_without_permissions.id, permission=INCOMING_SMS_TYPE)
|
||||||
|
|
||||||
dao_remove_service_permission(sample_service.id, EMAIL_TYPE)
|
dao_remove_service_permission(service_without_permissions.id, EMAIL_TYPE)
|
||||||
|
|
||||||
permissions = dao_fetch_service_permissions(sample_service.id)
|
permissions = dao_fetch_service_permissions(service_without_permissions.id)
|
||||||
assert len(permissions) == 1
|
assert len(permissions) == 1
|
||||||
assert permissions[0].permission == INCOMING_SMS_TYPE
|
assert permissions[0].permission == INCOMING_SMS_TYPE
|
||||||
assert permissions[0].service_id == sample_service.id
|
assert permissions[0].service_id == service_without_permissions.id
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ from app.dao.services_dao import (
|
|||||||
dao_resume_service,
|
dao_resume_service,
|
||||||
dao_fetch_active_users_for_service
|
dao_fetch_active_users_for_service
|
||||||
)
|
)
|
||||||
|
from app.dao.service_permissions_dao import dao_add_service_permission, dao_remove_service_permission
|
||||||
from app.dao.users_dao import save_model_user
|
from app.dao.users_dao import save_model_user
|
||||||
from app.models import (
|
from app.models import (
|
||||||
NotificationStatistics,
|
NotificationStatistics,
|
||||||
@@ -47,7 +48,11 @@ from app.models import (
|
|||||||
DVLA_ORG_HM_GOVERNMENT,
|
DVLA_ORG_HM_GOVERNMENT,
|
||||||
KEY_TYPE_NORMAL,
|
KEY_TYPE_NORMAL,
|
||||||
KEY_TYPE_TEAM,
|
KEY_TYPE_TEAM,
|
||||||
KEY_TYPE_TEST
|
KEY_TYPE_TEST,
|
||||||
|
EMAIL_TYPE,
|
||||||
|
SMS_TYPE,
|
||||||
|
LETTER_TYPE,
|
||||||
|
INTERNATIONAL_SMS_TYPE
|
||||||
)
|
)
|
||||||
|
|
||||||
from tests.app.db import create_user, create_service
|
from tests.app.db import create_user, create_service
|
||||||
@@ -245,6 +250,62 @@ def test_get_service_by_id_returns_service(service_factory):
|
|||||||
assert dao_fetch_service_by_id(service.id).name == 'testing'
|
assert dao_fetch_service_by_id(service.id).name == 'testing'
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_service_returns_service_with_default_permissions(service_factory):
|
||||||
|
service = service_factory.get('testing', email_from='testing')
|
||||||
|
|
||||||
|
service = dao_fetch_service_by_id(service.id)
|
||||||
|
assert len(service.permissions) == 2
|
||||||
|
assert all(p.permission in [SMS_TYPE, EMAIL_TYPE] for p in service.permissions)
|
||||||
|
|
||||||
|
|
||||||
|
# This test is only for backward compatibility and will be removed
|
||||||
|
# when the 'can_use' columns are dropped from the Service data model
|
||||||
|
@pytest.mark.parametrize("permission_to_add, can_send_letters, can_send_international_sms",
|
||||||
|
[(LETTER_TYPE, True, False),
|
||||||
|
(INTERNATIONAL_SMS_TYPE, False, True)])
|
||||||
|
def test_create_service_by_id_adding_service_permission_returns_service_with_permissions_set(
|
||||||
|
service_factory, permission_to_add, can_send_letters, can_send_international_sms):
|
||||||
|
service = service_factory.get('testing', email_from='testing')
|
||||||
|
|
||||||
|
dao_add_service_permission(service_id=service.id, permission=permission_to_add)
|
||||||
|
service.set_permissions()
|
||||||
|
|
||||||
|
service = dao_fetch_service_by_id(service.id)
|
||||||
|
assert len(service.permissions) == 3
|
||||||
|
assert all(p.permission in [SMS_TYPE, EMAIL_TYPE, permission_to_add] for p in service.permissions)
|
||||||
|
assert service.can_send_letters == can_send_letters
|
||||||
|
assert service.can_send_international_sms == can_send_international_sms
|
||||||
|
|
||||||
|
|
||||||
|
def test_remove_permission_from_service_by_id_returns_service_with_correct_permissions(service_factory):
|
||||||
|
service = service_factory.get('testing', email_from='testing')
|
||||||
|
dao_remove_service_permission(service_id=service.id, permission=SMS_TYPE)
|
||||||
|
|
||||||
|
service = dao_fetch_service_by_id(service.id)
|
||||||
|
assert len(service.permissions) == 1
|
||||||
|
assert service.permissions[0].permission == EMAIL_TYPE
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_service_by_id_adding_and_removing_letter_returns_service_without_letter(service_factory):
|
||||||
|
service = service_factory.get('testing', email_from='testing')
|
||||||
|
|
||||||
|
dao_add_service_permission(service_id=service.id, permission=LETTER_TYPE)
|
||||||
|
service.set_permissions()
|
||||||
|
|
||||||
|
service = dao_fetch_service_by_id(service.id)
|
||||||
|
assert len(service.permissions) == 3
|
||||||
|
assert all(p.permission in [SMS_TYPE, EMAIL_TYPE, LETTER_TYPE] for p in service.permissions)
|
||||||
|
assert service.can_send_letters
|
||||||
|
|
||||||
|
dao_remove_service_permission(service_id=service.id, permission=LETTER_TYPE)
|
||||||
|
service.set_permissions()
|
||||||
|
service = dao_fetch_service_by_id(service.id)
|
||||||
|
|
||||||
|
assert len(service.permissions) == 2
|
||||||
|
assert all(p.permission in [SMS_TYPE, EMAIL_TYPE] for p in service.permissions)
|
||||||
|
assert not service.can_send_letters
|
||||||
|
|
||||||
|
|
||||||
def test_create_service_creates_a_history_record_with_current_data(sample_user):
|
def test_create_service_creates_a_history_record_with_current_data(sample_user):
|
||||||
assert Service.query.count() == 0
|
assert Service.query.count() == 0
|
||||||
assert Service.get_history_model().query.count() == 0
|
assert Service.get_history_model().query.count() == 0
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ from app.dao.users_dao import save_model_user
|
|||||||
from app.dao.notifications_dao import dao_create_notification
|
from app.dao.notifications_dao import dao_create_notification
|
||||||
from app.dao.templates_dao import dao_create_template
|
from app.dao.templates_dao import dao_create_template
|
||||||
from app.dao.services_dao import dao_create_service
|
from app.dao.services_dao import dao_create_service
|
||||||
from app.dao.service_permissions_dao import dao_create_service_permission
|
from app.dao.service_permissions_dao import dao_add_service_permission
|
||||||
|
|
||||||
|
|
||||||
def create_user(mobile_number="+447700900986", email="notify@digital.cabinet-office.gov.uk", state='active'):
|
def create_user(mobile_number="+447700900986", email="notify@digital.cabinet-office.gov.uk", state='active'):
|
||||||
@@ -27,7 +27,8 @@ def create_user(mobile_number="+447700900986", email="notify@digital.cabinet-off
|
|||||||
return user
|
return user
|
||||||
|
|
||||||
|
|
||||||
def create_service(user=None, service_name="Sample service", service_id=None, restricted=False):
|
def create_service(
|
||||||
|
user=None, service_name="Sample service", service_id=None, service_permissions=[EMAIL_TYPE, SMS_TYPE]):
|
||||||
service = Service(
|
service = Service(
|
||||||
name=service_name,
|
name=service_name,
|
||||||
message_limit=1000,
|
message_limit=1000,
|
||||||
@@ -35,7 +36,7 @@ def create_service(user=None, service_name="Sample service", service_id=None, re
|
|||||||
email_from=service_name.lower().replace(' ', '.'),
|
email_from=service_name.lower().replace(' ', '.'),
|
||||||
created_by=user or create_user()
|
created_by=user or create_user()
|
||||||
)
|
)
|
||||||
dao_create_service(service, service.created_by, service_id)
|
dao_create_service(service, service.created_by, service_id, service_permissions=service_permissions)
|
||||||
return service
|
return service
|
||||||
|
|
||||||
|
|
||||||
@@ -147,7 +148,7 @@ def create_job(template,
|
|||||||
|
|
||||||
|
|
||||||
def create_service_permission(service_id, permission=EMAIL_TYPE):
|
def create_service_permission(service_id, permission=EMAIL_TYPE):
|
||||||
dao_create_service_permission(
|
dao_add_service_permission(
|
||||||
service_id if service_id else create_service().id, permission)
|
service_id if service_id else create_service().id, permission)
|
||||||
|
|
||||||
service_permissions = ServicePermission.query.all()
|
service_permissions = ServicePermission.query.all()
|
||||||
|
|||||||
Reference in New Issue
Block a user