Merge pull request #956 from alphagov/add-service-permissions

Add service permissions DAO and refactor user service permission mock

app/dao/service_permissions_dao.py

@@ -0,0 +1,21 @@
+from app import db
+from app.dao.dao_utils import transactional
+from app.models import ServicePermission, SERVICE_PERMISSION_TYPES
+
+
+def dao_fetch_service_permissions(service_id):
+    return ServicePermission.query.filter(
+        ServicePermission.service_id == service_id).all()
+
+
+@transactional
+def dao_create_service_permission(service_id, permission):
+    service_permission = ServicePermission(service_id=service_id, permission=permission)
+
+    db.session.add(service_permission)
+
+
+def dao_remove_service_permission(service_id, permission):
+    return ServicePermission.query.filter(
+        ServicePermission.service_id == service_id,
+        ServicePermission.permission == permission).delete()

migrations/README

@@ -1 +1,7 @@
 Generic single-database configuration.
+
+python application.py db migration to generate migration script.
+
+python application.py db upgrade to upgrade db with script.
+
+python application.py db downgrade to rollback db changes.

migrations/versions/0083_add_perm_types_and_svc_perm.py

@@ -15,10 +15,10 @@ import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
 def upgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
+    ### commands auto generated by Alembic - please adjust! ###
-    service_permission_types = op.create_table('service_permission_types',
+    service_permission_types=op.create_table('service_permission_types',
-                    sa.Column('name', sa.String(length=255), nullable=False),
+                                             sa.Column('name', sa.String(length=255), nullable=False),
-                    sa.PrimaryKeyConstraint('name'))
+                                             sa.PrimaryKeyConstraint('name'))
 
     op.bulk_insert(service_permission_types,
                    [

tests/app/conftest.py

@@ -666,11 +666,9 @@ def sample_permission(notify_db,
 
 
 @pytest.fixture(scope='function')
-def sample_service_permission(notify_db,
+def sample_user_service_permission(
-                              notify_db_session,
+    notify_db, notify_db_session, service=None, user=None, permission="manage_settings"
-                              service=None,
+):
-                              user=None,
-                              permission="manage_settings"):
     if user is None:
         user = create_user()
     if service is None:

tests/app/dao/test_service_permissions_dao.py

@@ -0,0 +1,36 @@
+from app.dao.service_permissions_dao import dao_fetch_service_permissions, dao_remove_service_permission
+from app.models import EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INCOMING_SMS_TYPE
+
+from tests.app.db import create_service_permission
+
+
+def test_create_service_permission(sample_service):
+    service_permissions = create_service_permission(service_id=sample_service.id, permission=SMS_TYPE)
+
+    assert len(service_permissions) == 1
+    assert service_permissions[0].service_id == sample_service.id
+    assert service_permissions[0].permission == SMS_TYPE
+
+
+def test_fetch_service_permissions_gets_service_permissions(sample_service):
+    create_service_permission(service_id=sample_service.id, permission=LETTER_TYPE)
+    create_service_permission(service_id=sample_service.id, permission=INTERNATIONAL_SMS_TYPE)
+    create_service_permission(service_id=sample_service.id, permission=SMS_TYPE)
+
+    service_permissions = dao_fetch_service_permissions(sample_service.id)
+
+    assert len(service_permissions) == 3
+    assert all(sp.service_id == sample_service.id for sp in service_permissions)
+    assert all(sp.permission in [LETTER_TYPE, INTERNATIONAL_SMS_TYPE, SMS_TYPE] for sp in service_permissions)
+
+
+def test_remove_service_permission(sample_service):
+    create_service_permission(service_id=sample_service.id, permission=EMAIL_TYPE)
+    create_service_permission(service_id=sample_service.id, permission=INCOMING_SMS_TYPE)
+
+    dao_remove_service_permission(sample_service.id, EMAIL_TYPE)
+
+    permissions = dao_fetch_service_permissions(sample_service.id)
+    assert len(permissions) == 1
+    assert permissions[0].permission == INCOMING_SMS_TYPE
+    assert permissions[0].service_id == sample_service.id

tests/app/db.py

@@ -2,11 +2,13 @@ from datetime import datetime
 import uuid
 
 from app.dao.jobs_dao import dao_create_job
-from app.models import Service, User, Template, Notification, SMS_TYPE, KEY_TYPE_NORMAL, Job
+from app.models import (Service, User, Template, Notification, EMAIL_TYPE, LETTER_TYPE,
+                        SMS_TYPE, KEY_TYPE_NORMAL, Job, ServicePermission)
 from app.dao.users_dao import save_model_user
 from app.dao.notifications_dao import dao_create_notification
 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_create_service_permission
 
 
 def create_user(mobile_number="+447700900986", email="notify@digital.cabinet-office.gov.uk", state='active'):
@@ -142,3 +144,12 @@ def create_job(template,
     job = Job(**data)
     dao_create_job(job)
     return job
+
+
+def create_service_permission(service_id, permission=EMAIL_TYPE):
+    dao_create_service_permission(
+        service_id if service_id else create_service().id, permission)
+
+    service_permissions = ServicePermission.query.all()
+
+    return service_permissions

tests/app/service/test_rest.py

@@ -15,7 +15,7 @@ from tests import create_authorization_header
 from tests.app.db import create_template
 from tests.app.conftest import (
     sample_service as create_service,
-    sample_service_permission as create_service_permission,
+    sample_user_service_permission as create_user_service_permission,
     sample_notification as create_sample_notification,
     sample_notification_history as create_notification_history,
     sample_notification_with_job
@@ -941,51 +941,51 @@ def test_add_unknown_user_to_service_returns404(notify_api, notify_db, notify_db
             assert result['message'] == expected_message
 
 
-def test_remove_user_from_service(notify_api, notify_db, notify_db_session, sample_service_permission):
+def test_remove_user_from_service(
-    with notify_api.test_request_context():
+    notify_db, notify_db_session, client, sample_user_service_permission
-        with notify_api.test_client() as client:
+):
-            second_user = create_user(email="new@digital.cabinet-office.gov.uk")
+    second_user = create_user(email="new@digital.cabinet-office.gov.uk")
-            # Simulates successfully adding a user to the service
+    # Simulates successfully adding a user to the service
-            second_permission = create_service_permission(
+    second_permission = create_user_service_permission(
-                notify_db,
+        notify_db,
-                notify_db_session,
+        notify_db_session,
-                user=second_user)
+        user=second_user)
-            endpoint = url_for(
+    endpoint = url_for(
-                'service.remove_user_from_service',
+        'service.remove_user_from_service',
-                service_id=str(second_permission.service.id),
+        service_id=str(second_permission.service.id),
-                user_id=str(second_permission.user.id))
+        user_id=str(second_permission.user.id))
-            auth_header = create_authorization_header()
+    auth_header = create_authorization_header()
-            resp = client.delete(
+    resp = client.delete(
-                endpoint,
+        endpoint,
-                headers=[('Content-Type', 'application/json'), auth_header])
+        headers=[('Content-Type', 'application/json'), auth_header])
-            assert resp.status_code == 204
+    assert resp.status_code == 204
 
 
-def test_remove_user_from_service(notify_api, notify_db, notify_db_session, sample_service_permission):
+def test_remove_non_existant_user_from_service(
-    with notify_api.test_request_context():
+    client, sample_user_service_permission
-        with notify_api.test_client() as client:
+):
-            second_user = create_user(email="new@digital.cabinet-office.gov.uk")
+    second_user = create_user(email="new@digital.cabinet-office.gov.uk")
-            endpoint = url_for(
+    endpoint = url_for(
-                'service.remove_user_from_service',
+        'service.remove_user_from_service',
-                service_id=str(sample_service_permission.service.id),
+        service_id=str(sample_user_service_permission.service.id),
-                user_id=str(second_user.id))
+        user_id=str(second_user.id))
-            auth_header = create_authorization_header()
+    auth_header = create_authorization_header()
-            resp = client.delete(
+    resp = client.delete(
-                endpoint,
+        endpoint,
-                headers=[('Content-Type', 'application/json'), auth_header])
+        headers=[('Content-Type', 'application/json'), auth_header])
-            assert resp.status_code == 404
+    assert resp.status_code == 404
 
 
 def test_cannot_remove_only_user_from_service(notify_api,
                                               notify_db,
                                               notify_db_session,
-                                              sample_service_permission):
+                                              sample_user_service_permission):
     with notify_api.test_request_context():
         with notify_api.test_client() as client:
             endpoint = url_for(
                 'service.remove_user_from_service',
-                service_id=str(sample_service_permission.service.id),
+                service_id=str(sample_user_service_permission.service.id),
-                user_id=str(sample_service_permission.user.id))
+                user_id=str(sample_user_service_permission.user.id))
             auth_header = create_authorization_header()
             resp = client.delete(
                 endpoint,

tests/app/user/test_rest.py

@@ -290,13 +290,13 @@ def test_get_user_by_email_bad_url_returns_404(client, sample_user):
     assert json_resp['message'] == 'Invalid request. Email query string param required'
 
 
-def test_get_user_with_permissions(client, sample_service_permission):
+def test_get_user_with_permissions(client, sample_user_service_permission):
     header = create_authorization_header()
-    response = client.get(url_for('user.get_user', user_id=str(sample_service_permission.user.id)),
+    response = client.get(url_for('user.get_user', user_id=str(sample_user_service_permission.user.id)),
                           headers=[header])
     assert response.status_code == 200
     permissions = json.loads(response.get_data(as_text=True))['data']['permissions']
-    assert sample_service_permission.permission in permissions[str(sample_service_permission.service.id)]
+    assert sample_user_service_permission.permission in permissions[str(sample_user_service_permission.service.id)]
 
 
 def test_set_user_permissions(client, sample_user, sample_service):

tests/conftest.py

@@ -77,7 +77,8 @@ def notify_db_session(notify_db):
                             "provider_details_history",
                             "template_process_type",
                             "dvla_organisation",
-                            "notification_status_types"]:
+                            "notification_status_types",
+                            "service_permission_types"]:
             notify_db.engine.execute(tbl.delete())
     notify_db.session.commit()