Command to run a query and upload a file to S3.

Need to create a bucket to save notification_history back ups to.

app/broadcast_message/rest.py

@@ -2,7 +2,6 @@ from datetime import datetime
 
 import iso8601
 from flask import Blueprint, jsonify, request, current_app
-
 from app.config import QueueNames
 from app.dao.templates_dao import dao_get_template_by_id_and_service_id
 from app.dao.users_dao import get_user_by_id
@@ -13,7 +12,7 @@ from app.dao.broadcast_message_dao import (
     dao_update_broadcast_message,
 )
 from app.dao.services_dao import dao_fetch_service_by_id
-from app.errors import register_errors
+from app.errors import register_errors, InvalidRequest
 from app.models import BroadcastMessage, BroadcastStatusType
 from app.celery.broadcast_message_tasks import send_broadcast_message
 from app.broadcast_message.broadcast_message_schema import (
@@ -37,6 +36,40 @@ def _parse_nullable_datetime(dt):
     return dt
 
 
+def _update_broadcast_message(broadcast_message, new_status, updating_user):
+    if updating_user not in broadcast_message.service.users:
+        raise InvalidRequest(
+            f'User {updating_user.id} cannot approve broadcast_message {broadcast_message.id} from other service',
+            status_code=400
+        )
+
+    if new_status not in BroadcastStatusType.ALLOWED_STATUS_TRANSITIONS[broadcast_message.status]:
+        raise InvalidRequest(
+            f'Cannot move broadcast_message {broadcast_message.id} from {broadcast_message.status} to {new_status}',
+            status_code=400
+        )
+
+    if new_status == BroadcastStatusType.BROADCASTING:
+        # TODO: Remove this platform admin shortcut when the feature goes live
+        if updating_user == broadcast_message.created_by and not updating_user.platform_admin:
+            raise InvalidRequest(
+                f'User {updating_user.id} cannot approve their own broadcast_message {broadcast_message.id}',
+                status_code=400
+            )
+        else:
+            broadcast_message.approved_at = datetime.utcnow()
+            broadcast_message.approved_by = updating_user
+
+    if new_status == BroadcastStatusType.CANCELLED:
+        broadcast_message.cancelled_at = datetime.utcnow()
+        broadcast_message.cancelled_by = updating_user
+
+    current_app.logger.info(
+        f'broadcast_message {broadcast_message.id} moving from {broadcast_message.status} to {new_status}'
+    )
+    broadcast_message.status = new_status
+
+
 @broadcast_message_blueprint.route('', methods=['GET'])
 def get_broadcast_messages_for_service(service_id):
     # TODO: should this return template content/data in some way? or can we rely on them being cached admin side.
@@ -85,6 +118,12 @@ def update_broadcast_message(service_id, broadcast_message_id):
 
     broadcast_message = dao_get_broadcast_message_by_id_and_service_id(broadcast_message_id, service_id)
 
+    if broadcast_message.status not in BroadcastStatusType.PRE_BROADCAST_STATUSES:
+        raise InvalidRequest(
+            f'Cannot update broadcast_message {broadcast_message.id} while it has status {broadcast_message.status}',
+            status_code=400
+        )
+
     if 'personalisation' in data:
         broadcast_message.personalisation = data['personalisation']
     if 'starts_at' in data:
@@ -107,21 +146,9 @@ def update_broadcast_message_status(service_id, broadcast_message_id):
     broadcast_message = dao_get_broadcast_message_by_id_and_service_id(broadcast_message_id, service_id)
 
     new_status = data['status']
+    updating_user = get_user_by_id(data['created_by'])
 
-    # TODO: Restrict status transitions
-    # TODO: validate that the user belongs to the same service, isn't the creator, has permissions, etc
-    if new_status == BroadcastStatusType.BROADCASTING:
-        broadcast_message.approved_at = datetime.utcnow()
-        broadcast_message.approved_by = get_user_by_id(data['created_by'])
-    if new_status == BroadcastStatusType.CANCELLED:
-        broadcast_message.cancelled_at = datetime.utcnow()
-        broadcast_message.cancelled_by = get_user_by_id(data['created_by'])
-
-    broadcast_message.status = new_status
-
-    current_app.logger.info(
-        f'broadcast_message {broadcast_message_id} moving from {broadcast_message.status} to {new_status}'
-    )
+    _update_broadcast_message(broadcast_message, new_status, updating_user)
     dao_update_broadcast_message(broadcast_message)
 
     if new_status == BroadcastStatusType.BROADCASTING:

app/celery/scheduled_tasks.py

@@ -12,10 +12,10 @@ from app import notify_celery, zendesk_client
 from app.celery.tasks import (
     process_job,
     get_recipient_csv_and_template_and_sender_id,
-    process_row
-)
+    process_row,
+    process_incomplete_jobs)
 from app.celery.letters_pdf_tasks import get_pdf_for_templated_letter
-from app.config import QueueNames, TaskNames
+from app.config import QueueNames
 from app.dao.invited_org_user_dao import delete_org_invitations_created_more_than_two_days_ago
 from app.dao.invited_user_dao import delete_invitations_created_more_than_two_days_ago
 from app.dao.jobs_dao import (
@@ -45,7 +45,6 @@ from app.models import (
     EMAIL_TYPE,
 )
 from app.notifications.process_notifications import send_notification_to_queue
-from app.v2.errors import JobIncompleteError
 
 
 @notify_celery.task(name="run-scheduled-jobs")
@@ -149,12 +148,11 @@ def check_job_status():
         job_ids.append(str(job.id))
 
     if job_ids:
-        notify_celery.send_task(
-            name=TaskNames.PROCESS_INCOMPLETE_JOBS,
-            args=(job_ids,),
+        current_app.logger.info("Job(s) {} have not completed.".format(job_ids))
+        process_incomplete_jobs.apply_async(
+            [job_ids],
             queue=QueueNames.JOBS
         )
-        raise JobIncompleteError("Job(s) {} have not completed.".format(job_ids))
 
 
 @notify_celery.task(name='replay-created-notifications')

app/commands.py

@@ -1,5 +1,6 @@
 import csv
 import functools
+import gzip
 import uuid
 from datetime import datetime, timedelta
 from decimal import Decimal
@@ -7,6 +8,8 @@ from decimal import Decimal
 import click
 import flask
 import itertools
+
+from boto3 import client as boto_client, client
 from click_datetime import Datetime as click_dt
 from flask import current_app, json
 from notifications_utils.recipients import RecipientCSV
@@ -927,3 +930,47 @@ def process_row_from_job(job_id, job_row_number):
             notification_id = process_row(row, template, job, job.service)
             current_app.logger.info("Process row {} for job {} created notification_id: {}".format(
                 job_row_number, job_id, notification_id))
+
+
+@notify_command(name='backup-postgres-table')
+@click.option('-q', '--query', required=True, help='Query for backup data')
+@click.option('-f', '--output_file_name', required=True, help='Output file name')
+def backup_postgres_table(query, output_file_name):
+    """
+    Copy the results of the SQL query passed in (query) to a file (dest_file).
+    """
+    try:
+        # Create temporary file to contain database data.
+        dest_filehandle = open(output_file_name, 'w+')
+        current_app.logger.info("Opened temporary file {} for storing data from database".format(output_file_name))
+    except Exception as e:
+        current_app.logger.error("Unable to create temporary file {}: {}".format(output_file_name, e))
+        return None
+
+    current_app.logger.info("Writing data from '{}' to {}".format(query, output_file_name))
+
+    # Note that need to create dest_file as a writeable file before calling the following method:
+    copy_out = "COPY ({}) TO STDOUT WITH CSV DELIMITER '|' HEADER".format(query)
+    # copy_out="COPY testtable TO STDOUT WITH CSV HEADER"
+    curs = db.session.connection().connection.cursor()
+    curs.copy_expert(sql=copy_out, file=dest_filehandle)
+
+    dest_filehandle.close()
+    comp_file = compress_file(output_file_name)
+
+    s3_client = client('s3', current_app.config['AWS_REGION'])
+    s3_client.upload_file(
+        comp_file,
+        'development-letters-pdf',
+        output_file_name,
+        ExtraArgs={'ServerSideEncryption': 'AES256'}
+    )
+
+
+def compress_file(src_file):
+    compressed_file = "{}.gz".format(str(src_file))
+    with open(src_file, 'rb') as f_in:
+        with gzip.open(compressed_file, 'wb') as f_out:
+            for line in f_in:
+                f_out.write(line)
+    return compressed_file

app/dao/broadcast_message_dao.py

@@ -1,5 +1,5 @@
 from app import db
-from app.models import BroadcastMessage
+from app.models import BroadcastMessage, BroadcastEvent
 from app.dao.dao_utils import transactional
 
 
@@ -28,3 +28,17 @@ def dao_get_broadcast_messages_for_service(service_id):
     return BroadcastMessage.query.filter(
         BroadcastMessage.service_id == service_id
     ).order_by(BroadcastMessage.created_at)
+
+
+def get_earlier_events_for_broadcast_event(broadcast_event_id):
+    """
+    This is used to build up the references list.
+    """
+    this_event = BroadcastEvent.query.get(broadcast_event_id)
+
+    return BroadcastEvent.query.filter(
+        BroadcastEvent.broadcast_message_id == this_event.broadcast_message_id,
+        BroadcastEvent.sent_at < this_event.sent_at
+    ).order_by(
+        BroadcastEvent.sent_at.asc()
+    ).all()

app/models.py

@@ -39,6 +39,7 @@ from app import (
     encryption,
     DATETIME_FORMAT,
     DATETIME_FORMAT_NO_TIMEZONE)
+from app.utils import get_dt_string_or_none
 
 from app.history_meta import Versioned
 
@@ -169,7 +170,7 @@ class User(db.Model):
             'current_session_id': self.current_session_id,
             'failed_login_count': self.failed_login_count,
             'email_access_validated_at': self.email_access_validated_at.strftime(DATETIME_FORMAT),
-            'logged_in_at': self.logged_in_at.strftime(DATETIME_FORMAT) if self.logged_in_at else None,
+            'logged_in_at': get_dt_string_or_none(self.logged_in_at),
             'mobile_number': self.mobile_number,
             'organisations': [x.id for x in self.organisations if x.active],
             'password_changed_at': self.password_changed_at.strftime(DATETIME_FORMAT_NO_TIMEZONE),
@@ -569,7 +570,7 @@ class AnnualBilling(db.Model):
             'service_id': self.service_id,
             'financial_year_start': self.financial_year_start,
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
-            "updated_at": self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None,
+            "updated_at": get_dt_string_or_none(self.updated_at),
             "service": serialize_service() if self.service else None,
         }
 
@@ -600,7 +601,7 @@ class InboundNumber(db.Model):
             "service": serialize_service() if self.service else None,
             "active": self.active,
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
-            "updated_at": self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None,
+            "updated_at": get_dt_string_or_none(self.updated_at),
         }
 
 
@@ -631,7 +632,7 @@ class ServiceSmsSender(db.Model):
             "archived": self.archived,
             "inbound_number_id": str(self.inbound_number_id) if self.inbound_number_id else None,
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
-            "updated_at": self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None,
+            "updated_at": get_dt_string_or_none(self.updated_at),
         }
 
 
@@ -723,7 +724,7 @@ class ServiceInboundApi(db.Model, Versioned):
             "url": self.url,
             "updated_by_id": str(self.updated_by_id),
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
-            "updated_at": self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None
+            "updated_at": get_dt_string_or_none(self.updated_at),
         }
 
 
@@ -762,7 +763,7 @@ class ServiceCallbackApi(db.Model, Versioned):
             "url": self.url,
             "updated_by_id": str(self.updated_by_id),
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
-            "updated_at": self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None
+            "updated_at": get_dt_string_or_none(self.updated_at),
         }
 
 
@@ -996,7 +997,7 @@ class TemplateBase(db.Model):
             "id": str(self.id),
             "type": self.template_type,
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
-            "updated_at": self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None,
+            "updated_at": get_dt_string_or_none(self.updated_at),
             "created_by": self.created_by.email_address,
             "version": self.version,
             "body": self.content,
@@ -1009,6 +1010,7 @@ class TemplateBase(db.Model):
                 for key in self._as_utils_template().placeholders
             },
             "postage": self.postage,
+            "letter_contact_block": self.service_letter_contact.contact_block if self.service_letter_contact else None,
         }
 
         return serialized
@@ -1627,7 +1629,7 @@ class Notification(db.Model):
             "subject": self.subject,
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
             "created_by_name": self.get_created_by_name(),
-            "sent_at": self.sent_at.strftime(DATETIME_FORMAT) if self.sent_at else None,
+            "sent_at": get_dt_string_or_none(self.sent_at),
             "completed_at": self.completed_at(),
             "scheduled_for": None,
             "postage": self.postage
@@ -1954,7 +1956,7 @@ class ServiceEmailReplyTo(db.Model):
             'is_default': self.is_default,
             'archived': self.archived,
             'created_at': self.created_at.strftime(DATETIME_FORMAT),
-            'updated_at': self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None
+            'updated_at': get_dt_string_or_none(self.updated_at),
         }
 
 
@@ -1980,7 +1982,7 @@ class ServiceLetterContact(db.Model):
             'is_default': self.is_default,
             'archived': self.archived,
             'created_at': self.created_at.strftime(DATETIME_FORMAT),
-            'updated_at': self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None
+            'updated_at': get_dt_string_or_none(self.updated_at),
         }
 
 
@@ -2057,7 +2059,7 @@ class Complaint(db.Model):
             'service_name': self.service.name,
             'ses_feedback_id': str(self.ses_feedback_id),
             'complaint_type': self.complaint_type,
-            'complaint_date': self.complaint_date.strftime(DATETIME_FORMAT) if self.complaint_date else None,
+            'complaint_date': get_dt_string_or_none(self.complaint_date),
             'created_at': self.created_at.strftime(DATETIME_FORMAT),
         }
 
@@ -2091,7 +2093,7 @@ class ServiceDataRetention(db.Model):
             "notification_type": self.notification_type,
             "days_of_retention": self.days_of_retention,
             "created_at": self.created_at.strftime(DATETIME_FORMAT),
-            "updated_at": self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None,
+            "updated_at": get_dt_string_or_none(self.updated_at),
         }
 
 
@@ -2122,7 +2124,8 @@ class ServiceContactList(db.Model):
     updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.datetime.utcnow)
     archived = db.Column(db.Boolean, nullable=False, default=False)
 
-    def get_job_count(self):
+    @property
+    def job_count(self):
         today = datetime.datetime.utcnow().date()
         return Job.query.filter(
             Job.contact_list_id == self.id,
@@ -2136,13 +2139,20 @@ class ServiceContactList(db.Model):
             )
         ).count()
 
+    @property
+    def has_jobs(self):
+        return bool(Job.query.filter(
+            Job.contact_list_id == self.id,
+        ).first())
+
     def serialize(self):
         created_at_in_bst = convert_utc_to_bst(self.created_at)
         contact_list = {
             "id": str(self.id),
             "original_file_name": self.original_file_name,
             "row_count": self.row_count,
-            "job_count": self.get_job_count(),
+            "recent_job_count": self.job_count,
+            "has_jobs": self.has_jobs,
             "template_type": self.template_type,
             "service_id": str(self.service_id),
             "created_by": self.created_by.name,
@@ -2163,6 +2173,24 @@ class BroadcastStatusType(db.Model):
 
     STATUSES = [DRAFT, PENDING_APPROVAL, REJECTED, BROADCASTING, COMPLETED, CANCELLED, TECHNICAL_FAILURE]
 
+    # a broadcast message can be edited while in one of these states
+    PRE_BROADCAST_STATUSES = [DRAFT, PENDING_APPROVAL, REJECTED]
+    LIVE_STATUSES = [BROADCASTING, COMPLETED, CANCELLED]
+
+    # these are only the transitions we expect to administer via the API code.
+    ALLOWED_STATUS_TRANSITIONS = {
+        DRAFT: {
+            PENDING_APPROVAL,
+            BROADCASTING,  # TODO: Remove me once we have pending approval flow put in properly
+        },
+        PENDING_APPROVAL: {REJECTED, DRAFT, BROADCASTING},
+        REJECTED: {DRAFT, PENDING_APPROVAL},
+        BROADCASTING: {COMPLETED, CANCELLED},
+        COMPLETED: {},
+        CANCELLED: {},
+        TECHNICAL_FAILURE: {},
+    }
+
     name = db.Column(db.String, primary_key=True)
 
 
@@ -2239,15 +2267,92 @@ class BroadcastMessage(db.Model):
 
             'status': self.status,
 
-            'starts_at': self.starts_at.strftime(DATETIME_FORMAT) if self.starts_at else None,
-            'finishes_at': self.finishes_at.strftime(DATETIME_FORMAT) if self.finishes_at else None,
+            'starts_at': get_dt_string_or_none(self.starts_at),
+            'finishes_at': get_dt_string_or_none(self.finishes_at),
 
-            'created_at': self.created_at.strftime(DATETIME_FORMAT) if self.created_at else None,
-            'approved_at': self.approved_at.strftime(DATETIME_FORMAT) if self.approved_at else None,
-            'cancelled_at': self.cancelled_at.strftime(DATETIME_FORMAT) if self.cancelled_at else None,
-            'updated_at': self.updated_at.strftime(DATETIME_FORMAT) if self.updated_at else None,
+            'created_at': get_dt_string_or_none(self.created_at),
+            'approved_at': get_dt_string_or_none(self.approved_at),
+            'cancelled_at': get_dt_string_or_none(self.cancelled_at),
+            'updated_at': get_dt_string_or_none(self.updated_at),
 
             'created_by_id': str(self.created_by_id),
             'approved_by_id': str(self.approved_by_id),
             'cancelled_by_id': str(self.cancelled_by_id),
         }
+
+
+class BroadcastEventMessageType:
+    ALERT = 'alert'
+    UPDATE = 'update'
+    CANCEL = 'cancel'
+
+    MESSAGE_TYPES = [ALERT, UPDATE, CANCEL]
+
+
+class BroadcastEvent(db.Model):
+    """
+    This table represents a single CAP XML blob that we sent to the mobile network providers.
+
+    We should be able to create the complete CAP message without joining from this to any other tables, eg
+    template, service, or broadcast_message.
+
+    The only exception to this is that we will have to join to itself to find other broadcast_events with the
+    same broadcast_message_id when building up the `<references>` xml field for updating/cancelling an existing message.
+
+    As such, this shouldn't have foreign keys to things that can change or be deleted.
+    """
+    __tablename__ = 'broadcast_event'
+
+    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+
+    service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'))
+    service = db.relationship('Service')
+
+    broadcast_message_id = db.Column(UUID(as_uuid=True), db.ForeignKey('broadcast_message.id'), nullable=False)
+    broadcast_message = db.relationship('BroadcastMessage', backref='events')
+
+    # this is used for <sent> in the cap xml
+    sent_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow)
+
+    # msgType. alert, cancel, or update. (other options in the spec are "ack" and "error")
+    message_type = db.Column(db.String, nullable=False)
+
+    # this will be json containing anything that isnt hardcoded in utils/cbc proxy. for now just body but may grow to
+    # include, eg, title, headline, instructions.
+    transmitted_content = db.Column(
+        JSONB(none_as_null=True),
+        nullable=True
+    )
+    # unsubstantiated reckon: even if we're sending a cancel, we'll still need to provide areas
+    transmitted_areas = db.Column(JSONB(none_as_null=True), nullable=False, default=list)
+    transmitted_sender = db.Column(db.String(), nullable=False)
+
+    # we may only need this starts_at if this is scheduled for the future. Interested to see how this affects
+    # updates/cancels (ie: can you schedule an update for the future?)
+    transmitted_starts_at = db.Column(db.DateTime, nullable=True)
+    transmitted_finishes_at = db.Column(db.DateTime, nullable=True)
+
+    # @property
+    # def reference(self):
+    #     # TODO: write this `from_event` function
+    #     return BroadcastMessageTemplate.from_event(self.serialize()).reference
+
+    def serialize(self):
+        return {
+            'id': self.id,
+
+            'service_id': self.service_id,
+
+            # 'reference': self.reference,
+
+            'broadcast_message_id': self.broadcast_message_id,
+            'sent_at': self.sent_at,
+            'message_type': self.message_type,
+
+            'transmitted_content': self.transmitted_content,
+            'transmitted_areas': self.transmitted_areas,
+            'transmitted_sender': self.transmitted_sender,
+
+            'transmitted_starts_at': get_dt_string_or_none(self.transmitted_starts_at),
+            'transmitted_finishes_at': get_dt_string_or_none(self.transmitted_finishes_at),
+        }

app/utils.py

@@ -11,6 +11,7 @@ from notifications_utils.template import (
     BroadcastMessageTemplate,
 )
 
+from app import DATETIME_FORMAT
 
 local_timezone = pytz.timezone("Europe/London")
 
@@ -141,3 +142,7 @@ def get_notification_table_to_use(service, notification_type, process_day, has_d
 def get_archived_db_column_value(column):
     date = datetime.utcnow().strftime("%Y-%m-%d")
     return f'_archived_{date}_{column}'
+
+
+def get_dt_string_or_none(val):
+    return val.strftime(DATETIME_FORMAT) if val else None

app/v2/errors.py

@@ -10,23 +10,6 @@ from app.authentication.auth import AuthError
 from app.errors import InvalidRequest
 
 
-class JobIncompleteError(Exception):
-    def __init__(self, message):
-        self.message = message
-        self.status_code = 500
-
-    def to_dict_v2(self):
-        return {
-            'status_code': self.status_code,
-            "errors": [
-                {
-                    "error": 'JobIncompleteError',
-                    "message": self.message
-                }
-            ]
-        }
-
-
 class TooManyRequestsError(InvalidRequest):
     status_code = 429
     message_template = 'Exceeded send limits ({}) for today'
@@ -91,10 +74,6 @@ def register_errors(blueprint):
         current_app.logger.info(error)
         return jsonify(json.loads(error.message)), 400
 
-    @blueprint.errorhandler(JobIncompleteError)
-    def job_incomplete_error(error):
-        return jsonify(error.to_dict_v2()), 500
-
     @blueprint.errorhandler(NoResultFound)
     @blueprint.errorhandler(DataError)
     def no_result_found(e):

migrations/versions/0326_broadcast_event.py

@@ -0,0 +1,43 @@
+"""
+
+Revision ID: 0326_broadcast_event
+Revises: 0325_int_letter_rates_fix
+Create Date: 2020-07-24 12:40:35.809523
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision = '0326_broadcast_event'
+down_revision = '0325_int_letter_rates_fix'
+
+
+def upgrade():
+    op.create_table('broadcast_event',
+        sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column('service_id', postgresql.UUID(as_uuid=True), nullable=True),
+        sa.Column('broadcast_message_id', postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column('sent_at', sa.DateTime(), nullable=False),
+        sa.Column('message_type', sa.String(), nullable=False),
+        sa.Column('transmitted_content', postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=True),
+        sa.Column('transmitted_areas', postgresql.JSONB(none_as_null=True, astext_type=sa.Text()), nullable=False),
+        sa.Column('transmitted_sender', sa.String(), nullable=False),
+        sa.Column('transmitted_starts_at', sa.DateTime(), nullable=True),
+        sa.Column('transmitted_finishes_at', sa.DateTime(), nullable=True),
+        sa.ForeignKeyConstraint(['broadcast_message_id'], ['broadcast_message.id'], ),
+        sa.ForeignKeyConstraint(['service_id'], ['services.id'], ),
+        sa.PrimaryKeyConstraint('id')
+    )
+    # this shouldn't be nullable. it defaults to `[]` in python.
+    op.alter_column('broadcast_message', 'areas', existing_type=postgresql.JSONB(astext_type=sa.Text()), nullable=False)
+    # this can't be nullable. it defaults to 'draft' in python.
+    op.alter_column('broadcast_message', 'status', existing_type=sa.VARCHAR(), nullable=False)
+    op.create_foreign_key(None, 'broadcast_message', 'broadcast_status_type', ['status'], ['name'])
+
+
+def downgrade():
+    op.drop_constraint('broadcast_message_status_fkey', 'broadcast_message', type_='foreignkey')
+    op.alter_column('broadcast_message', 'status', existing_type=sa.VARCHAR(), nullable=True)
+    op.alter_column('broadcast_message', 'areas', existing_type=postgresql.JSONB(astext_type=sa.Text()), nullable=True)
+    op.drop_table('broadcast_event')

tests/app/broadcast_message/test_rest.py

@@ -132,9 +132,14 @@ def test_create_broadcast_message_400s_if_json_schema_fails_validation(
     assert response['errors'] == expected_errors
 
 
-def test_update_broadcast_message(admin_request, sample_service):
+@pytest.mark.parametrize('status', [
+    BroadcastStatusType.DRAFT,
+    BroadcastStatusType.PENDING_APPROVAL,
+    BroadcastStatusType.REJECTED,
+])
+def test_update_broadcast_message_allows_edit_while_not_yet_live(admin_request, sample_service, status):
     t = create_template(sample_service, BROADCAST_TYPE)
-    bm = create_broadcast_message(t, areas=['manchester'])
+    bm = create_broadcast_message(t, areas=['manchester'], status=status)
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message',
@@ -149,6 +154,26 @@ def test_update_broadcast_message(admin_request, sample_service):
     assert response['updated_at'] is not None
 
 
+@pytest.mark.parametrize('status', [
+    BroadcastStatusType.BROADCASTING,
+    BroadcastStatusType.CANCELLED,
+    BroadcastStatusType.COMPLETED,
+    BroadcastStatusType.TECHNICAL_FAILURE,
+])
+def test_update_broadcast_message_doesnt_allow_edits_after_broadcast_goes_live(admin_request, sample_service, status):
+    t = create_template(sample_service, BROADCAST_TYPE)
+    bm = create_broadcast_message(t, areas=['manchester'], status=status)
+
+    response = admin_request.post(
+        'broadcast_message.update_broadcast_message',
+        _data={'areas': ['london', 'glasgow']},
+        service_id=t.service_id,
+        broadcast_message_id=bm.id,
+        _expected_status=400
+    )
+    assert f'status {status}' in response['message']
+
+
 def test_update_broadcast_message_sets_finishes_at_separately(admin_request, sample_service):
     t = create_template(sample_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, areas=['manchester'])
@@ -243,7 +268,8 @@ def test_update_broadcast_message_status_doesnt_let_you_update_other_things(admi
 def test_update_broadcast_message_status_stores_cancelled_by_and_cancelled_at(admin_request, sample_service):
     t = create_template(sample_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, status=BroadcastStatusType.BROADCASTING)
-    canceller = create_user('canceller@gov.uk')
+    canceller = create_user(email='canceller@gov.uk')
+    sample_service.users.append(canceller)
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',
@@ -265,7 +291,8 @@ def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_
 ):
     t = create_template(sample_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
-    approver = create_user('approver@gov.uk')
+    approver = create_user(email='approver@gov.uk')
+    sample_service.users.append(approver)
     mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
 
     response = admin_request.post(
@@ -280,3 +307,105 @@ def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_
     assert response['approved_at'] is not None
     assert response['approved_by_id'] == str(approver.id)
     mock_task.assert_called_once_with(kwargs={'broadcast_message_id': str(bm.id)}, queue='notify-internal-tasks')
+
+
+def test_update_broadcast_message_status_rejects_approval_from_creator(
+    admin_request,
+    sample_service,
+    mocker
+):
+    t = create_template(sample_service, BROADCAST_TYPE)
+    bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+
+    response = admin_request.post(
+        'broadcast_message.update_broadcast_message_status',
+        _data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(t.created_by_id)},
+        service_id=t.service_id,
+        broadcast_message_id=bm.id,
+        _expected_status=400
+    )
+
+    assert mock_task.called is False
+    assert f'cannot approve their own broadcast' in response['message']
+
+
+def test_update_broadcast_message_status_allows_platform_admin_to_approve_own_message(
+    notify_db,
+    admin_request,
+    sample_service,
+    mocker
+):
+    user = sample_service.created_by
+    user.platform_admin = True
+    t = create_template(sample_service, BROADCAST_TYPE)
+    bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+
+    response = admin_request.post(
+        'broadcast_message.update_broadcast_message_status',
+        _data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(user.id)},
+        service_id=t.service_id,
+        broadcast_message_id=bm.id,
+        _expected_status=200
+    )
+
+    assert response['status'] == BroadcastStatusType.BROADCASTING
+    assert response['approved_at'] is not None
+    assert response['created_by_id'] == str(user.id)
+    assert response['approved_by_id'] == str(user.id)
+    mock_task.assert_called_once_with(kwargs={'broadcast_message_id': str(bm.id)}, queue='notify-internal-tasks')
+
+
+def test_update_broadcast_message_status_rejects_approval_from_user_not_on_that_service(
+    admin_request,
+    sample_service,
+    mocker
+):
+    t = create_template(sample_service, BROADCAST_TYPE)
+    bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
+    approver = create_user(email='approver@gov.uk')
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+
+    response = admin_request.post(
+        'broadcast_message.update_broadcast_message_status',
+        _data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(approver.id)},
+        service_id=t.service_id,
+        broadcast_message_id=bm.id,
+        _expected_status=400
+    )
+
+    assert mock_task.called is False
+    assert f'cannot approve broadcast' in response['message']
+
+
+@pytest.mark.parametrize('current_status, new_status', [
+    (BroadcastStatusType.DRAFT, BroadcastStatusType.DRAFT),
+    (BroadcastStatusType.BROADCASTING, BroadcastStatusType.PENDING_APPROVAL),
+    (BroadcastStatusType.COMPLETED, BroadcastStatusType.BROADCASTING),
+    (BroadcastStatusType.CANCELLED, BroadcastStatusType.DRAFT),
+    pytest.param(BroadcastStatusType.DRAFT, BroadcastStatusType.BROADCASTING, marks=pytest.mark.xfail()),
+])
+def test_update_broadcast_message_status_restricts_status_transitions_to_explicit_list(
+    admin_request,
+    sample_service,
+    mocker,
+    current_status,
+    new_status
+):
+    t = create_template(sample_service, BROADCAST_TYPE)
+    bm = create_broadcast_message(t, status=current_status)
+    approver = create_user(email='approver@gov.uk')
+    sample_service.users.append(approver)
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+
+    response = admin_request.post(
+        'broadcast_message.update_broadcast_message_status',
+        _data={'status': new_status, 'created_by': str(approver.id)},
+        service_id=t.service_id,
+        broadcast_message_id=bm.id,
+        _expected_status=400
+    )
+
+    assert mock_task.called is False
+    assert f'from {current_status} to {new_status}' in response['message']

tests/app/celery/test_scheduled_tasks.py

@@ -19,7 +19,7 @@ from app.celery.scheduled_tasks import (
     check_for_services_with_high_failure_rates_or_sending_to_tv_numbers,
     switch_current_sms_provider_on_slow_delivery,
 )
-from app.config import QueueNames, TaskNames, Config
+from app.config import QueueNames, Config
 from app.dao.jobs_dao import dao_get_job_by_id
 from app.dao.provider_details_dao import get_provider_details_by_identifier
 from app.models import (
@@ -29,7 +29,6 @@ from app.models import (
     NOTIFICATION_DELIVERED,
     NOTIFICATION_PENDING_VIRUS_CHECK,
 )
-from app.v2.errors import JobIncompleteError
 from tests.app import load_example_csv
 
 from tests.app.db import (
@@ -141,44 +140,40 @@ def test_switch_current_sms_provider_on_slow_delivery_does_nothing_if_no_need(
     assert mock_reduce.called is False
 
 
-def test_check_job_status_task_raises_job_incomplete_error(mocker, sample_template):
-    mock_celery = mocker.patch('app.celery.tasks.notify_celery.send_task')
+def test_check_job_status_task_calls_process_incomplete_jobs(mocker, sample_template):
+    mock_celery = mocker.patch('app.celery.tasks.process_incomplete_jobs.apply_async')
     job = create_job(template=sample_template, notification_count=3,
                      created_at=datetime.utcnow() - timedelta(minutes=31),
                      processing_started=datetime.utcnow() - timedelta(minutes=31),
                      job_status=JOB_STATUS_IN_PROGRESS)
     create_notification(template=sample_template, job=job)
-    with pytest.raises(expected_exception=JobIncompleteError) as e:
-        check_job_status()
-    assert e.value.message == "Job(s) ['{}'] have not completed.".format(str(job.id))
+    check_job_status()
 
     mock_celery.assert_called_once_with(
-        name=TaskNames.PROCESS_INCOMPLETE_JOBS,
-        args=([str(job.id)],),
+        [[str(job.id)]],
         queue=QueueNames.JOBS
     )
 
 
-def test_check_job_status_task_raises_job_incomplete_error_when_scheduled_job_is_not_complete(mocker, sample_template):
-    mock_celery = mocker.patch('app.celery.tasks.notify_celery.send_task')
+def test_check_job_status_task_calls_process_incomplete_jobs_when_scheduled_job_is_not_complete(
+    mocker, sample_template
+):
+    mock_celery = mocker.patch('app.celery.tasks.process_incomplete_jobs.apply_async')
     job = create_job(template=sample_template, notification_count=3,
                      created_at=datetime.utcnow() - timedelta(hours=2),
                      scheduled_for=datetime.utcnow() - timedelta(minutes=31),
                      processing_started=datetime.utcnow() - timedelta(minutes=31),
                      job_status=JOB_STATUS_IN_PROGRESS)
-    with pytest.raises(expected_exception=JobIncompleteError) as e:
-        check_job_status()
-    assert e.value.message == "Job(s) ['{}'] have not completed.".format(str(job.id))
+    check_job_status()
 
     mock_celery.assert_called_once_with(
-        name=TaskNames.PROCESS_INCOMPLETE_JOBS,
-        args=([str(job.id)],),
+        [[str(job.id)]],
         queue=QueueNames.JOBS
     )
 
 
-def test_check_job_status_task_raises_job_incomplete_error_for_multiple_jobs(mocker, sample_template):
-    mock_celery = mocker.patch('app.celery.tasks.notify_celery.send_task')
+def test_check_job_status_task_calls_process_incomplete_jobs_for_multiple_jobs(mocker, sample_template):
+    mock_celery = mocker.patch('app.celery.tasks.process_incomplete_jobs.apply_async')
     job = create_job(template=sample_template, notification_count=3,
                      created_at=datetime.utcnow() - timedelta(hours=2),
                      scheduled_for=datetime.utcnow() - timedelta(minutes=31),
@@ -189,20 +184,16 @@ def test_check_job_status_task_raises_job_incomplete_error_for_multiple_jobs(moc
                        scheduled_for=datetime.utcnow() - timedelta(minutes=31),
                        processing_started=datetime.utcnow() - timedelta(minutes=31),
                        job_status=JOB_STATUS_IN_PROGRESS)
-    with pytest.raises(expected_exception=JobIncompleteError) as e:
-        check_job_status()
-    assert str(job.id) in e.value.message
-    assert str(job_2.id) in e.value.message
+    check_job_status()
 
     mock_celery.assert_called_once_with(
-        name=TaskNames.PROCESS_INCOMPLETE_JOBS,
-        args=([str(job.id), str(job_2.id)],),
+        [[str(job.id), str(job_2.id)]],
         queue=QueueNames.JOBS
     )
 
 
 def test_check_job_status_task_only_sends_old_tasks(mocker, sample_template):
-    mock_celery = mocker.patch('app.celery.tasks.notify_celery.send_task')
+    mock_celery = mocker.patch('app.celery.tasks.process_incomplete_jobs.apply_async')
     job = create_job(
         template=sample_template,
         notification_count=3,
@@ -211,28 +202,24 @@ def test_check_job_status_task_only_sends_old_tasks(mocker, sample_template):
         processing_started=datetime.utcnow() - timedelta(minutes=31),
         job_status=JOB_STATUS_IN_PROGRESS
     )
-    job_2 = create_job(
+    create_job(
         template=sample_template,
         notification_count=3,
         created_at=datetime.utcnow() - timedelta(minutes=31),
         processing_started=datetime.utcnow() - timedelta(minutes=29),
         job_status=JOB_STATUS_IN_PROGRESS
     )
-    with pytest.raises(expected_exception=JobIncompleteError) as e:
-        check_job_status()
-    assert str(job.id) in e.value.message
-    assert str(job_2.id) not in e.value.message
+    check_job_status()
 
     # job 2 not in celery task
     mock_celery.assert_called_once_with(
-        name=TaskNames.PROCESS_INCOMPLETE_JOBS,
-        args=([str(job.id)],),
+        [[str(job.id)]],
         queue=QueueNames.JOBS
     )
 
 
 def test_check_job_status_task_sets_jobs_to_error(mocker, sample_template):
-    mock_celery = mocker.patch('app.celery.tasks.notify_celery.send_task')
+    mock_celery = mocker.patch('app.celery.tasks.process_incomplete_jobs.apply_async')
     job = create_job(
         template=sample_template,
         notification_count=3,
@@ -248,15 +235,11 @@ def test_check_job_status_task_sets_jobs_to_error(mocker, sample_template):
         processing_started=datetime.utcnow() - timedelta(minutes=29),
         job_status=JOB_STATUS_IN_PROGRESS
     )
-    with pytest.raises(expected_exception=JobIncompleteError) as e:
-        check_job_status()
-    assert str(job.id) in e.value.message
-    assert str(job_2.id) not in e.value.message
+    check_job_status()
 
     # job 2 not in celery task
     mock_celery.assert_called_once_with(
-        name=TaskNames.PROCESS_INCOMPLETE_JOBS,
-        args=([str(job.id)],),
+        [[str(job.id)]],
         queue=QueueNames.JOBS
     )
     assert job.job_status == JOB_STATUS_ERROR

tests/app/dao/test_broadcast_message_dao.py

@@ -0,0 +1,43 @@
+from datetime import datetime
+from app.models import BROADCAST_TYPE
+from app.models import BroadcastEventMessageType
+from app.dao.broadcast_message_dao import get_earlier_events_for_broadcast_event
+
+from tests.app.db import create_broadcast_message, create_template, create_broadcast_event
+
+
+def test_get_earlier_events_for_broadcast_event(sample_service):
+    t = create_template(sample_service, BROADCAST_TYPE)
+    bm = create_broadcast_message(t)
+
+    events = [
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 12, 0, 0),
+            message_type=BroadcastEventMessageType.ALERT,
+            transmitted_content={'body': 'Initial content'}
+        ),
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 13, 0, 0),
+            message_type=BroadcastEventMessageType.UPDATE,
+            transmitted_content={'body': 'Updated content'}
+        ),
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 14, 0, 0),
+            message_type=BroadcastEventMessageType.UPDATE,
+            transmitted_content={'body': 'Updated content'},
+            transmitted_areas=['wales']
+        ),
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 15, 0, 0),
+            message_type=BroadcastEventMessageType.CANCEL,
+            transmitted_finishes_at=datetime(2020, 1, 1, 15, 0, 0),
+        )
+    ]
+
+    # only fetches earlier events, and they're in time order
+    earlier_events = get_earlier_events_for_broadcast_event(events[2].id)
+    assert earlier_events == [events[0], events[1]]

tests/app/db.py

@@ -62,10 +62,12 @@ from app.models import (
     ServiceContactList,
     BroadcastMessage,
     BroadcastStatusType,
+    BroadcastEvent
 )
 
 
 def create_user(
+    *,
     mobile_number="+447700900986",
     email="notify@digital.cabinet-office.gov.uk",
     state='active',
@@ -184,6 +186,7 @@ def create_template(
         folder=None,
         postage=None,
         process_type='normal',
+        contact_block_id=None
 ):
     data = {
         'name': template_name or '{} Template Name'.format(template_type),
@@ -194,10 +197,12 @@ def create_template(
         'reply_to': reply_to,
         'hidden': hidden,
         'folder': folder,
-        'process_type': process_type
+        'process_type': process_type,
     }
     if template_type == LETTER_TYPE:
         data["postage"] = postage or "second"
+        if contact_block_id:
+            data['service_letter_contact_id'] = contact_block_id
     if template_type != SMS_TYPE:
         data['subject'] = subject
     template = Template(**data)
@@ -1008,7 +1013,7 @@ def create_broadcast_message(
         template_id=template.id,
         template_version=template.version,
         personalisation=personalisation,
-        status=BroadcastStatusType.DRAFT,
+        status=status,
         starts_at=starts_at,
         finishes_at=finishes_at,
         created_by_id=created_by.id if created_by else template.created_by_id,
@@ -1017,3 +1022,29 @@ def create_broadcast_message(
     db.session.add(broadcast_message)
     db.session.commit()
     return broadcast_message
+
+
+def create_broadcast_event(
+    broadcast_message,
+    sent_at=None,
+    message_type='alert',
+    transmitted_content=None,
+    transmitted_areas=None,
+    transmitted_sender=None,
+    transmitted_starts_at=None,
+    transmitted_finishes_at=None,
+):
+    b_e = BroadcastEvent(
+        service=broadcast_message.service,
+        broadcast_message=broadcast_message,
+        sent_at=sent_at or datetime.utcnow(),
+        message_type=message_type,
+        transmitted_content=transmitted_content or {'body': 'this is an emergency broadcast message'},
+        transmitted_areas=transmitted_areas or ['london'],
+        transmitted_sender=transmitted_sender or 'www.notifications.service.gov.uk',
+        transmitted_starts_at=transmitted_starts_at,
+        transmitted_finishes_at=transmitted_finishes_at,
+    )
+    db.session.add(b_e)
+    db.session.commit()
+    return b_e

tests/app/service/test_service_contact_list_rest.py

@@ -69,7 +69,7 @@ def test_get_contact_list(admin_request, notify_db_session):
 
     assert len(response) == 1
     assert response[0] == contact_list.serialize()
-    assert response[0]['job_count'] == 0
+    assert response[0]['recent_job_count'] == 0
 
 
 @pytest.mark.parametrize('days_of_email_retention, expected_job_count', (
@@ -107,10 +107,12 @@ def test_get_contact_list_counts_jobs(
     assert len(response) == 2
 
     assert response[0]['id'] == str(contact_list_2.id)
-    assert response[0]['job_count'] == expected_job_count
+    assert response[0]['recent_job_count'] == expected_job_count
+    assert response[0]['has_jobs'] is True
 
     assert response[1]['id'] == str(contact_list_1.id)
-    assert response[1]['job_count'] == 0
+    assert response[1]['recent_job_count'] == 0
+    assert response[1]['has_jobs'] is False
 
 
 def test_get_contact_list_returns_for_service(admin_request, notify_db_session):

tests/app/v2/template/test_get_template.py

@@ -19,7 +19,12 @@ valid_version_params = [None, 1]
 def test_get_template_by_id_returns_200(
     client, sample_service, tmp_type, expected_name, expected_subject, version, postage
 ):
-    template = create_template(sample_service, template_type=tmp_type)
+    letter_contact_block_id = None
+    if tmp_type == 'letter':
+        letter_contact_block = create_letter_contact(sample_service, "Buckingham Palace, London, SW1A 1AA")
+        letter_contact_block_id = letter_contact_block.id
+
+    template = create_template(sample_service, template_type=tmp_type, contact_block_id=(letter_contact_block_id))
     auth_header = create_authorization_header(service_id=sample_service.id)
 
     version_path = '/version/{}'.format(version) if version else ''
@@ -44,6 +49,7 @@ def test_get_template_by_id_returns_200(
         'name': expected_name,
         'personalisation': {},
         'postage': postage,
+        'letter_contact_block': letter_contact_block.contact_block if letter_contact_block_id else None,
     }
 
     assert json_response == expected_response

tests/app/v2/test_errors.py

@@ -8,7 +8,7 @@ def app_for_test():
     import flask
     from flask import Blueprint
     from app.authentication.auth import AuthError
-    from app.v2.errors import BadRequestError, TooManyRequestsError, JobIncompleteError
+    from app.v2.errors import BadRequestError, TooManyRequestsError
     from app import init_app
 
     app = flask.Flask(__name__)
@@ -42,10 +42,6 @@ def app_for_test():
     def raising_data_error():
         raise DataError("There was a db problem", "params", "orig")
 
-    @blue.route("raise_job_incomplete_error", methods=["GET"])
-    def raising_job_incomplete_error():
-        raise JobIncompleteError("Raising job incomplete error")
-
     @blue.route("raise_exception", methods=["GET"])
     def raising_exception():
         raise AssertionError("Raising any old exception")
@@ -114,16 +110,6 @@ def test_data_errors(app_for_test):
                              "errors": [{"error": "DataError", "message": "No result found"}]}
 
 
-def test_job_incomplete_errors(app_for_test):
-    with app_for_test.test_request_context():
-        with app_for_test.test_client() as client:
-            response = client.get(url_for('v2_under_test.raising_job_incomplete_error'))
-            assert response.status_code == 500
-            error = response.json
-            assert error == {"status_code": 500,
-                             "errors": [{"error": "JobIncompleteError", "message": "Raising job incomplete error"}]}
-
-
 def test_internal_server_error_handler(app_for_test):
     with app_for_test.test_request_context():
         with app_for_test.test_client() as client:

tests/conftest.py

@@ -117,6 +117,7 @@ def notify_db_session(notify_db, sms_providers):
                             "organisation_types",
                             "service_permission_types",
                             "auth_type",
+                            "broadcast_status_type",
                             "invite_status_type",
                             "service_callback_type"]:
             notify_db.engine.execute(tbl.delete())