mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-16 10:12:32 -05:00
Check service.permissions for the existence of schedule_notifications if the notications is being created with a scheduled_for param.
This commit is contained in:
@@ -146,8 +146,10 @@ class DVLAOrganisation(db.Model):
|
||||
|
||||
INTERNATIONAL_SMS_TYPE = 'international_sms'
|
||||
INBOUND_SMS_TYPE = 'inbound_sms'
|
||||
SCHEDULE_NOTIFICATIONS = 'schedule_notifications'
|
||||
|
||||
SERVICE_PERMISSION_TYPES = [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INBOUND_SMS_TYPE]
|
||||
SERVICE_PERMISSION_TYPES = [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INBOUND_SMS_TYPE,
|
||||
SCHEDULE_NOTIFICATIONS]
|
||||
|
||||
|
||||
class ServicePermissionTypes(db.Model):
|
||||
@@ -977,7 +979,7 @@ INVITED_USER_STATUS_TYPES = ['pending', 'accepted', 'cancelled']
|
||||
class ScheduledNotification(db.Model):
|
||||
__tablename__ = 'scheduled_notifications'
|
||||
|
||||
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4())
|
||||
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
||||
notification_id = db.Column(UUID(as_uuid=True), db.ForeignKey('notifications.id'), index=True, nullable=False)
|
||||
notification = db.relationship('Notification', uselist=False)
|
||||
scheduled_for = db.Column(db.DateTime, index=False, nullable=False)
|
||||
|
||||
@@ -6,7 +6,7 @@ from notifications_utils.recipients import (
|
||||
)
|
||||
|
||||
from app.dao import services_dao
|
||||
from app.models import KEY_TYPE_TEST, KEY_TYPE_TEAM, SMS_TYPE
|
||||
from app.models import KEY_TYPE_TEST, KEY_TYPE_TEAM, SMS_TYPE, SCHEDULE_NOTIFICATIONS
|
||||
from app.service.utils import service_allowed_to_send_to
|
||||
from app.v2.errors import TooManyRequestsError, BadRequestError, RateLimitError
|
||||
from app import redis_store
|
||||
@@ -92,6 +92,7 @@ def check_sms_content_char_count(content_count):
|
||||
raise BadRequestError(message=message)
|
||||
|
||||
|
||||
def service_can_schedule_notification(service):
|
||||
# TODO: implement once the service permission works.
|
||||
def service_can_schedule_notification(service, scheduled_for):
|
||||
if scheduled_for:
|
||||
if SCHEDULE_NOTIFICATIONS not in [p.permission for p in service.permissions]:
|
||||
raise BadRequestError(message="Your service must be invited to schedule notifications via the API.")
|
||||
|
||||
@@ -34,9 +34,8 @@ def post_notification(notification_type):
|
||||
form = validate(request.get_json(), post_sms_request)
|
||||
|
||||
scheduled_for = form.get("scheduled_for", None)
|
||||
if scheduled_for:
|
||||
if not service_can_schedule_notification(authenticated_service):
|
||||
return
|
||||
service_can_schedule_notification(authenticated_service, scheduled_for)
|
||||
|
||||
check_rate_limiting(authenticated_service, api_user)
|
||||
|
||||
form_send_to = form['phone_number'] if notification_type == SMS_TYPE else form['email_address']
|
||||
|
||||
@@ -3,13 +3,14 @@ import uuid
|
||||
import pytest
|
||||
from freezegun import freeze_time
|
||||
|
||||
from app.models import Notification, ScheduledNotification
|
||||
from app.models import Notification, ScheduledNotification, SCHEDULE_NOTIFICATIONS, EMAIL_TYPE, SMS_TYPE
|
||||
from flask import json, current_app
|
||||
|
||||
from app.models import Notification
|
||||
from app.v2.errors import RateLimitError
|
||||
from tests import create_authorization_header
|
||||
from tests.app.conftest import sample_template as create_sample_template, sample_service
|
||||
from tests.app.db import create_service, create_template
|
||||
|
||||
|
||||
@pytest.mark.parametrize("reference", [None, "reference_from_client"])
|
||||
@@ -350,26 +351,28 @@ def test_post_sms_should_persist_supplied_sms_number(client, sample_template_wit
|
||||
assert mocked.called
|
||||
|
||||
|
||||
@pytest.mark.skip("Once the service can be invited to schedule notifications we can add this test.")
|
||||
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
|
||||
[("sms", "phone_number", "07700 900 855"),
|
||||
("email", "email_address", "sample@email.com")])
|
||||
@freeze_time("2017-05-14 14:00:00")
|
||||
def test_post_notification_with_scheduled_for(client, sample_template, sample_email_template,
|
||||
def test_post_notification_with_scheduled_for(client, notify_db, notify_db_session,
|
||||
notification_type, key_send_to, send_to):
|
||||
service = create_service(service_name=str(uuid.uuid4()),
|
||||
service_permissions=[EMAIL_TYPE, SMS_TYPE, SCHEDULE_NOTIFICATIONS])
|
||||
template = create_template(service=service, template_type=notification_type)
|
||||
data = {
|
||||
key_send_to: send_to,
|
||||
'template_id': str(sample_email_template.id) if notification_type == 'email' else str(sample_template.id),
|
||||
'template_id': str(template.id) if notification_type == 'email' else str(template.id),
|
||||
'scheduled_for': '2017-05-14 14:15'
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
auth_header = create_authorization_header(service_id=service.id)
|
||||
|
||||
response = client.post('/v2/notifications/{}'.format(notification_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
assert response.status_code == 201
|
||||
resp_json = json.loads(response.get_data(as_text=True))
|
||||
scheduled_notification = ScheduledNotification.query.all()
|
||||
scheduled_notification = ScheduledNotification.query.filter_by(notification_id=resp_json["id"]).all()
|
||||
assert len(scheduled_notification) == 1
|
||||
assert resp_json["id"] == str(scheduled_notification[0].notification_id)
|
||||
assert resp_json["scheduled_for"] == '2017-05-14 14:15'
|
||||
@@ -379,8 +382,8 @@ def test_post_notification_with_scheduled_for(client, sample_template, sample_em
|
||||
[("sms", "phone_number", "07700 900 855"),
|
||||
("email", "email_address", "sample@email.com")])
|
||||
@freeze_time("2017-05-14 14:00:00")
|
||||
def test_post_notification_with_scheduled_for_raises_bad_request(client, sample_template, sample_email_template,
|
||||
notification_type, key_send_to, send_to):
|
||||
def test_post_notification_raises_bad_request_if_service_not_invited_to_schedule(
|
||||
client, sample_template, sample_email_template, notification_type, key_send_to, send_to):
|
||||
data = {
|
||||
key_send_to: send_to,
|
||||
'template_id': str(sample_email_template.id) if notification_type == 'email' else str(sample_template.id),
|
||||
|
||||
Reference in New Issue
Block a user