darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-01-22 10:31:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1945 from alphagov/flexible-data-retention

Browse Source 
Flexible data retention
This commit is contained in:
Rebecca Law
2018-07-17 16:03:18 +01:00
committed by GitHub
parent 081c7c7083 ab695d24d5
commit 9c99e45008
7 changed files with 436 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
tests/app/dao/test_service_data_retention_dao.py Normal file
View File

@@ -0,0 +1,133 @@
import uuid
from datetime import datetime
import pytest
from sqlalchemy.exc import IntegrityError
from app.dao.service_data_retention_dao import (
fetch_service_data_retention,
insert_service_data_retention,
update_service_data_retention,
fetch_service_data_retention_by_id
)
from app.models import ServiceDataRetention
from tests.app.db import create_service
def test_fetch_service_data_retention(sample_service):
email_data_retention = insert_service_data_retention(sample_service.id, 'email', 3)
letter_data_retention = insert_service_data_retention(sample_service.id, 'letter', 30)
sms_data_retention = insert_service_data_retention(sample_service.id, 'sms', 5)
list_of_data_retention = fetch_service_data_retention(sample_service.id)
assert len(list_of_data_retention) == 3
assert list_of_data_retention[0] == email_data_retention
assert list_of_data_retention[1] == sms_data_retention
assert list_of_data_retention[2] == letter_data_retention
def test_fetch_service_data_retention_only_returns_row_for_service(sample_service):
another_service = create_service(service_name="Another service")
email_data_retention = insert_service_data_retention(sample_service.id, 'email', 3)
letter_data_retention = insert_service_data_retention(sample_service.id, 'letter', 30)
insert_service_data_retention(another_service.id, 'sms', 5)
list_of_data_retention = fetch_service_data_retention(sample_service.id)
assert len(list_of_data_retention) == 2
assert list_of_data_retention[0] == email_data_retention
assert list_of_data_retention[1] == letter_data_retention
def test_fetch_service_data_retention_returns_empty_list_when_no_rows_for_service(sample_service):
empty_list = fetch_service_data_retention(sample_service.id)
assert not empty_list
def test_fetch_service_data_retention_by_id(sample_service):
email_data_retention = insert_service_data_retention(sample_service.id, 'email', 3)
insert_service_data_retention(sample_service.id, 'sms', 13)
result = fetch_service_data_retention_by_id(sample_service.id, email_data_retention.id)
assert result == email_data_retention
def test_fetch_service_data_retention_by_id_returns_none_if_not_found(sample_service):
result = fetch_service_data_retention_by_id(sample_service.id, uuid.uuid4())
assert not result
def test_fetch_service_data_retention_by_id_returns_none_if_id_not_for_service(sample_service):
another_service = create_service(service_name="Another service")
email_data_retention = insert_service_data_retention(sample_service.id, 'email', 3)
result = fetch_service_data_retention_by_id(another_service.id, email_data_retention.id)
assert not result
def test_insert_service_data_retention(sample_service):
insert_service_data_retention(
service_id=sample_service.id,
notification_type='email',
days_of_retention=3
)
results = ServiceDataRetention.query.all()
assert len(results) == 1
assert results[0].service_id == sample_service.id
assert results[0].notification_type == 'email'
assert results[0].days_of_retention == 3
assert results[0].created_at.date() == datetime.utcnow().date()
def test_insert_service_data_retention_throws_unique_constraint(sample_service):
insert_service_data_retention(service_id=sample_service.id,
notification_type='email',
days_of_retention=3
)
with pytest.raises(expected_exception=IntegrityError):
insert_service_data_retention(service_id=sample_service.id,
notification_type='email',
days_of_retention=5
)
def test_update_service_data_retention(sample_service):
data_retention = insert_service_data_retention(service_id=sample_service.id,
notification_type='sms',
days_of_retention=3
)
updated_count = update_service_data_retention(service_data_retention_id=data_retention.id,
service_id=sample_service.id,
days_of_retention=5
)
assert updated_count == 1
results = ServiceDataRetention.query.all()
assert len(results) == 1
assert results[0].id == data_retention.id
assert results[0].service_id == sample_service.id
assert results[0].notification_type == 'sms'
assert results[0].days_of_retention == 5
assert results[0].created_at.date() == datetime.utcnow().date()
assert results[0].updated_at.date() == datetime.utcnow().date()
def test_update_service_data_retention_does_not_update_if_row_does_not_exist(sample_service):
updated_count = update_service_data_retention(
service_data_retention_id=uuid.uuid4(),
service_id=sample_service.id,
days_of_retention=5
)
assert updated_count == 0
assert len(ServiceDataRetention.query.all()) == 0
def test_update_service_data_retention_does_not_update_row_if_data_retention_is_for_different_service(
sample_service
):
data_retention = insert_service_data_retention(service_id=sample_service.id,
notification_type='email',
days_of_retention=3
)
updated_count = update_service_data_retention(service_data_retention_id=data_retention.id,
service_id=uuid.uuid4(),
days_of_retention=5)
assert updated_count == 0

16
tests/app/db.py
View File

@@ -3,6 +3,7 @@ import uuid
from app import db
from app.dao.jobs_dao import dao_create_job
from app.dao.service_data_retention_dao import insert_service_data_retention
from app.dao.service_inbound_api_dao import save_service_inbound_api
from app.dao.service_callback_api_dao import save_service_callback_api
from app.dao.service_sms_sender_dao import update_existing_sms_sender_with_inbound_number, dao_update_service_sms_sender
@@ -35,7 +36,7 @@ from app.models import (
InvitedOrganisationUser,
FactBilling,
FactNotificationStatus,
Complaint
Complaint,
)
from app.dao.users_dao import save_model_user
from app.dao.notifications_dao import (
@@ -663,3 +664,16 @@ def ses_notification_callback():
'dd426d95ee9390147a5624348ee.pem",' \
'\n "UnsubscribeURL" : "https://sns.eu-west-1.amazonaws.com/?Action=Unsubscribe&S' \
'subscriptionArn=arn:aws:sns:eu-west-1:302763885840:preview-emails:d6aad3ef-83d6-4cf3-a470-54e2e75916da"\n}'
def create_service_data_retention(
service_id,
notification_type='sms',
days_of_retention=3
):
data_retention = insert_service_data_retention(
service_id=service_id,
notification_type=notification_type,
days_of_retention=days_of_retention
)
return data_retention

152
tests/app/service/test_service_data_retention_rest.py Normal file
View File

@@ -0,0 +1,152 @@
import json
import uuid
from app.models import ServiceDataRetention
from tests import create_authorization_header
from tests.app.db import create_service_data_retention
def test_get_service_data_retention(client, sample_service):
sms_data_retention = create_service_data_retention(service_id=sample_service.id)
email_data_retention = create_service_data_retention(service_id=sample_service.id, notification_type='email',
days_of_retention=10)
letter_data_retention = create_service_data_retention(service_id=sample_service.id, notification_type='letter',
days_of_retention=30)
response = client.get(
'/service/{}/data-retention'.format(str(sample_service.id)),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
)
assert response.status_code == 200
json_response = json.loads(response.get_data(as_text=True))
assert len(json_response) == 3
assert json_response[0] == email_data_retention.serialize()
assert json_response[1] == sms_data_retention.serialize()
assert json_response[2] == letter_data_retention.serialize()
def test_get_service_data_retention_returns_empty_list(client, sample_service):
response = client.get(
'/service/{}/data-retention'.format(str(sample_service.id)),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
)
assert response.status_code == 200
assert len(json.loads(response.get_data(as_text=True))) == 0
def test_get_service_data_retention_by_id(client, sample_service):
sms_data_retention = create_service_data_retention(service_id=sample_service.id)
create_service_data_retention(service_id=sample_service.id, notification_type='email',
days_of_retention=10)
create_service_data_retention(service_id=sample_service.id, notification_type='letter',
days_of_retention=30)
response = client.get(
'/service/{}/data-retention/{}'.format(str(sample_service.id), sms_data_retention.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == sms_data_retention.serialize()
def test_get_service_data_retention_by_id_returns_none_when_no_data_retention_exists(client, sample_service):
response = client.get(
'/service/{}/data-retention/{}'.format(str(sample_service.id), uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == {}
def test_create_service_data_retention(client, sample_service):
data = {
"notification_type": 'sms',
"days_of_retention": 3
}
response = client.post(
'/service/{}/data-retention'.format(str(sample_service.id)),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
data=json.dumps(data)
)
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))['result']
results = ServiceDataRetention.query.all()
assert len(results) == 1
data_retention = results[0]
assert json_resp == data_retention.serialize()
def test_create_service_data_retention_returns_400_when_notification_type_is_invalid(client):
data = {
"notification_type": 'unknown',
"days_of_retention": 3
}
response = client.post(
'/service/{}/data-retention'.format(str(uuid.uuid4())),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
data=json.dumps(data)
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['errors'][0]['error'] == 'ValidationError'
assert json_resp['errors'][0]['message'] == 'notification_type unknown is not one of [sms, letter, email]'
def test_create_service_data_retention_returns_400_when_data_retention_for_notification_type_already_exists(
client, sample_service
):
create_service_data_retention(service_id=sample_service.id)
data = {
"notification_type": "sms",
"days_of_retention": 3
}
response = client.post(
'/service/{}/data-retention'.format(str(uuid.uuid4())),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
data=json.dumps(data)
)
assert response.status_code == 400
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'Service already has data retention for sms notification type'
def test_modify_service_data_retention(client, sample_service):
data_retention = create_service_data_retention(service_id=sample_service.id)
data = {
"days_of_retention": 3
}
response = client.post(
'/service/{}/data-retention/{}'.format(sample_service.id, data_retention.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
data=json.dumps(data)
)
assert response.status_code == 204
assert response.get_data(as_text=True) == ''
def test_modify_service_data_retention_returns_400_when_data_retention_does_not_exist(client, sample_service):
data = {
"days_of_retention": 3
}
response = client.post(
'/service/{}/data-retention/{}'.format(sample_service.id, uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
data=json.dumps(data)
)
assert response.status_code == 404
def test_modify_service_data_retention_returns_400_when_data_is_invalid(client):
data = {
"bad_key": 3
}
response = client.post(
'/service/{}/data-retention/{}'.format(uuid.uuid4(), uuid.uuid4()),
headers=[('Content-Type', 'application/json'), create_authorization_header()],
data=json.dumps(data)
)
assert response.status_code == 400