Adding rest endpoints for letter-branding

app/__init__.py

@@ -128,6 +128,7 @@ def register_blueprint(application):
     from app.complaint.complaint_rest import complaint_blueprint
     from app.platform_stats.rest import platform_stats_blueprint
     from app.template_folder.rest import template_folder_blueprint
+    from app.letter_branding.letter_branding_rest import letter_branding_blueprint
 
     service_blueprint.before_request(requires_admin_auth)
     application.register_blueprint(service_blueprint, url_prefix='/service')
@@ -213,6 +214,9 @@ def register_blueprint(application):
     template_folder_blueprint.before_request(requires_admin_auth)
     application.register_blueprint(template_folder_blueprint)
 
+    letter_branding_blueprint.before_request(requires_admin_auth())
+    application.register_blueprint(letter_branding_blueprint)
+
 
 def register_v2_blueprints(application):
     from app.v2.inbound_sms.get_inbound_sms import v2_inbound_sms_blueprint as get_inbound_sms

app/dao/letter_branding_dao.py

@@ -1,7 +1,9 @@
+from app import db
+from app.dao.dao_utils import transactional
 from app.models import LetterBranding
 
 
-def get_letter_branding_or_platform_default(domain=None):
+def dao_get_letter_branding_or_platform_default(domain=None):
     letter_branding = None
     if domain:
         letter_branding = LetterBranding.query.filter(
@@ -14,5 +16,19 @@ def get_letter_branding_or_platform_default(domain=None):
     return letter_branding
 
 
-def get_all_letter_branding():
+def dao_get_all_letter_branding():
     return LetterBranding.query.order_by(LetterBranding.name).all()
+
+
+@transactional
+def dao_create_letter_branding(letter_branding):
+    db.session.add(letter_branding)
+
+
+@transactional
+def dao_update_letter_branding(letter_branding_id, **kwargs):
+    letter_branding = LetterBranding.query.get(letter_branding_id)
+    for key, value in kwargs.items():
+        setattr(letter_branding, key, value or None)
+    db.session.add(letter_branding)
+    return letter_branding

app/letter_branding/letter_branding_rest.py

@@ -1,6 +1,76 @@
-from flask import Blueprint
+from celery import current_app
+from sqlalchemy.exc import IntegrityError
 
+from flask import Blueprint, jsonify, request
+
+from app.dao.letter_branding_dao import (
+    dao_get_all_letter_branding, dao_create_letter_branding,
+    dao_update_letter_branding
+)
 from app.errors import register_errors
+from app.letter_branding.letter_branding_schema import post_letter_branding_schema
+from app.models import LetterBranding
+from app.schema_validation import validate
 
-email_branding_blueprint = Blueprint('letter_branding', __name__, url_prefix='letter-branding')
+letter_branding_blueprint = Blueprint('letter_branding', __name__, url_prefix='letter-branding')
-register_errors(email_branding_blueprint)
+register_errors(letter_branding_blueprint)
+
+
+@letter_branding_blueprint.errorhandler(IntegrityError)
+def handle_integrity_error(exc):
+    """
+    Handle integrity errors caused by the unique constraint
+    """
+    if 'domain' in str(exc):
+        return jsonify(
+            result='error',
+            message={'name': ["Duplicate domain '{}'".format(
+                exc.params.get('domain')
+            )]}
+        ), 400
+    if 'name' in str(exc):
+        return jsonify(
+            result='error',
+            message={'name': ["Duplicate name '{}'".format(
+                exc.params.get('name')
+            )]}
+        ), 400
+    if 'filename' in str(exc):
+        return jsonify(
+            result='error',
+            message={'name': ["Duplicate filename '{}'".format(
+                exc.params.get('fileaname')
+            )]}
+        ), 400
+    current_app.logger.exception(exc)
+    return jsonify(result='error', message="Internal server error"), 500
+
+
+@letter_branding_blueprint.route('', methods=['GET'])
+def get_all_letter_brands():
+    letter_brands = dao_get_all_letter_branding()
+
+    return jsonify([lb.serialize() for lb in letter_brands])
+
+
+@letter_branding_blueprint.route('', methods=['POST'])
+def create_letter_brand():
+    data = request.get_json()
+
+    validate(data, post_letter_branding_schema)
+
+    letter_branding = LetterBranding(**data)
+    dao_create_letter_branding(letter_branding)
+
+    return jsonify(letter_branding.serialize()), 201
+
+
+@letter_branding_blueprint.route('/<uuid: letter_branding_id>', methods=['POST'])
+def create_letter_branding(letter_branding_id):
+    data = request.get_json()
+
+    validate(data, post_letter_branding_schema)
+
+    letter_branding = dao_update_letter_branding(letter_branding_id, )
+
+    return jsonify(letter_branding.serialize()), 201

app/letter_branding/letter_branding_schema.py

@@ -0,0 +1,11 @@
+post_letter_branding_schema = {
+    "$schema": "http://json-schema.org/draft-04/schema#",
+    "description": "POST schema for creating or updating a letter brand",
+    "type": "object",
+    "properties": {
+        "name": {"type": ["string", "null"]},
+        "filename": {"type": ["string", "null"]},
+        "domain": {"type": ["string", "null"]},
+    },
+    "required": ("name", "filename", "domain")
+}

app/models.py

@@ -257,6 +257,15 @@ class LetterBranding(db.Model):
     domain = db.Column(db.Text, unique=True, nullable=True)
     platform_default = db.Column(db.Boolean, nullable=False, default=False)
 
+    def serialize(self):
+        return {
+            "id": self.id,
+            "name": self.name,
+            "filename": self.filename,
+            "domain": self.domain,
+            "platform_default": self.platform_default
+        }
+
 
 service_letter_branding = db.Table(
     'service_letter_branding',

app/service/rest.py

@@ -27,7 +27,7 @@ from app.dao.fact_notification_status_dao import (
     fetch_stats_for_all_services_by_date_range, fetch_monthly_template_usage_for_service
 )
 from app.dao.inbound_numbers_dao import dao_allocate_number_for_service
-from app.dao.letter_branding_dao import get_letter_branding_or_platform_default
+from app.dao.letter_branding_dao import dao_get_letter_branding_or_platform_default
 from app.dao.organisation_dao import dao_get_organisation_by_service_id
 from app.dao.service_data_retention_dao import (
     fetch_service_data_retention,
@@ -192,7 +192,7 @@ def create_service():
     # unpack valid json into service object
     valid_service = Service.from_json(data)
 
-    letter_branding = get_letter_branding_or_platform_default(domain)
+    letter_branding = dao_get_letter_branding_or_platform_default(domain)
     dao_create_service(valid_service, user, letter_branding=letter_branding)
 
     return jsonify(data=service_schema.dump(valid_service).data), 201

tests/app/dao/test_letter_branding_dao.py

@@ -1,23 +1,72 @@
-from app.dao.letter_branding_dao import get_letter_branding_or_platform_default
+from app.dao.letter_branding_dao import (
+    dao_get_letter_branding_or_platform_default,
+    dao_get_all_letter_branding,
+    dao_create_letter_branding,
+    dao_update_letter_branding
+)
+from app.models import LetterBranding
 from tests.app.db import create_letter_branding
 
 
-def test_get_letter_branding_or_platform_default_returns_platform_default_if_domain_is_none(notify_db_session):
+def test_dao_get_letter_branding_or_platform_default_returns_platform_default_if_domain_is_none(notify_db_session):
     create_letter_branding()
-    result = get_letter_branding_or_platform_default(domain=None)
+    result = dao_get_letter_branding_or_platform_default(domain=None)
     assert result.filename == 'hm-government'
 
 
-def test_get_letter_branding_or_platform_default_if_domain_is_not_associated_with_a_brand(notify_db_session):
+def test_dao_get_letter_branding_or_platform_default_if_domain_is_not_associated_with_a_brand(notify_db_session):
     create_letter_branding()
-    result = get_letter_branding_or_platform_default(domain="foo.bar")
+    result = dao_get_letter_branding_or_platform_default(domain="foo.bar")
     assert result.filename == 'hm-government'
 
 
-def test_get_letter_branding_or_platform_default_returns_correct_brand_for_domain(notify_db_session):
+def test_dao_get_letter_branding_or_platform_default_returns_correct_brand_for_domain(notify_db_session):
     create_letter_branding()
     test_domain_branding = create_letter_branding(
         name='test domain', filename='test-domain', domain='test.domain', platform_default=False
     )
-    result = get_letter_branding_or_platform_default(domain='test.domain')
+    result = dao_get_letter_branding_or_platform_default(domain='test.domain')
     result == test_domain_branding
+
+
+def test_dao_get_all_letter_branding(notify_db_session):
+    platform_default = create_letter_branding()
+    test_domain = create_letter_branding(
+        name='test domain', filename='test-domain', domain='test.domain', platform_default=False
+    )
+
+    results = dao_get_all_letter_branding()
+
+    assert platform_default in results
+    assert test_domain in results
+    assert len(results) == 2
+
+
+def test_dao_get_all_letter_branding_returns_empty_list_if_no_brands_exist(notify_db):
+    assert dao_get_all_letter_branding() == []
+
+
+def test_dao_create_letter_branding(notify_db_session):
+    data = {
+        'name': 'test-logo',
+        'domain': 'test.co.uk',
+        'filename': 'test-logo'
+    }
+    assert LetterBranding.query.count() == 0
+    dao_create_letter_branding(LetterBranding(**data))
+
+    assert LetterBranding.query.count() == 1
+
+    new_letter_branding = LetterBranding.query.first()
+    assert new_letter_branding.name == data['name']
+    assert new_letter_branding.domain == data['domain']
+    assert new_letter_branding.filename == data['name']
+    assert not new_letter_branding.platform_default
+
+
+def test_dao_update_letter_branding(notify_db_session):
+    create_letter_branding(name='original')
+    letter_branding = LetterBranding.query.first()
+    assert letter_branding.name == 'original'
+    dao_update_letter_branding(letter_branding.id, name='new name')
+    assert LetterBranding.query.first().name == 'new name'

tests/app/letter_branding/test_letter_branding_rest.py

@@ -0,0 +1,14 @@
+from tests import create_authorization_header
+from tests.app.db import create_letter_branding
+
+
+def test_get_letter_brandings(client, notify_db_session):
+    platform_default = create_letter_branding()
+    test_domain_branding = create_letter_branding(
+        name='test domain', filename='test-domain', domain='test.domain', platform_default=False
+    )
+    response = client.get('/letter-branding', headers=[create_authorization_header()])
+    assert response.status_code == 200
+    json_response = response.get_data(as_text=True)
+    assert platform_default.serialize() in json_response
+    assert test_domain_branding.serialize() in json_response