diff --git a/Makefile b/Makefile
index aac64e7c4..732d669a4 100644
--- a/Makefile
+++ b/Makefile
@@ -304,6 +304,7 @@ cf-rollback: ## Rollbacks the app to the previous release
 .PHONY: cf-push
 cf-push:
 	$(if ${CF_APP},,$(error Must specify CF_APP))
+	cf target -o ${CF_ORG} -s ${CF_SPACE}
 	cf push ${CF_APP} -f ${CF_MANIFEST_FILE}
 
 .PHONY: check-if-migrations-to-run
diff --git a/app/celery/tasks.py b/app/celery/tasks.py
index b5e732df5..c4ac1740a 100644
--- a/app/celery/tasks.py
+++ b/app/celery/tasks.py
@@ -326,6 +326,7 @@ def build_dvla_file(self, job_id):
             current_app.logger.info("All notifications for job {} are not persisted".format(job_id))
             self.retry(queue=QueueNames.RETRY, exc="All notifications for job {} are not persisted".format(job_id))
     except Exception as e:
+        # ? should this retry?
         current_app.logger.exception("build_dvla_file threw exception")
         raise e
 
diff --git a/app/commands.py b/app/commands.py
index a217f480a..f1e125e01 100644
--- a/app/commands.py
+++ b/app/commands.py
@@ -351,3 +351,14 @@ class PopulateAnnualBilling(Command):
             db.session.commit()
 
             print("Populated annual billing {} for {} services".format(fy, services_result1.rowcount))
+
+
+class ReRunBuildDvlaFileForJob(Command):
+    option_list = (
+        Option('-j', '--job_id', dest='job_id', help="Enter the job id to rebuild the dvla file for"),
+    )
+
+    def run(self, job_id):
+        from app.celery.tasks import build_dvla_file
+        from app.config import QueueNames
+        build_dvla_file.apply_async([job_id], queue=QueueNames.JOBS)
diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index 1c010c86b..c28b2a686 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -14,7 +14,7 @@ from notifications_utils.recipients import (
     InvalidEmailError,
 )
 from werkzeug.datastructures import MultiDict
-from sqlalchemy import (desc, func, or_, and_, asc)
+from sqlalchemy import (desc, func, or_, asc)
 from sqlalchemy.orm import joinedload
 from sqlalchemy.sql.expression import case
 from sqlalchemy.sql import functions
@@ -22,17 +22,15 @@ from notifications_utils.international_billing_rates import INTERNATIONAL_BILLIN
 
 from app import db, create_uuid
 from app.dao import days_ago
-from app.dao.date_util import get_financial_year
 from app.models import (
-    Service,
     Notification,
     NotificationEmailReplyTo,
     NotificationHistory,
-    NotificationStatistics,
     ScheduledNotification,
     ServiceEmailReplyTo,
     Template,
     EMAIL_TYPE,
+    SMS_TYPE,
     KEY_TYPE_NORMAL,
     KEY_TYPE_TEST,
     LETTER_TYPE,
@@ -43,69 +41,15 @@ from app.models import (
     NOTIFICATION_TECHNICAL_FAILURE,
     NOTIFICATION_TEMPORARY_FAILURE,
     NOTIFICATION_PERMANENT_FAILURE,
-    NOTIFICATION_SENT
+    NOTIFICATION_SENT,
+    NotificationSmsSender,
+    ServiceSmsSender
 )
 
 from app.dao.dao_utils import transactional
 from app.statsd_decorators import statsd
 
 
-def dao_get_notification_statistics_for_service_and_day(service_id, day):
-    # only used by stat-updating code in tasks.py
-    return NotificationStatistics.query.filter_by(
-        service_id=service_id,
-        day=day
-    ).order_by(desc(NotificationStatistics.day)).first()
-
-
-@statsd(namespace="dao")
-def dao_get_potential_notification_statistics_for_day(day):
-    all_services = db.session.query(
-        Service.id,
-        NotificationStatistics
-    ).outerjoin(
-        NotificationStatistics,
-        and_(
-            Service.id == NotificationStatistics.service_id,
-            or_(
-                NotificationStatistics.day == day,
-                NotificationStatistics.day == None  # noqa
-            )
-        )
-    ).order_by(
-        asc(Service.created_at)
-    )
-
-    notification_statistics = []
-    for service_notification_stats_pair in all_services:
-        if service_notification_stats_pair.NotificationStatistics:
-            notification_statistics.append(
-                service_notification_stats_pair.NotificationStatistics
-            )
-        else:
-            notification_statistics.append(
-                create_notification_statistics_dict(
-                    service_notification_stats_pair,
-                    day
-                )
-            )
-    return notification_statistics
-
-
-def create_notification_statistics_dict(service_id, day):
-    return {
-        'id': None,
-        'emails_requested': 0,
-        'emails_delivered': 0,
-        'emails_failed': 0,
-        'sms_requested': 0,
-        'sms_delivered': 0,
-        'sms_failed': 0,
-        'day': day.isoformat(),
-        'service': service_id
-    }
-
-
 @statsd(namespace="dao")
 def dao_get_template_usage(service_id, limit_days=None):
     query_filter = []
@@ -372,15 +316,19 @@ def delete_notifications_created_more_than_a_week_ago_by_type(notification_type)
     seven_days_ago = date.today() - timedelta(days=7)
 
     # Following could be refactored when NotificationSmsReplyTo and NotificationLetterContact in models.py
-    if notification_type == EMAIL_TYPE:
+    if notification_type in [EMAIL_TYPE, SMS_TYPE]:
         subq = db.session.query(Notification.id).filter(
             func.date(Notification.created_at) < seven_days_ago,
             Notification.notification_type == notification_type
         ).subquery()
-        deleted = db.session.query(
-            NotificationEmailReplyTo
+        if notification_type == EMAIL_TYPE:
+            notification_sender_mapping_table = NotificationEmailReplyTo
+        if notification_type == SMS_TYPE:
+            notification_sender_mapping_table = NotificationSmsSender
+        db.session.query(
+            notification_sender_mapping_table
         ).filter(
-            NotificationEmailReplyTo.notification_id.in_(subq)
+            notification_sender_mapping_table.notification_id.in_(subq)
         ).delete(synchronize_session='fetch')
 
     deleted = db.session.query(Notification).filter(
@@ -650,3 +598,23 @@ def dao_get_last_notification_added_for_job_id(job_id):
     ).first()
 
     return last_notification_added
+
+
+@transactional
+def dao_create_notification_sms_sender_mapping(notification_id, sms_sender_id):
+    notification_to_sms_sender = NotificationSmsSender(
+        notification_id=notification_id,
+        service_sms_sender_id=sms_sender_id
+    )
+    db.session.add(notification_to_sms_sender)
+
+
+def dao_get_notification_sms_sender_mapping(notification_id):
+    sms_sender = ServiceSmsSender.query.join(
+        NotificationSmsSender
+    ).filter(
+        NotificationSmsSender.notification_id == notification_id
+    ).first()
+
+    if sms_sender:
+        return sms_sender.sms_sender
diff --git a/app/dao/provider_statistics_dao.py b/app/dao/provider_statistics_dao.py
index 761e79de6..705d08fec 100644
--- a/app/dao/provider_statistics_dao.py
+++ b/app/dao/provider_statistics_dao.py
@@ -1,6 +1,7 @@
 from sqlalchemy import func
 
 from app import db
+from app.dao.date_util import get_financial_year
 from app.models import (
     NotificationHistory,
     SMS_TYPE,
@@ -8,7 +9,6 @@ from app.models import (
     NOTIFICATION_STATUS_TYPES_BILLABLE,
     KEY_TYPE_TEST
 )
-from app.dao.notifications_dao import get_financial_year
 
 
 def get_fragment_count(service_id, year=None):
diff --git a/app/dao/services_dao.py b/app/dao/services_dao.py
index e0a418230..93905e7e6 100644
--- a/app/dao/services_dao.py
+++ b/app/dao/services_dao.py
@@ -10,10 +10,9 @@ from app.dao.dao_utils import (
     transactional,
     version_class
 )
-from app.dao.notifications_dao import get_financial_year
+from app.dao.date_util import get_financial_year
 from app.dao.service_sms_sender_dao import insert_service_sms_sender
 from app.models import (
-    NotificationStatistics,
     ProviderStatistics,
     VerifyCode,
     ApiKey,
@@ -232,7 +231,6 @@ def delete_service_and_all_associated_db_objects(service):
     _delete_commit(TemplateRedacted.query.filter(TemplateRedacted.template_id.in_(subq)))
 
     _delete_commit(ServiceSmsSender.query.filter_by(service=service))
-    _delete_commit(NotificationStatistics.query.filter_by(service=service))
     _delete_commit(ProviderStatistics.query.filter_by(service=service))
     _delete_commit(InvitedUser.query.filter_by(service=service))
     _delete_commit(Permission.query.filter_by(service=service))
diff --git a/app/delivery/send_to_providers.py b/app/delivery/send_to_providers.py
index c253748ad..324c21582 100644
--- a/app/delivery/send_to_providers.py
+++ b/app/delivery/send_to_providers.py
@@ -11,8 +11,8 @@ from notifications_utils.template import HTMLEmailTemplate, PlainTextEmailTempla
 from app import clients, statsd_client, create_uuid
 from app.dao.notifications_dao import (
     dao_update_notification,
-    dao_get_notification_email_reply_for_notification
-)
+    dao_get_notification_email_reply_for_notification,
+    dao_get_notification_sms_sender_mapping)
 from app.dao.provider_details_dao import (
     get_provider_details_by_notification_type,
     dao_toggle_sms_provider
@@ -69,11 +69,15 @@ def send_sms_to_provider(notification):
                 raise
         else:
             try:
+                sms_sender = dao_get_notification_sms_sender_mapping(notification.id)
+                if not sms_sender:
+                    sms_sender = service.get_default_sms_sender()
+
                 provider.send_sms(
                     to=validate_and_format_phone_number(notification.to, international=notification.international),
                     content=str(template),
                     reference=str(notification.id),
-                    sender=service.get_default_sms_sender()
+                    sender=sms_sender
                 )
             except Exception as e:
                 dao_toggle_sms_provider(provider.name)
diff --git a/app/models.py b/app/models.py
index ff8cb1acf..76bf0edfd 100644
--- a/app/models.py
+++ b/app/models.py
@@ -516,25 +516,6 @@ class KeyTypes(db.Model):
     name = db.Column(db.String(255), primary_key=True)
 
 
-class NotificationStatistics(db.Model):
-    __tablename__ = 'notification_statistics'
-
-    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
-    day = db.Column(db.Date, index=True, nullable=False, unique=False, default=datetime.date.today)
-    service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, nullable=False)
-    service = db.relationship('Service', backref=db.backref('service_notification_stats', lazy='dynamic'))
-    emails_requested = db.Column(db.BigInteger, index=False, unique=False, nullable=False, default=0)
-    emails_delivered = db.Column(db.BigInteger, index=False, unique=False, nullable=False, default=0)
-    emails_failed = db.Column(db.BigInteger, index=False, unique=False, nullable=False, default=0)
-    sms_requested = db.Column(db.BigInteger, index=False, unique=False, nullable=False, default=0)
-    sms_delivered = db.Column(db.BigInteger, index=False, unique=False, nullable=False, default=0)
-    sms_failed = db.Column(db.BigInteger, index=False, unique=False, nullable=False, default=0)
-
-    __table_args__ = (
-        UniqueConstraint('service_id', 'day', name='uix_service_to_day'),
-    )
-
-
 class TemplateProcessTypes(db.Model):
     __tablename__ = 'template_process_type'
     name = db.Column(db.String(255), primary_key=True)
diff --git a/app/notifications/process_notifications.py b/app/notifications/process_notifications.py
index 21fda224e..76e18ebf4 100644
--- a/app/notifications/process_notifications.py
+++ b/app/notifications/process_notifications.py
@@ -17,7 +17,8 @@ from app.models import SMS_TYPE, Notification, KEY_TYPE_TEST, EMAIL_TYPE, NOTIFI
 from app.dao.notifications_dao import (dao_create_notification,
                                        dao_delete_notifications_and_history_by_id,
                                        dao_created_scheduled_notification,
-                                       dao_create_notification_email_reply_to_mapping)
+                                       dao_create_notification_email_reply_to_mapping,
+                                       dao_create_notification_sms_sender_mapping)
 from app.v2.errors import BadRequestError
 from app.utils import get_template_instance, cache_key_for_service_template_counter, convert_bst_to_utc
 
@@ -146,3 +147,7 @@ def persist_scheduled_notification(notification_id, scheduled_for):
 
 def persist_email_reply_to_id_for_notification(notification_id, email_reply_to_id):
     dao_create_notification_email_reply_to_mapping(notification_id, email_reply_to_id)
+
+
+def persist_sms_sender_id_for_notification(notification_id, sms_sender_id):
+    dao_create_notification_sms_sender_mapping(notification_id, sms_sender_id)
diff --git a/app/notifications/rest.py b/app/notifications/rest.py
index 85ee022a7..5a640b0e7 100644
--- a/app/notifications/rest.py
+++ b/app/notifications/rest.py
@@ -34,9 +34,7 @@ from app.schemas import (
     email_notification_schema,
     sms_template_notification_schema,
     notification_with_personalisation_schema,
-    notifications_filter_schema,
-    notifications_statistics_schema,
-    day_schema
+    notifications_filter_schema
 )
 from app.service.utils import service_allowed_to_send_to
 from app.utils import pagination_links, get_template_instance, get_public_notify_type_text
@@ -86,16 +84,6 @@ def get_all_notifications():
     ), 200
 
 
-@notifications.route('/notifications/statistics')
-def get_notification_statistics_for_day():
-    data = day_schema.load(request.args).data
-    statistics = notifications_dao.dao_get_potential_notification_statistics_for_day(
-        day=data['day']
-    )
-    data, errors = notifications_statistics_schema.dump(statistics, many=True)
-    return jsonify(data=data), 200
-
-
 @notifications.route('/notifications/<string:notification_type>', methods=['POST'])
 def send_notification(notification_type):
 
diff --git a/app/notifications/validators.py b/app/notifications/validators.py
index 181d48b6e..4f05977ce 100644
--- a/app/notifications/validators.py
+++ b/app/notifications/validators.py
@@ -8,6 +8,7 @@ from notifications_utils.recipients import (
 from notifications_utils.clients.redis import rate_limit_cache_key, daily_limit_cache_key
 
 from app.dao import services_dao, templates_dao
+from app.dao.service_sms_sender_dao import dao_get_service_sms_senders_by_id
 from app.models import (
     INTERNATIONAL_SMS_TYPE, SMS_TYPE, EMAIL_TYPE,
     KEY_TYPE_TEST, KEY_TYPE_TEAM, SCHEDULE_NOTIFICATIONS
@@ -135,11 +136,27 @@ def validate_template(template_id, personalisation, service, notification_type):
     return template, template_with_content
 
 
-def check_service_email_reply_to_id(service_id, reply_to_id):
-    if not (reply_to_id is None):
+def check_service_email_reply_to_id(service_id, reply_to_id, notification_type):
+    if reply_to_id:
+        if notification_type != EMAIL_TYPE:
+            message = 'email_reply_to_id is not a valid option for {} notification'.format(notification_type)
+            raise BadRequestError(message=message)
         try:
             dao_get_reply_to_by_id(service_id, reply_to_id)
         except NoResultFound:
             message = 'email_reply_to_id {} does not exist in database for service id {}'\
                 .format(reply_to_id, service_id)
             raise BadRequestError(message=message)
+
+
+def check_service_sms_sender_id(service_id, sms_sender_id, notification_type):
+    if sms_sender_id:
+        if notification_type != SMS_TYPE:
+            message = 'sms_sender_id is not a valid option for {} notification'.format(notification_type)
+            raise BadRequestError(message=message)
+        try:
+            dao_get_service_sms_senders_by_id(service_id, sms_sender_id)
+        except NoResultFound:
+            message = 'sms_sender_id {} does not exist in database for service id {}'\
+                .format(sms_sender_id, service_id)
+            raise BadRequestError(message=message)
diff --git a/app/schemas.py b/app/schemas.py
index b646b7c5c..a143e3bbb 100644
--- a/app/schemas.py
+++ b/app/schemas.py
@@ -107,6 +107,7 @@ class UserSchema(BaseSchema):
 
 
 class UserUpdateAttributeSchema(BaseSchema):
+    auth_type = field_for(models.User, 'auth_type')
 
     class Meta:
         model = models.User
@@ -324,18 +325,6 @@ class TemplateHistorySchema(BaseSchema):
         model = models.TemplateHistory
 
 
-class NotificationsStatisticsSchema(BaseSchema):
-    class Meta:
-        model = models.NotificationStatistics
-        strict = True
-
-    @pre_dump
-    def handle_date_str(self, in_data):
-        if isinstance(in_data, dict) and 'day' in in_data:
-            in_data['day'] = datetime.strptime(in_data['day'], '%Y-%m-%d').date()
-        return in_data
-
-
 class ApiKeySchema(BaseSchema):
 
     created_by = field_for(models.ApiKey, 'created_by', required=True)
@@ -672,7 +661,6 @@ notification_with_personalisation_schema = NotificationWithPersonalisationSchema
 invited_user_schema = InvitedUserSchema()
 permission_schema = PermissionSchema()
 email_data_request_schema = EmailDataSchema()
-notifications_statistics_schema = NotificationsStatisticsSchema()
 notifications_filter_schema = NotificationsFilterSchema()
 service_history_schema = ServiceHistorySchema()
 api_key_history_schema = ApiKeyHistorySchema()
diff --git a/app/service/send_notification.py b/app/service/send_notification.py
index d1ed9a5a3..fa6dd1bb2 100644
--- a/app/service/send_notification.py
+++ b/app/service/send_notification.py
@@ -2,13 +2,13 @@ from app.config import QueueNames
 from app.notifications.validators import (
     check_service_over_daily_message_limit,
     validate_and_format_recipient,
-    validate_template,
-    check_service_email_reply_to_id)
+    validate_template)
 from app.notifications.process_notifications import (
-    create_content_for_notification,
     persist_notification,
     send_notification_to_queue,
-    persist_email_reply_to_id_for_notification)
+    persist_email_reply_to_id_for_notification,
+    persist_sms_sender_id_for_notification
+)
 from app.models import (
     KEY_TYPE_NORMAL,
     PRIORITY,
@@ -64,9 +64,11 @@ def send_one_off_notification(service_id, post_data):
         created_by_id=post_data['created_by']
     )
     sender_id = post_data.get('sender_id', None)
-    if sender_id and template.template_type == EMAIL_TYPE:
-        check_service_email_reply_to_id(service_id, sender_id)
-        persist_email_reply_to_id_for_notification(notification.id, sender_id)
+    if sender_id:
+        if template.template_type == EMAIL_TYPE:
+            persist_email_reply_to_id_for_notification(notification.id, sender_id)
+        if template.template_type == SMS_TYPE:
+            persist_sms_sender_id_for_notification(notification.id, sender_id)
 
     queue_name = QueueNames.PRIORITY if template.process_type == PRIORITY else None
     send_notification_to_queue(
diff --git a/app/service/utils.py b/app/service/utils.py
index 475d420e2..687f8939e 100644
--- a/app/service/utils.py
+++ b/app/service/utils.py
@@ -1,12 +1,13 @@
 import itertools
 
+from app.dao.date_util import get_financial_year
 from app.models import (
     ServiceWhitelist,
     MOBILE_TYPE, EMAIL_TYPE,
     KEY_TYPE_TEST, KEY_TYPE_TEAM, KEY_TYPE_NORMAL)
 
 from notifications_utils.recipients import allowed_to_send_to
-from app.dao.notifications_dao import get_financial_year
+
 from datetime import datetime
 
 
diff --git a/app/v2/notifications/notification_schemas.py b/app/v2/notifications/notification_schemas.py
index 1689a7319..258b5df0d 100644
--- a/app/v2/notifications/notification_schemas.py
+++ b/app/v2/notifications/notification_schemas.py
@@ -119,7 +119,8 @@ post_sms_request = {
         "phone_number": {"type": "string", "format": "phone_number"},
         "template_id": uuid,
         "personalisation": personalisation,
-        "scheduled_for": {"type": ["string", "null"], "format": "datetime"}
+        "scheduled_for": {"type": ["string", "null"], "format": "datetime"},
+        "sms_sender_id": uuid
     },
     "required": ["phone_number", "template_id"]
 }
diff --git a/app/v2/notifications/post_notifications.py b/app/v2/notifications/post_notifications.py
index b4c837306..87d55159d 100644
--- a/app/v2/notifications/post_notifications.py
+++ b/app/v2/notifications/post_notifications.py
@@ -20,8 +20,8 @@ from app.notifications.process_notifications import (
     persist_notification,
     persist_scheduled_notification,
     send_notification_to_queue,
-    simulated_recipient
-)
+    simulated_recipient,
+    persist_sms_sender_id_for_notification)
 from app.notifications.process_letter_notifications import (
     create_letter_notification
 )
@@ -31,7 +31,8 @@ from app.notifications.validators import (
     check_service_can_schedule_notification,
     check_service_has_permission,
     validate_template,
-    check_service_email_reply_to_id
+    check_service_email_reply_to_id,
+    check_service_sms_sender_id
 )
 from app.schema_validation import validate
 from app.v2.errors import BadRequestError
@@ -63,12 +64,14 @@ def post_notification(notification_type):
 
     scheduled_for = form.get("scheduled_for", None)
     service_email_reply_to_id = form.get("email_reply_to_id", None)
+    service_sms_sender_id = form.get("sms_sender_id", None)
 
     check_service_can_schedule_notification(authenticated_service.permissions, scheduled_for)
 
     check_rate_limiting(authenticated_service, api_user)
 
-    check_service_email_reply_to_id(str(authenticated_service.id), service_email_reply_to_id)
+    check_service_email_reply_to_id(str(authenticated_service.id), service_email_reply_to_id, notification_type)
+    check_service_sms_sender_id(str(authenticated_service.id), service_sms_sender_id, notification_type)
 
     template, template_with_content = validate_template(
         form['template_id'],
@@ -142,9 +145,7 @@ def process_sms_or_email_notification(*, form, notification_type, api_key, templ
         simulated=simulated
     )
 
-    email_reply_to_id = form.get("email_reply_to_id", None)
-    if email_reply_to_id:
-        persist_email_reply_to_id_for_notification(notification.id, email_reply_to_id)
+    persist_sender_to_notification_mapping(form, notification)
 
     scheduled_for = form.get("scheduled_for", None)
     if scheduled_for:
@@ -163,6 +164,15 @@ def process_sms_or_email_notification(*, form, notification_type, api_key, templ
     return notification
 
 
+def persist_sender_to_notification_mapping(form, notification):
+    email_reply_to_id = form.get("email_reply_to_id", None)
+    if email_reply_to_id:
+        persist_email_reply_to_id_for_notification(notification.id, email_reply_to_id)
+    sms_sender_id = form.get("sms_sender_id", None)
+    if sms_sender_id:
+        persist_sms_sender_id_for_notification(notification.id, sms_sender_id)
+
+
 def process_letter_notification(*, letter_data, api_key, template):
     if api_key.key_type == KEY_TYPE_TEAM:
         raise BadRequestError(message='Cannot send letters with a team api key', status_code=403)
diff --git a/application.py b/application.py
index fb4f815c7..b606aee81 100644
--- a/application.py
+++ b/application.py
@@ -23,8 +23,8 @@ manager.add_command('populate_service_sms_sender', commands.PopulateServiceSmsSe
 manager.add_command('populate_service_letter_contact', commands.PopulateServiceLetterContact)
 manager.add_command('populate_service_and_service_history_free_sms_fragment_limit',
                     commands.PopulateServiceAndServiceHistoryFreeSmsFragmentLimit)
-manager.add_command('populate_annual_billing',
-                    commands.PopulateAnnualBilling)
+manager.add_command('populate_annual_billing', commands.PopulateAnnualBilling)
+manager.add_command('rerun_build_dvla_file', commands.ReRunBuildDvlaFileForJob)
 
 
 @manager.command
diff --git a/requirements.txt b/requirements.txt
index 388c38300..8b9d8624f 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -12,7 +12,7 @@ gunicorn==19.7.1
 iso8601==0.1.12
 jsonschema==2.6.0
 marshmallow-sqlalchemy==0.13.2
-marshmallow==2.13.6
+marshmallow==2.14.0
 monotonic==1.4
 psycopg2==2.7.3.2
 PyJWT==1.5.3
@@ -26,6 +26,6 @@ notifications-python-client==4.5.0
 awscli>=1.11,<1.12
 awscli-cwlogs>=1.4,<1.5
 
-git+https://github.com/alphagov/notifications-utils.git@21.5.0#egg=notifications-utils==21.5.0
+git+https://github.com/alphagov/notifications-utils.git@21.5.1#egg=notifications-utils==21.5.1
 
 git+https://github.com/alphagov/boto.git@2.43.0-patch3#egg=boto==2.43.0-patch3
diff --git a/tests/app/conftest.py b/tests/app/conftest.py
index 7f3bd3008..e03b2b7dd 100644
--- a/tests/app/conftest.py
+++ b/tests/app/conftest.py
@@ -24,7 +24,6 @@ from app.models import (
     ProviderDetails,
     ProviderDetailsHistory,
     ProviderRates,
-    NotificationStatistics,
     ScheduledNotification,
     ServiceWhitelist,
     KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM,
@@ -840,35 +839,6 @@ def sample_provider_statistics(notify_db,
     return stats
 
 
-@pytest.fixture(scope='function')
-def sample_notification_statistics(notify_db,
-                                   notify_db_session,
-                                   service=None,
-                                   day=None,
-                                   emails_requested=2,
-                                   emails_delivered=1,
-                                   emails_failed=1,
-                                   sms_requested=2,
-                                   sms_delivered=1,
-                                   sms_failed=1):
-    if service is None:
-        service = sample_service(notify_db, notify_db_session)
-    if day is None:
-        day = date.today()
-    stats = NotificationStatistics(
-        service=service,
-        day=day,
-        emails_requested=emails_requested,
-        emails_delivered=emails_delivered,
-        emails_failed=emails_failed,
-        sms_requested=sms_requested,
-        sms_delivered=sms_delivered,
-        sms_failed=sms_failed)
-    notify_db.session.add(stats)
-    notify_db.session.commit()
-    return stats
-
-
 @pytest.fixture(scope='function')
 def mock_firetext_client(mocker, statsd_client=None):
     client = FiretextClient()
diff --git a/tests/app/dao/notification_dao/test_notification_dao.py b/tests/app/dao/notification_dao/test_notification_dao.py
index 562fd8555..cec15552d 100644
--- a/tests/app/dao/notification_dao/test_notification_dao.py
+++ b/tests/app/dao/notification_dao/test_notification_dao.py
@@ -12,10 +12,10 @@ from app.models import (
     Notification,
     NotificationEmailReplyTo,
     NotificationHistory,
-    NotificationStatistics,
+    NotificationSmsSender,
     ScheduledNotification,
-    ServiceEmailReplyTo,
     EMAIL_TYPE,
+    SMS_TYPE,
     NOTIFICATION_STATUS_TYPES,
     NOTIFICATION_STATUS_TYPES_FAILED,
     NOTIFICATION_SENT,
@@ -23,23 +23,26 @@ from app.models import (
     KEY_TYPE_NORMAL,
     KEY_TYPE_TEAM,
     KEY_TYPE_TEST,
-    JOB_STATUS_IN_PROGRESS)
+    JOB_STATUS_IN_PROGRESS
+)
 
 from app.dao.notifications_dao import (
     dao_create_notification,
     dao_create_notification_email_reply_to_mapping,
+    dao_create_notification_sms_sender_mapping,
     dao_created_scheduled_notification,
     dao_delete_notifications_and_history_by_id,
+    dao_get_last_notification_added_for_job_id,
     dao_get_last_template_usage,
     dao_get_notification_email_reply_for_notification,
     dao_get_notifications_by_to_field,
-    dao_get_notification_statistics_for_service_and_day,
-    dao_get_potential_notification_statistics_for_day,
+    dao_get_notification_sms_sender_mapping,
     dao_get_scheduled_notifications,
     dao_get_template_usage,
     dao_timeout_notifications,
     dao_update_notification,
     dao_update_notifications_for_job_to_sent_to_dvla,
+    dao_update_notifications_by_reference,
     delete_notifications_created_more_than_a_week_ago_by_type,
     get_notification_by_id,
     get_notification_for_job,
@@ -50,16 +53,16 @@ from app.dao.notifications_dao import (
     is_delivery_slow_for_provider,
     set_scheduled_notification_to_processed,
     update_notification_status_by_id,
-    update_notification_status_by_reference,
-    dao_get_last_notification_added_for_job_id, dao_update_notifications_by_reference)
+    update_notification_status_by_reference
+)
 
 from app.dao.services_dao import dao_update_service
 from tests.app.db import (
     create_api_key,
     create_job,
     create_notification,
-    create_reply_to_email
-)
+    create_reply_to_email,
+    create_service_sms_sender)
 from tests.app.conftest import (
     sample_notification,
     sample_template,
@@ -73,7 +76,6 @@ from tests.app.conftest import (
 def test_should_have_decorated_notifications_dao_functions():
     assert dao_get_last_template_usage.__wrapped__.__name__ == 'dao_get_last_template_usage'  # noqa
     assert dao_get_template_usage.__wrapped__.__name__ == 'dao_get_template_usage'  # noqa
-    assert dao_get_potential_notification_statistics_for_day.__wrapped__.__name__ == 'dao_get_potential_notification_statistics_for_day'  # noqa
     assert dao_create_notification.__wrapped__.__name__ == 'dao_create_notification'  # noqa
     assert update_notification_status_by_id.__wrapped__.__name__ == 'update_notification_status_by_id'  # noqa
     assert dao_update_notification.__wrapped__.__name__ == 'dao_update_notification'  # noqa
@@ -575,31 +577,10 @@ def test_should_return_zero_count_if_no_notification_with_reference():
     assert not update_notification_status_by_reference('something', 'delivered')
 
 
-def test_should_return_none_if_no_statistics_for_a_service_for_a_day(sample_template, mmg_provider):
-    data = _notification_json(sample_template)
-
-    notification = Notification(**data)
-    dao_create_notification(notification)
-    assert not dao_get_notification_statistics_for_service_and_day(
-        sample_template.service.id, (datetime.utcnow() - timedelta(days=1)).date())
-
-
-def test_should_be_able_to_get_all_statistics_for_a_service(sample_template, mmg_provider):
-    data = _notification_json(sample_template)
-
-    notification_1 = Notification(**data)
-    notification_2 = Notification(**data)
-    notification_3 = Notification(**data)
-    dao_create_notification(notification_1)
-    dao_create_notification(notification_2)
-    dao_create_notification(notification_3)
-
-
 def test_create_notification_creates_notification_with_personalisation(notify_db, notify_db_session,
                                                                        sample_template_with_placeholders,
                                                                        sample_job, mmg_provider):
     assert Notification.query.count() == 0
-    assert NotificationStatistics.query.count() == 0
 
     data = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
                                template=sample_template_with_placeholders,
@@ -622,7 +603,6 @@ def test_create_notification_creates_notification_with_personalisation(notify_db
 
 def test_save_notification_creates_sms(sample_template, sample_job, mmg_provider):
     assert Notification.query.count() == 0
-    assert NotificationStatistics.query.count() == 0
 
     data = _notification_json(sample_template, job_id=sample_job.id)
 
@@ -643,7 +623,6 @@ def test_save_notification_creates_sms(sample_template, sample_job, mmg_provider
 
 def test_save_notification_and_create_email(sample_email_template, sample_job):
     assert Notification.query.count() == 0
-    assert NotificationStatistics.query.count() == 0
 
     data = _notification_json(sample_email_template, job_id=sample_job.id)
 
@@ -767,7 +746,6 @@ def test_not_save_notification_and_not_create_stats_on_commit_error(sample_templ
 
     assert Notification.query.count() == 0
     assert Job.query.get(sample_job.id).notifications_sent == 0
-    assert NotificationStatistics.query.count() == 0
 
 
 def test_save_notification_and_increment_job(sample_template, sample_job, mmg_provider):
@@ -1042,6 +1020,38 @@ def test_should_delete_notification_to_email_reply_to_after_seven_days(
         assert notification.created_at.date() >= date(2016, 1, 3)
 
 
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_should_delete_notification_to_sms_sender_after_seven_days(
+    sample_template
+):
+    assert len(Notification.query.all()) == 0
+
+    sms_sender = create_service_sms_sender(service=sample_template.service, sms_sender='123456', is_default=False)
+
+    # create one notification a day between 1st and 10th from 11:00 to 19:00 of each type
+    for i in range(1, 11):
+        past_date = '2016-01-{0:02d}  {0:02d}:00:00.000000'.format(i)
+        with freeze_time(past_date):
+            notification = create_notification(template=sample_template, sms_sender_id=sms_sender.id)
+
+    all_notifications = Notification.query.all()
+    assert len(all_notifications) == 10
+
+    all_notification_sms_senders = NotificationSmsSender.query.all()
+    assert len(all_notification_sms_senders) == 10
+
+    # Records before 3rd should be deleted
+    delete_notifications_created_more_than_a_week_ago_by_type(SMS_TYPE)
+    remaining_notifications = Notification.query.filter_by(notification_type=SMS_TYPE).all()
+    remaining_notification_to_sms_sender = NotificationSmsSender.query.filter_by().all()
+
+    assert len(remaining_notifications) == 8
+    assert len(remaining_notification_to_sms_sender) == 8
+
+    for notification in remaining_notifications:
+        assert notification.created_at.date() >= date(2016, 1, 3)
+
+
 @pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
 @freeze_time("2016-01-10 12:00:00.000000")
 def test_should_not_delete_notification_history(notify_db, notify_db_session, sample_service, notification_type):
@@ -2100,3 +2110,26 @@ def test_dao_update_notifications_by_reference_returns_zero_when_no_notification
                                                                        "billable_units": 2}
                                                           )
     assert updated_count == 0
+
+
+def test_dao_create_notification_sms_sender_mapping(sample_notification):
+    sms_sender = create_service_sms_sender(service=sample_notification.service, sms_sender='123456')
+    dao_create_notification_sms_sender_mapping(notification_id=sample_notification.id,
+                                               sms_sender_id=sms_sender.id)
+    notification_to_senders = NotificationSmsSender.query.all()
+    assert len(notification_to_senders) == 1
+    assert notification_to_senders[0].notification_id == sample_notification.id
+    assert notification_to_senders[0].service_sms_sender_id == sms_sender.id
+
+
+def test_dao_get_notification_sms_sender_mapping(sample_notification):
+    sms_sender = create_service_sms_sender(service=sample_notification.service, sms_sender='123456')
+    dao_create_notification_sms_sender_mapping(notification_id=sample_notification.id,
+                                               sms_sender_id=sms_sender.id)
+    notification_to_sender = dao_get_notification_sms_sender_mapping(sample_notification.id)
+    assert notification_to_sender == '123456'
+
+
+def test_dao_get_notification_sms_sender_mapping_returns_none(sample_notification):
+    notification_to_sender = dao_get_notification_sms_sender_mapping(sample_notification.id)
+    assert not notification_to_sender
diff --git a/tests/app/dao/test_services_dao.py b/tests/app/dao/test_services_dao.py
index 70bc79eef..76c06d088 100644
--- a/tests/app/dao/test_services_dao.py
+++ b/tests/app/dao/test_services_dao.py
@@ -35,7 +35,6 @@ from app.dao.services_dao import (
 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.models import (
-    NotificationStatistics,
     ProviderStatistics,
     VerifyCode,
     ApiKey,
@@ -447,7 +446,6 @@ def test_delete_service_and_associated_objects(notify_db,
     assert ServicePermission.query.count() == 3
 
     delete_service_and_all_associated_db_objects(sample_service)
-    assert NotificationStatistics.query.count() == 0
     assert ProviderStatistics.query.count() == 0
     assert VerifyCode.query.count() == 0
     assert ApiKey.query.count() == 0
diff --git a/tests/app/db.py b/tests/app/db.py
index 0cdf7cfee..4be3a3799 100644
--- a/tests/app/db.py
+++ b/tests/app/db.py
@@ -29,7 +29,11 @@ from app.models import (
     KEY_TYPE_NORMAL
 )
 from app.dao.users_dao import save_model_user
-from app.dao.notifications_dao import dao_create_notification, dao_created_scheduled_notification
+from app.dao.notifications_dao import (
+    dao_create_notification,
+    dao_created_scheduled_notification,
+    dao_create_notification_sms_sender_mapping
+)
 from app.dao.templates_dao import dao_create_template
 from app.dao.services_dao import dao_create_service
 from app.dao.service_permissions_dao import dao_add_service_permission
@@ -128,6 +132,7 @@ def create_notification(
     scheduled_for=None,
     normalised_to=None,
     one_off=False,
+    sms_sender_id=None
 ):
     if created_at is None:
         created_at = datetime.utcnow()
@@ -184,6 +189,9 @@ def create_notification(
         if status != 'created':
             scheduled_notification.pending = False
         dao_created_scheduled_notification(scheduled_notification)
+    if sms_sender_id:
+        dao_create_notification_sms_sender_mapping(notification_id=notification.id,
+                                                   sms_sender_id=sms_sender_id)
     return notification
 
 
diff --git a/tests/app/delivery/test_send_to_providers.py b/tests/app/delivery/test_send_to_providers.py
index 6733c5846..2d6a10cdf 100644
--- a/tests/app/delivery/test_send_to_providers.py
+++ b/tests/app/delivery/test_send_to_providers.py
@@ -32,8 +32,8 @@ from tests.app.db import (
     create_notification,
     create_inbound_number,
     create_reply_to_email,
-    create_reply_to_email_for_notification
-)
+    create_reply_to_email_for_notification,
+    create_service_sms_sender)
 
 
 def test_should_return_highest_priority_active_provider(restore_provider_details):
@@ -343,6 +343,28 @@ def test_should_send_sms_with_downgraded_content(notify_db_session, mocker):
     )
 
 
+def test_send_sms_should_use_service_sms_sender(
+        sample_service,
+        sample_template,
+        mocker):
+    mocker.patch('app.mmg_client.send_sms')
+    mocker.patch('app.delivery.send_to_providers.create_initial_notification_statistic_tasks')
+
+    sms_sender = create_service_sms_sender(service=sample_service, sms_sender='123456', is_default=False)
+    db_notification = create_notification(template=sample_template, sms_sender_id=sms_sender.id)
+
+    send_to_providers.send_sms_to_provider(
+        db_notification,
+    )
+
+    app.mmg_client.send_sms.assert_called_once_with(
+        to=ANY,
+        content=ANY,
+        reference=ANY,
+        sender=sms_sender.sms_sender
+    )
+
+
 @pytest.mark.parametrize('research_mode,key_type', [
     (True, KEY_TYPE_NORMAL),
     (False, KEY_TYPE_TEST)
diff --git a/tests/app/notifications/rest/test_callbacks.py b/tests/app/notifications/rest/test_callbacks.py
index 03fede0fa..ff2cb417e 100644
--- a/tests/app/notifications/rest/test_callbacks.py
+++ b/tests/app/notifications/rest/test_callbacks.py
@@ -9,7 +9,6 @@ import app.celery.tasks
 from app.dao.notifications_dao import (
     get_notification_by_id
 )
-from app.models import NotificationStatistics
 from tests.app.notifications.test_notifications_ses_callback import ses_confirmation_callback
 from tests.app.conftest import sample_notification as create_sample_notification
 
@@ -431,10 +430,6 @@ def test_firetext_callback_should_record_statsd(client, notify_db, notify_db_ses
         app.statsd_client.incr.assert_any_call("callback.firetext.delivered")
 
 
-def get_notification_stats(service_id):
-    return NotificationStatistics.query.filter_by(service_id=service_id).one()
-
-
 def _sample_sns_s3_callback():
     return json.dumps({
         "SigningCertURL": "foo.pem",
diff --git a/tests/app/notifications/rest/test_notification_statistics.py b/tests/app/notifications/rest/test_notification_statistics.py
deleted file mode 100644
index 6e9de437a..000000000
--- a/tests/app/notifications/rest/test_notification_statistics.py
+++ /dev/null
@@ -1,225 +0,0 @@
-from datetime import date, timedelta
-
-from flask import json
-from freezegun import freeze_time
-from datetime import datetime
-
-from tests import create_authorization_header
-from tests.app.conftest import (
-    sample_notification_statistics as create_sample_notification_statistics,
-    sample_service as create_sample_service
-)
-
-
-def test_get_notification_statistics(notify_api, sample_notification_statistics):
-    with notify_api.test_request_context():
-        with notify_api.test_client() as client:
-            auth_header = create_authorization_header(
-                service_id=sample_notification_statistics.service_id
-            )
-
-            response = client.get(
-                '/notifications/statistics?day={}'.format(date.today().isoformat()),
-                headers=[auth_header]
-            )
-
-            notifications = json.loads(response.get_data(as_text=True))
-
-            assert len(notifications['data']) == 1
-            stats = notifications['data'][0]
-            assert stats['emails_requested'] == 2
-            assert stats['emails_delivered'] == 1
-            assert stats['emails_failed'] == 1
-            assert stats['sms_requested'] == 2
-            assert stats['sms_delivered'] == 1
-            assert stats['sms_failed'] == 1
-            assert stats['service'] == str(sample_notification_statistics.service_id)
-            assert response.status_code == 200
-
-
-@freeze_time('1955-11-05T12:00:00')
-def test_get_notification_statistics_only_returns_today(notify_api, notify_db, notify_db_session, sample_service):
-    with notify_api.test_request_context():
-        with notify_api.test_client() as client:
-            yesterdays_notification_statistics = create_sample_notification_statistics(
-                notify_db,
-                notify_db_session,
-                service=sample_service,
-                day=date.today() - timedelta(days=1)
-            )
-            todays_notification_statistics = create_sample_notification_statistics(
-                notify_db,
-                notify_db_session,
-                service=sample_service,
-                day=date.today()
-            )
-            tomorrows_notification_statistics = create_sample_notification_statistics(
-                notify_db,
-                notify_db_session,
-                service=sample_service,
-                day=date.today() + timedelta(days=1)
-            )
-
-            auth_header = create_authorization_header(
-                service_id=sample_service.id
-            )
-
-            response = client.get(
-                '/notifications/statistics?day={}'.format(date.today().isoformat()),
-                headers=[auth_header]
-            )
-
-            notifications = json.loads(response.get_data(as_text=True))
-
-            assert len(notifications['data']) == 1
-            assert notifications['data'][0]['day'] == date.today().isoformat()
-            assert response.status_code == 200
-
-
-def test_get_notification_statistics_fails_if_no_date(notify_api, sample_notification_statistics):
-    with notify_api.test_request_context():
-        with notify_api.test_client() as client:
-            auth_header = create_authorization_header(
-                service_id=sample_notification_statistics.service_id
-            )
-
-            response = client.get(
-                '/notifications/statistics',
-                headers=[auth_header]
-            )
-
-            resp = json.loads(response.get_data(as_text=True))
-            assert resp['result'] == 'error'
-            assert resp['message'] == {'day': ['Missing data for required field.']}
-            assert response.status_code == 400
-
-
-def test_get_notification_statistics_fails_if_invalid_date(notify_api, sample_notification_statistics):
-    with notify_api.test_request_context():
-        with notify_api.test_client() as client:
-            auth_header = create_authorization_header(
-                service_id=sample_notification_statistics.service_id
-            )
-
-            response = client.get(
-                '/notifications/statistics?day=2016-99-99',
-                headers=[auth_header]
-            )
-
-            resp = json.loads(response.get_data(as_text=True))
-            assert resp['result'] == 'error'
-            assert resp['message'] == {'day': ['Not a valid date.']}
-            assert response.status_code == 400
-
-
-def test_get_notification_statistics_returns_zeros_if_not_in_db(notify_api, sample_service):
-    with notify_api.test_request_context():
-        with notify_api.test_client() as client:
-            auth_header = create_authorization_header(
-                service_id=sample_service.id
-            )
-
-            response = client.get(
-                '/notifications/statistics?day={}'.format(date.today().isoformat()),
-                headers=[auth_header]
-            )
-
-            notifications = json.loads(response.get_data(as_text=True))
-
-            assert len(notifications['data']) == 1
-            stats = notifications['data'][0]
-            assert stats['emails_requested'] == 0
-            assert stats['emails_delivered'] == 0
-            assert stats['emails_failed'] == 0
-            assert stats['sms_requested'] == 0
-            assert stats['sms_delivered'] == 0
-            assert stats['sms_failed'] == 0
-            assert stats['service'] == str(sample_service.id)
-            assert response.status_code == 200
-
-
-def test_get_notification_statistics_returns_both_existing_stats_and_generated_zeros(
-    notify_api,
-    notify_db,
-    notify_db_session
-):
-    with notify_api.test_request_context():
-        with notify_api.test_client() as client:
-            service_with_stats = create_sample_service(
-                notify_db,
-                notify_db_session,
-                service_name='service_with_stats',
-                email_from='service_with_stats'
-            )
-            service_without_stats = create_sample_service(
-                notify_db,
-                notify_db_session,
-                service_name='service_without_stats',
-                email_from='service_without_stats'
-            )
-            notification_statistics = create_sample_notification_statistics(
-                notify_db,
-                notify_db_session,
-                service=service_with_stats,
-                day=date.today()
-            )
-            auth_header = create_authorization_header(
-                service_id=service_with_stats.id
-            )
-
-            response = client.get(
-                '/notifications/statistics?day={}'.format(date.today().isoformat()),
-                headers=[auth_header]
-            )
-
-            notifications = json.loads(response.get_data(as_text=True))
-
-            assert len(notifications['data']) == 2
-            retrieved_stats = notifications['data'][0]
-            generated_stats = notifications['data'][1]
-
-            assert retrieved_stats['emails_requested'] == 2
-            assert retrieved_stats['emails_delivered'] == 1
-            assert retrieved_stats['emails_failed'] == 1
-            assert retrieved_stats['sms_requested'] == 2
-            assert retrieved_stats['sms_delivered'] == 1
-            assert retrieved_stats['sms_failed'] == 1
-            assert retrieved_stats['service'] == str(service_with_stats.id)
-
-            assert generated_stats['emails_requested'] == 0
-            assert generated_stats['emails_delivered'] == 0
-            assert generated_stats['emails_failed'] == 0
-            assert generated_stats['sms_requested'] == 0
-            assert generated_stats['sms_delivered'] == 0
-            assert generated_stats['sms_failed'] == 0
-            assert generated_stats['service'] == str(service_without_stats.id)
-
-            assert response.status_code == 200
-
-
-def test_get_notification_statistics_returns_zeros_when_only_stats_for_different_date(
-    notify_api,
-    sample_notification_statistics
-):
-    with notify_api.test_request_context():
-        with notify_api.test_client() as client:
-            with freeze_time('1985-10-26T00:06:00'):
-                auth_header = create_authorization_header(
-                    service_id=sample_notification_statistics.service_id
-                )
-                response = client.get(
-                    '/notifications/statistics?day={}'.format(datetime.utcnow().isoformat()),
-                    headers=[auth_header]
-                )
-
-            notifications = json.loads(response.get_data(as_text=True))
-
-            assert response.status_code == 200
-            assert len(notifications['data']) == 1
-            assert notifications['data'][0]['emails_requested'] == 0
-            assert notifications['data'][0]['emails_delivered'] == 0
-            assert notifications['data'][0]['emails_failed'] == 0
-            assert notifications['data'][0]['sms_requested'] == 0
-            assert notifications['data'][0]['sms_delivered'] == 0
-            assert notifications['data'][0]['sms_failed'] == 0
-            assert notifications['data'][0]['service'] == str(sample_notification_statistics.service_id)
diff --git a/tests/app/notifications/test_process_notification.py b/tests/app/notifications/test_process_notification.py
index e8e67568b..ff9521906 100644
--- a/tests/app/notifications/test_process_notification.py
+++ b/tests/app/notifications/test_process_notification.py
@@ -8,19 +8,28 @@ from freezegun import freeze_time
 from collections import namedtuple
 
 from app.dao.service_email_reply_to_dao import dao_get_reply_to_by_service_id
-from app.models import Template, Notification, NotificationHistory, ScheduledNotification, NotificationEmailReplyTo
+from app.models import (
+    Notification,
+    NotificationHistory,
+    NotificationEmailReplyTo,
+    NotificationSmsSender,
+    ScheduledNotification,
+    Template
+)
 from app.notifications.process_notifications import (
     create_content_for_notification,
     persist_notification,
-    send_notification_to_queue,
-    simulated_recipient,
+    persist_email_reply_to_id_for_notification,
     persist_scheduled_notification,
-    persist_email_reply_to_id_for_notification)
+    persist_sms_sender_id_for_notification,
+    send_notification_to_queue,
+    simulated_recipient
+)
 from notifications_utils.recipients import validate_and_format_phone_number, validate_and_format_email_address
 from app.utils import cache_key_for_service_template_counter
 from app.v2.errors import BadRequestError
 from tests.app.conftest import sample_api_key as create_api_key
-from tests.app.db import create_reply_to_email
+from tests.app.db import create_reply_to_email, create_service_sms_sender
 
 
 def test_create_content_for_notification_passes(sample_email_template):
@@ -452,3 +461,15 @@ def test_persist_email_reply_to_id_for_notification(sample_service, sample_notif
     assert len(email_reply_to) == 1
     assert email_reply_to[0].notification_id == sample_notification.id
     assert email_reply_to[0].service_email_reply_to_id == reply_to_address[0].id
+
+
+def test_persist_sms_sender_id_for_notification(sample_service, sample_notification):
+    sms_sender = create_service_sms_sender(service=sample_service, sms_sender="123456")
+    persist_sms_sender_id_for_notification(notification_id=sample_notification.id,
+                                           sms_sender_id=sms_sender.id)
+
+    notification_to_sms_sender = NotificationSmsSender.query.all()
+
+    assert len(notification_to_sms_sender) == 1
+    assert notification_to_sms_sender[0].notification_id == sample_notification.id
+    assert notification_to_sms_sender[0].service_sms_sender_id == sms_sender.id
diff --git a/tests/app/notifications/test_validators.py b/tests/app/notifications/test_validators.py
index fecfc4da5..ede6d940d 100644
--- a/tests/app/notifications/test_validators.py
+++ b/tests/app/notifications/test_validators.py
@@ -11,8 +11,8 @@ from app.notifications.validators import (
     check_sms_content_char_count,
     check_service_over_api_rate_limit,
     validate_and_format_recipient,
-    check_service_email_reply_to_id
-)
+    check_service_email_reply_to_id,
+    check_service_sms_sender_id)
 
 from app.v2.errors import (
     BadRequestError,
@@ -23,7 +23,7 @@ from tests.app.conftest import (
     sample_service as create_service,
     sample_service_whitelist,
     sample_api_key)
-from tests.app.db import create_reply_to_email
+from tests.app.db import create_reply_to_email, create_service_sms_sender
 
 
 @pytest.mark.parametrize('key_type', ['test', 'team', 'normal'])
@@ -311,9 +311,9 @@ def test_should_not_rate_limit_if_limiting_is_disabled(
 
 @pytest.mark.parametrize('key_type', ['test', 'normal'])
 def test_rejects_api_calls_with_international_numbers_if_service_does_not_allow_int_sms(
-    key_type,
-    notify_db,
-    notify_db_session,
+        key_type,
+        notify_db,
+        notify_db_session,
 ):
     service = create_service(notify_db, notify_db_session, permissions=[SMS_TYPE])
     with pytest.raises(BadRequestError) as e:
@@ -331,27 +331,73 @@ def test_allows_api_calls_with_international_numbers_if_service_does_allow_int_s
     assert result == '201212341234'
 
 
-def test_check_service_email_reply_to_id_where_reply_to_id_is_none():
-    assert check_service_email_reply_to_id(None, None) is None
+@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
+def test_check_service_email_reply_to_id_where_reply_to_id_is_none(notification_type):
+    assert check_service_email_reply_to_id(None, None, notification_type) is None
+
+
+def test_check_service_email_reply_to_where_email_reply_to_is_found(sample_service):
+    reply_to_address = create_reply_to_email(sample_service, "test@test.com")
+    assert check_service_email_reply_to_id(sample_service.id, reply_to_address.id, EMAIL_TYPE) is None
 
 
 def test_check_service_email_reply_to_id_where_service_id_is_not_found(sample_service, fake_uuid):
     reply_to_address = create_reply_to_email(sample_service, "test@test.com")
     with pytest.raises(BadRequestError) as e:
-        check_service_email_reply_to_id(fake_uuid, reply_to_address.id)
+        check_service_email_reply_to_id(fake_uuid, reply_to_address.id, EMAIL_TYPE)
     assert e.value.status_code == 400
-    assert e.value.message == 'email_reply_to_id {} does not exist in database for service id {}'\
+    assert e.value.message == 'email_reply_to_id {} does not exist in database for service id {}' \
         .format(reply_to_address.id, fake_uuid)
 
 
 def test_check_service_email_reply_to_id_where_reply_to_id_is_not_found(sample_service, fake_uuid):
     with pytest.raises(BadRequestError) as e:
-        check_service_email_reply_to_id(sample_service.id, fake_uuid)
+        check_service_email_reply_to_id(sample_service.id, fake_uuid, EMAIL_TYPE)
     assert e.value.status_code == 400
-    assert e.value.message == 'email_reply_to_id {} does not exist in database for service id {}'\
+    assert e.value.message == 'email_reply_to_id {} does not exist in database for service id {}' \
         .format(fake_uuid, sample_service.id)
 
 
-def test_check_service_email_reply_to_id_where_reply_to_id_is_found(sample_service):
-    reply_to_email = create_reply_to_email(sample_service, 'test@test.com')
-    assert check_service_email_reply_to_id(sample_service.id, reply_to_email.id) is None
+@pytest.mark.parametrize('notification_type', ['sms', 'letter'])
+def test_check_service_email_reply_to_id_when_channel_type_is_wrong(sample_service, notification_type):
+    reply_to_address = create_reply_to_email(sample_service, "test@test.com")
+    with pytest.raises(BadRequestError) as e:
+        check_service_email_reply_to_id(sample_service.id, reply_to_address.id, notification_type)
+    assert e.value.status_code == 400
+    assert e.value.message == 'email_reply_to_id is not a valid option for {} notification'.format(notification_type)
+
+
+@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
+def test_check_service_sms_sender_id_where_sms_sender_id_is_none(notification_type):
+    assert check_service_sms_sender_id(None, None, notification_type) is None
+
+
+def test_check_service_sms_sender_id_where_sms_sender_id_is_found(sample_service):
+    sms_sender = create_service_sms_sender(service=sample_service, sms_sender='123456')
+    assert check_service_sms_sender_id(sample_service.id, sms_sender.id, SMS_TYPE) is None
+
+
+def test_check_service_sms_sender_id_where_service_id_is_not_found(sample_service, fake_uuid):
+    sms_sender = create_service_sms_sender(service=sample_service, sms_sender='123456')
+    with pytest.raises(BadRequestError) as e:
+        check_service_sms_sender_id(fake_uuid, sms_sender.id, SMS_TYPE)
+    assert e.value.status_code == 400
+    assert e.value.message == 'sms_sender_id {} does not exist in database for service id {}' \
+        .format(sms_sender.id, fake_uuid)
+
+
+def test_check_service_sms_sender_id_where_sms_sender_is_not_found(sample_service, fake_uuid):
+    with pytest.raises(BadRequestError) as e:
+        check_service_sms_sender_id(sample_service.id, fake_uuid, SMS_TYPE)
+    assert e.value.status_code == 400
+    assert e.value.message == 'sms_sender_id {} does not exist in database for service id {}' \
+        .format(fake_uuid, sample_service.id)
+
+
+@pytest.mark.parametrize('notification_type', ['email', 'letter'])
+def test_check_service_sms_sender_id_when_channel_type_is_wrong(sample_service, notification_type):
+    sms_sender = create_service_sms_sender(service=sample_service, sms_sender='123456')
+    with pytest.raises(BadRequestError) as e:
+        check_service_sms_sender_id(sample_service.id, sms_sender.id, notification_type)
+    assert e.value.status_code == 400
+    assert e.value.message == 'sms_sender_id is not a valid option for {} notification'.format(notification_type)
diff --git a/tests/app/service/test_send_one_off_notification.py b/tests/app/service/test_send_one_off_notification.py
index 1fe9dc155..64650434c 100644
--- a/tests/app/service/test_send_one_off_notification.py
+++ b/tests/app/service/test_send_one_off_notification.py
@@ -3,6 +3,7 @@ from unittest.mock import Mock
 
 import pytest
 from notifications_utils.recipients import InvalidPhoneError
+from sqlalchemy.exc import SQLAlchemyError
 
 from app.v2.errors import BadRequestError, TooManyRequestsError
 from app.config import QueueNames
@@ -12,9 +13,11 @@ from app.models import (
     PRIORITY,
     SMS_TYPE,
     NotificationEmailReplyTo,
-    Notification)
+    Notification,
+    NotificationSmsSender
+)
 
-from tests.app.db import create_user, create_reply_to_email
+from tests.app.db import create_user, create_reply_to_email, create_service_sms_sender
 
 
 @pytest.fixture
@@ -206,7 +209,7 @@ def test_send_one_off_notification_should_add_email_reply_to_id_for_email(sample
 
 
 def test_send_one_off_notification_should_throw_exception_if_reply_to_id_doesnot_exist(
-        sample_email_template, celery_mock
+        sample_email_template
 ):
     data = {
         'to': 'ok@ok.com',
@@ -215,5 +218,39 @@ def test_send_one_off_notification_should_throw_exception_if_reply_to_id_doesnot
         'created_by': str(sample_email_template.service.created_by_id)
     }
 
-    with pytest.raises(expected_exception=BadRequestError):
+    with pytest.raises(expected_exception=SQLAlchemyError):
         send_one_off_notification(service_id=sample_email_template.service.id, post_data=data)
+
+
+def test_send_one_off_notification_should_add_sms_sender_mapping_for_sms(sample_template, celery_mock):
+    sms_sender = create_service_sms_sender(service=sample_template.service, sms_sender='123456')
+    data = {
+        'to': '07700 900 001',
+        'template_id': str(sample_template.id),
+        'sender_id': sms_sender.id,
+        'created_by': str(sample_template.service.created_by_id)
+    }
+
+    notification_id = send_one_off_notification(service_id=sample_template.service.id, post_data=data)
+    notification = Notification.query.get(notification_id['id'])
+    celery_mock.assert_called_once_with(
+        notification=notification,
+        research_mode=False,
+        queue=None
+    )
+    mapping_row = NotificationSmsSender.query.filter_by(notification_id=notification_id['id']).first()
+    assert mapping_row.service_sms_sender_id == sms_sender.id
+
+
+def test_send_one_off_notification_should_throw_exception_if_sms_sender_id_doesnot_exist(
+        sample_template
+):
+    data = {
+        'to': '07700 900 001',
+        'template_id': str(sample_template.id),
+        'sender_id': str(uuid.uuid4()),
+        'created_by': str(sample_template.service.created_by_id)
+    }
+
+    with pytest.raises(expected_exception=SQLAlchemyError):
+        send_one_off_notification(service_id=sample_template.service.id, post_data=data)
diff --git a/tests/app/test_model.py b/tests/app/test_model.py
index 8809a51bb..7062e7a30 100644
--- a/tests/app/test_model.py
+++ b/tests/app/test_model.py
@@ -28,7 +28,6 @@ from tests.app.db import (
     create_service,
     create_inbound_number,
     create_reply_to_email,
-    create_service_sms_sender,
     create_letter_contact
 )
 from tests.conftest import set_config
diff --git a/tests/app/user/test_rest.py b/tests/app/user/test_rest.py
index 28fcef05b..2a36d488e 100644
--- a/tests/app/user/test_rest.py
+++ b/tests/app/user/test_rest.py
@@ -546,3 +546,15 @@ def test_update_user_resets_failed_login_count_if_updating_password(client, samp
 
     assert resp.status_code == 200
     assert user.failed_login_count == 0
+
+
+def test_update_user_auth_type(admin_request, sample_user):
+    assert sample_user.auth_type == 'sms_auth'
+    resp = admin_request.post(
+        'user.update_user_attribute',
+        user_id=sample_user.id,
+        _data={'auth_type': 'email_auth'},
+    )
+
+    assert resp['data']['id'] == str(sample_user.id)
+    assert resp['data']['auth_type'] == 'email_auth'
diff --git a/tests/app/v2/notifications/test_post_notifications.py b/tests/app/v2/notifications/test_post_notifications.py
index 9d63c6bfb..d4055df3f 100644
--- a/tests/app/v2/notifications/test_post_notifications.py
+++ b/tests/app/v2/notifications/test_post_notifications.py
@@ -8,14 +8,15 @@ from app.models import (
     ScheduledNotification,
     SCHEDULE_NOTIFICATIONS,
     EMAIL_TYPE,
-    SMS_TYPE
-)
+    SMS_TYPE,
+    NotificationSmsSender)
 from flask import json, current_app
 
 from app.models import Notification
 from app.schema_validation import validate
 from app.v2.errors import RateLimitError
 from app.v2.notifications.notification_schemas import post_sms_response, post_email_response
+from app.v2.notifications.post_notifications import persist_sender_to_notification_mapping
 from tests import create_authorization_header
 from tests.app.conftest import (
     sample_template as create_sample_template,
@@ -27,8 +28,8 @@ from tests.app.conftest import (
 from tests.app.db import (
     create_service,
     create_template,
-    create_reply_to_email
-)
+    create_reply_to_email,
+    create_service_sms_sender, create_notification)
 
 
 @pytest.mark.parametrize("reference", [None, "reference_from_client"])
@@ -93,6 +94,34 @@ def test_post_sms_notification_uses_inbound_number_as_sender(client, notify_db_s
     mocked.assert_called_once_with([str(notification_id)], queue='send-sms-tasks')
 
 
+def test_post_sms_notification_returns_201_with_sms_sender_id(
+        client, sample_template_with_placeholders, mocker
+):
+
+    sms_sender = create_service_sms_sender(service=sample_template_with_placeholders.service, sms_sender='123456')
+    mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
+    data = {
+        'phone_number': '+447700900855',
+        'template_id': str(sample_template_with_placeholders.id),
+        'personalisation': {' Name': 'Jo'},
+        'sms_sender_id': str(sms_sender.id)
+    }
+    auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
+
+    response = client.post(
+        path='/v2/notifications/sms',
+        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))
+    assert validate(resp_json, post_sms_response) == resp_json
+    notification_to_sms_sender = NotificationSmsSender.query.all()
+    assert len(notification_to_sms_sender) == 1
+    assert str(notification_to_sms_sender[0].notification_id) == resp_json['id']
+    assert notification_to_sms_sender[0].service_sms_sender_id == sms_sender.id
+    mocked.assert_called_once_with([resp_json['id']], queue='send-sms-tasks')
+
+
 @pytest.mark.parametrize("notification_type, key_send_to, send_to",
                          [("sms", "phone_number", "+447700900855"),
                           ("email", "email_address", "sample@email.com")])
@@ -501,30 +530,40 @@ def test_post_notification_raises_bad_request_if_not_valid_notification_type(cli
     assert 'The requested URL was not found on the server.' in error_json['message']
 
 
-@pytest.mark.parametrize("reference", [None, "reference_from_client"])
-def test_post_sms_notification_with_invalid_reply_to_email_id(
+@pytest.mark.parametrize("notification_type",
+                         ['sms', 'email'])
+def test_post_notification_with_wrong_type_of_sender(
         client,
-        sample_template_with_placeholders,
-        reference,
+        sample_template,
+        sample_email_template,
+        notification_type,
         fake_uuid):
-    data = {
-        'phone_number': '+447700900855',
-        'template_id': str(sample_template_with_placeholders.id),
-        'personalisation': {' Name': 'Jo'},
-        'email_reply_to_id': fake_uuid
-    }
-    if reference:
-        data.update({"reference": reference})
-    auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
+    if notification_type == EMAIL_TYPE:
+        template = sample_email_template
+        form_label = 'sms_sender_id'
+        data = {
+            'email_address': 'test@test.com',
+            'template_id': str(sample_email_template.id),
+            form_label: fake_uuid
+        }
+    elif notification_type == SMS_TYPE:
+        template = sample_template
+        form_label = 'email_reply_to_id'
+        data = {
+            'phone_number': '+447700900855',
+            'template_id': str(template.id),
+            form_label: fake_uuid
+        }
+    auth_header = create_authorization_header(service_id=template.service_id)
 
     response = client.post(
-        path='/v2/notifications/sms',
+        path='/v2/notifications/{}'.format(notification_type),
         data=json.dumps(data),
         headers=[('Content-Type', 'application/json'), auth_header])
     assert response.status_code == 400
     resp_json = json.loads(response.get_data(as_text=True))
-    assert 'email_reply_to_id {} does not exist in database for service id {}'.\
-        format(fake_uuid, sample_template_with_placeholders.service_id) in resp_json['errors'][0]['message']
+    assert '{} is not a valid option for {} notification'.\
+        format(form_label, notification_type) in resp_json['errors'][0]['message']
     assert 'BadRequestError' in resp_json['errors'][0]['error']
 
 
@@ -592,3 +631,37 @@ def test_post_email_notification_with_valid_reply_to_id_returns_201(client, samp
 
     assert email_reply_to.notification_id == notification.id
     assert email_reply_to.service_email_reply_to_id == reply_to_email.id
+
+
+def test_persist_sender_to_notification_mapping_for_email(notify_db_session):
+    service = create_service()
+    template = create_template(service=service, template_type=EMAIL_TYPE)
+    sender = create_reply_to_email(service=service, email_address='reply@test.com', is_default=False)
+    form = {
+        "email_address": "recipient@test.com",
+        "template_id": str(template.id),
+        'email_reply_to_id': str(sender.id)
+    }
+    notification = create_notification(template=template)
+    persist_sender_to_notification_mapping(form=form, notification=notification)
+    notification_to_email_reply_to = NotificationEmailReplyTo.query.all()
+    assert len(notification_to_email_reply_to) == 1
+    assert notification_to_email_reply_to[0].notification_id == notification.id
+    assert notification_to_email_reply_to[0].service_email_reply_to_id == sender.id
+
+
+def test_persist_sender_to_notification_mapping_for_sms(notify_db_session):
+    service = create_service()
+    template = create_template(service=service, template_type=SMS_TYPE)
+    sender = create_service_sms_sender(service=service, sms_sender='12345', is_default=False)
+    form = {
+        'phone_number': '+447700900855',
+        'template_id': str(template.id),
+        'sms_sender_id': str(sender.id)
+    }
+    notification = create_notification(template=template)
+    persist_sender_to_notification_mapping(form=form, notification=notification)
+    notification_to_sms_sender = NotificationSmsSender.query.all()
+    assert len(notification_to_sms_sender) == 1
+    assert notification_to_sms_sender[0].notification_id == notification.id
+    assert notification_to_sms_sender[0].service_sms_sender_id == sender.id