mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-08-11 09:28:27 -04:00
Fixed download bug for services with api notifications. Re-organised the mock job and job_json methods.
This commit is contained in:
@@ -35,6 +35,10 @@ def generate_uuid():
|
||||
return uuid.uuid4()
|
||||
|
||||
|
||||
def created_by_json(id_, name='', email_address=''):
|
||||
return {'id': id_, 'name': name, 'email_address': email_address}
|
||||
|
||||
|
||||
def service_json(id_, name, users, message_limit=1000, active=False, restricted=True, email_from=None, reply_to_email_address=None): # noqa
|
||||
return {
|
||||
'id': id_,
|
||||
@@ -75,11 +79,11 @@ def template_version_json(service_id,
|
||||
created_at=None,
|
||||
**kwargs):
|
||||
template = template_json(service_id, id_, **kwargs)
|
||||
template['created_by'] = {
|
||||
'id': created_by.id,
|
||||
'name': created_by.name,
|
||||
'email_address': created_by.email_address
|
||||
}
|
||||
template['created_by'] = created_by_json(
|
||||
created_by.id,
|
||||
created_by.name,
|
||||
created_by.email_address
|
||||
)
|
||||
if created_at is None:
|
||||
created_at = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
|
||||
template['created_at'] = created_at
|
||||
@@ -122,39 +126,41 @@ def create_test_api_user(state, permissions={}):
|
||||
return user
|
||||
|
||||
|
||||
def job_json():
|
||||
job_id = str(generate_uuid())
|
||||
created_at = str(datetime.utcnow().time())
|
||||
def job_json(service_id,
|
||||
created_by,
|
||||
job_id=None,
|
||||
template_id=None,
|
||||
template_version=1,
|
||||
created_at=None,
|
||||
bucket_name='',
|
||||
original_file_name="thisisatest.csv",
|
||||
notification_count=1,
|
||||
notifications_sent=1,
|
||||
status=''):
|
||||
if job_id is None:
|
||||
job_id = str(generate_uuid())
|
||||
if template_id is None:
|
||||
template_id = str(generate_uuid())
|
||||
if created_at is None:
|
||||
created_at = str(datetime.utcnow().time())
|
||||
data = {
|
||||
'id': job_id,
|
||||
'service': 1,
|
||||
'template': 1,
|
||||
'template_version': 1,
|
||||
'original_file_name': 'thisisatest.csv',
|
||||
'service': service_id,
|
||||
'template': template_id,
|
||||
'template_version': template_version,
|
||||
'original_file_name': original_file_name,
|
||||
'created_at': created_at,
|
||||
'notification_count': 1,
|
||||
'notifications_sent': 1,
|
||||
'status': ''
|
||||
'notification_count': notification_count,
|
||||
'notifications_sent': notifications_sent,
|
||||
'status': status,
|
||||
'created_by': created_by_json(
|
||||
created_by.id,
|
||||
created_by.name,
|
||||
created_by.email_address)
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
def job_json_with_created_by(service_id=None, job_id=None):
|
||||
data = {
|
||||
'id': job_id if job_id else str(generate_uuid()),
|
||||
'service': service_id if service_id else str(generate_uuid()),
|
||||
'template': 1,
|
||||
'template_version': 1,
|
||||
'original_file_name': 'thisisatest.csv',
|
||||
'created_at': str(datetime.now().time()),
|
||||
'notification_count': 1,
|
||||
'notifications_sent': 1,
|
||||
'status': '',
|
||||
'created_by': {'name': 'Test User'}
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
def notification_json(service_id,
|
||||
job=None,
|
||||
template=None,
|
||||
@@ -165,8 +171,6 @@ def notification_json(service_id,
|
||||
created_at=None,
|
||||
updated_at=None,
|
||||
with_links=False):
|
||||
if job is None:
|
||||
job = job_json()
|
||||
if template is None:
|
||||
template = template_json(service_id, str(generate_uuid()))
|
||||
if sent_at is None:
|
||||
@@ -189,7 +193,7 @@ def notification_json(service_id,
|
||||
'id': template['id'],
|
||||
'name': template['name'],
|
||||
'template_type': template['template_type']},
|
||||
'job': {'id': job['id'], 'original_file_name': job['original_file_name']},
|
||||
'job': {'id': job['id'], 'original_file_name': job['original_file_name']} if job else None,
|
||||
'sent_at': sent_at,
|
||||
'status': status,
|
||||
'created_at': created_at,
|
||||
|
||||
@@ -2,7 +2,7 @@ from flask import url_for
|
||||
from bs4 import BeautifulSoup
|
||||
import json
|
||||
from app.utils import generate_notifications_csv
|
||||
from tests import notification_json, job_json_with_created_by
|
||||
from tests import (notification_json, job_json)
|
||||
from tests.conftest import fake_uuid
|
||||
from tests.conftest import mock_get_job as mock_get_job1
|
||||
from freezegun import freeze_time
|
||||
@@ -31,17 +31,16 @@ def test_should_show_page_for_one_job(
|
||||
service_one,
|
||||
active_user_with_permissions,
|
||||
mock_get_service_template,
|
||||
mock_get_job,
|
||||
mocker,
|
||||
mock_get_notifications
|
||||
mock_get_notifications,
|
||||
fake_uuid
|
||||
):
|
||||
data = job_json_with_created_by(service_id=service_one['id'])
|
||||
job_id = data['id']
|
||||
file_name = data['original_file_name']
|
||||
file_name = mock_get_job(service_one['id'], fake_uuid)['data']['original_file_name']
|
||||
with app_.test_request_context():
|
||||
with app_.test_client() as client:
|
||||
client.login(active_user_with_permissions, mocker, service_one)
|
||||
mock_get_job1(mocker=mocker, job_data=data)
|
||||
response = client.get(url_for('main.view_job', service_id=service_one['id'], job_id=job_id))
|
||||
response = client.get(url_for('main.view_job', service_id=service_one['id'], job_id=fake_uuid))
|
||||
|
||||
assert response.status_code == 200
|
||||
content = response.get_data(as_text=True)
|
||||
@@ -55,15 +54,15 @@ def test_should_show_updates_for_one_job_as_json(
|
||||
service_one,
|
||||
active_user_with_permissions,
|
||||
mock_get_notifications,
|
||||
mocker
|
||||
mock_get_job,
|
||||
mocker,
|
||||
fake_uuid
|
||||
):
|
||||
job_id = fake_uuid()
|
||||
data = job_json_with_created_by(job_id, service_one['id'])
|
||||
job_json = mock_get_job(service_one['id'], fake_uuid)['data']
|
||||
with app_.test_request_context():
|
||||
with app_.test_client() as client:
|
||||
client.login(active_user_with_permissions, mocker, service_one)
|
||||
mock_get_job1(mocker=mocker, job_data=data)
|
||||
response = client.get(url_for('main.view_job_updates', service_id=service_one['id'], job_id=job_id))
|
||||
response = client.get(url_for('main.view_job_updates', service_id=service_one['id'], job_id=fake_uuid))
|
||||
|
||||
assert response.status_code == 200
|
||||
content = json.loads(response.get_data(as_text=True))
|
||||
@@ -72,7 +71,7 @@ def test_should_show_updates_for_one_job_as_json(
|
||||
assert 'Recipient' in content['notifications']
|
||||
assert 'Status' in content['notifications']
|
||||
assert 'Started' in content['status']
|
||||
assert 'Uploaded by Test User' in content['status']
|
||||
assert job_json['status'] in content['status']
|
||||
|
||||
|
||||
def test_should_show_notifications_for_a_service(app_,
|
||||
@@ -91,7 +90,7 @@ def test_should_show_notifications_for_a_service(app_,
|
||||
assert notification['to'] in content
|
||||
assert notification['status'] in content
|
||||
assert notification['template']['name'] in content
|
||||
assert '.csv' in content
|
||||
assert 'Download as a CSV file' in content
|
||||
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
|
||||
assert page.h1.string.strip() == 'Activity'
|
||||
|
||||
@@ -119,7 +118,7 @@ def test_can_view_only_sms_notifications_for_a_service(app_,
|
||||
assert notification['to'] in content
|
||||
assert notification['status'] in content
|
||||
assert notification['template']['name'] in content
|
||||
assert '.csv' in content
|
||||
assert 'Download as a CSV file' in content
|
||||
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
|
||||
assert page.h1.string.strip() == 'Text messages'
|
||||
|
||||
@@ -147,7 +146,7 @@ def test_can_view_only_email_notifications_for_a_service(app_,
|
||||
assert notification['to'] in content
|
||||
assert notification['status'] in content
|
||||
assert notification['template']['name'] in content
|
||||
assert '.csv' in content
|
||||
assert 'Download as a CSV file' in content
|
||||
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
|
||||
assert page.h1.string.strip() == 'Emails'
|
||||
|
||||
@@ -173,7 +172,7 @@ def test_can_view_successful_notifications_for_a_service(app_,
|
||||
assert notification['to'] in content
|
||||
assert notification['status'] in content
|
||||
assert notification['template']['name'] in content
|
||||
assert '.csv' in content
|
||||
assert 'Download as a CSV file' in content
|
||||
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
|
||||
assert page.h1.string.strip() == 'Successful emails and text messages'
|
||||
|
||||
@@ -199,7 +198,7 @@ def test_can_view_failed_notifications_for_a_service(app_,
|
||||
assert notification['to'] in content
|
||||
assert notification['status'] in content
|
||||
assert notification['template']['name'] in content
|
||||
assert '.csv' in content
|
||||
assert 'Download as a CSV file' in content
|
||||
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
|
||||
assert page.h1.string.strip() == 'Failed emails and text messages'
|
||||
|
||||
@@ -265,6 +264,30 @@ def test_should_download_notifications_for_a_service(app_,
|
||||
assert 'text/csv' in response.headers['Content-Type']
|
||||
|
||||
|
||||
def test_should_download_api_notifications_for_a_service(app_,
|
||||
api_user_active,
|
||||
mock_login,
|
||||
mock_get_service,
|
||||
mock_get_job,
|
||||
mock_get_notifications,
|
||||
mock_get_template_version,
|
||||
mock_has_permissions,
|
||||
fake_uuid):
|
||||
with app_.test_request_context():
|
||||
with app_.test_client() as client:
|
||||
client.login(api_user_active)
|
||||
response = client.get(url_for(
|
||||
'main.view_notifications',
|
||||
service_id=fake_uuid,
|
||||
download='csv'))
|
||||
notifications = mock_get_notifications(fake_uuid)['notifications']
|
||||
assert all([x['job'] is None for x in notifications])
|
||||
csv_content = generate_notifications_csv(notifications)
|
||||
assert response.status_code == 200
|
||||
assert response.get_data(as_text=True) == csv_content
|
||||
assert 'text/csv' in response.headers['Content-Type']
|
||||
|
||||
|
||||
@freeze_time("2016-01-01 11:09:00.061258")
|
||||
def test_should_download_notifications_for_a_job(app_,
|
||||
api_user_active,
|
||||
|
||||
@@ -7,7 +7,7 @@ from glob import glob
|
||||
from bs4 import BeautifulSoup
|
||||
from flask import url_for
|
||||
from unittest.mock import ANY
|
||||
from tests import validate_route_permission, job_json_with_created_by
|
||||
from tests import (validate_route_permission, job_json)
|
||||
|
||||
template_types = ['email', 'sms']
|
||||
|
||||
@@ -321,20 +321,19 @@ def test_create_job_should_call_api(
|
||||
app_,
|
||||
service_one,
|
||||
active_user_with_permissions,
|
||||
job_data,
|
||||
mock_create_job,
|
||||
mock_get_job,
|
||||
mock_get_notifications,
|
||||
mock_get_service_template,
|
||||
mocker
|
||||
mocker,
|
||||
fake_uuid
|
||||
):
|
||||
service_id = service_one['id']
|
||||
job_id = job_data['id']
|
||||
original_file_name = job_data['original_file_name']
|
||||
template_id = job_data['template']
|
||||
notification_count = job_data['notification_count']
|
||||
|
||||
from tests.conftest import mock_get_job as mock_get_job1
|
||||
data = mock_get_job(service_one['id'], fake_uuid)['data']
|
||||
job_id = data['id']
|
||||
original_file_name = data['original_file_name']
|
||||
template_id = data['template']
|
||||
notification_count = data['notification_count']
|
||||
with app_.test_request_context():
|
||||
with app_.test_client() as client:
|
||||
client.login(active_user_with_permissions, mocker, service_one)
|
||||
@@ -344,9 +343,7 @@ def test_create_job_should_call_api(
|
||||
'notification_count': notification_count,
|
||||
'valid': True}
|
||||
url = url_for('main.start_job', service_id=service_one['id'], upload_id=job_id)
|
||||
data = job_json_with_created_by(service_id=service_one['id'], job_id=job_data['id'])
|
||||
mock_get_job1(mocker=mocker, job_data=data)
|
||||
response = client.post(url, data=job_data, follow_redirects=True)
|
||||
response = client.post(url, data={}, follow_redirects=True)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert original_file_name in response.get_data(as_text=True)
|
||||
@@ -357,14 +354,15 @@ def test_check_messages_should_revalidate_file_when_uploading_file(
|
||||
app_,
|
||||
service_one,
|
||||
active_user_with_permissions,
|
||||
job_data,
|
||||
mock_create_job,
|
||||
mock_get_job,
|
||||
mock_get_service_template_with_placeholders,
|
||||
mock_s3_upload,
|
||||
mocker,
|
||||
mock_has_permissions,
|
||||
mock_get_service_statistics,
|
||||
mock_get_users_by_service
|
||||
mock_get_users_by_service,
|
||||
fake_uuid
|
||||
):
|
||||
|
||||
service_id = service_one['id']
|
||||
@@ -377,16 +375,17 @@ def test_check_messages_should_revalidate_file_when_uploading_file(
|
||||
+447700900986,,,,
|
||||
"""
|
||||
)
|
||||
data = mock_get_job(service_one['id'], fake_uuid)['data']
|
||||
with app_.test_request_context():
|
||||
with app_.test_client() as client:
|
||||
client.login(active_user_with_permissions, mocker, service_one)
|
||||
with client.session_transaction() as session:
|
||||
session['upload_data'] = {'original_file_name': 'invalid.csv',
|
||||
'template_id': job_data['template'],
|
||||
'notification_count': job_data['notification_count'],
|
||||
'template_id': data['template'],
|
||||
'notification_count': data['notification_count'],
|
||||
'valid': True}
|
||||
response = client.post(
|
||||
url_for('main.start_job', service_id=service_id, upload_id=job_data['id']),
|
||||
url_for('main.start_job', service_id=service_id, upload_id=data['id']),
|
||||
data={'file': (BytesIO(''.encode('utf-8')), 'invalid.csv')},
|
||||
content_type='multipart/form-data',
|
||||
follow_redirects=True
|
||||
|
||||
@@ -15,8 +15,7 @@ from . import (
|
||||
notification_json,
|
||||
invite_json,
|
||||
sample_uuid,
|
||||
generate_uuid,
|
||||
job_json_with_created_by)
|
||||
generate_uuid)
|
||||
from app.notify_client.models import (
|
||||
User,
|
||||
InvitedUser
|
||||
@@ -769,42 +768,36 @@ def mock_check_verify_code_code_expired(mocker):
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def job_data():
|
||||
return job_json()
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def mock_create_job(mocker, job_data):
|
||||
def mock_create_job(mocker, api_user_active):
|
||||
def _create(job_id, service_id, template_id, file_name, notification_count):
|
||||
job_data['id'] = job_id
|
||||
job_data['service'] = service_id
|
||||
job_data['template'] = template_id
|
||||
job_data['bucket_name'] = 'service-{}-notify'.format(job_id)
|
||||
job_data['original_file_name'] = file_name
|
||||
job_data['file_name'] = '{}.csv'.format(job_id)
|
||||
job_data['notification_count'] = notification_count
|
||||
job_data = job_json(
|
||||
service_id,
|
||||
api_user_active,
|
||||
job_id=job_id,
|
||||
template_id=template_id,
|
||||
bucket_name='service-{}-notify'.format(job_id),
|
||||
original_file_name='{}.csv'.format(job_id),
|
||||
notification_count=notification_count)
|
||||
return job_data
|
||||
|
||||
return mocker.patch('app.job_api_client.create_job', side_effect=_create)
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def mock_get_job(mocker, job_data):
|
||||
def mock_get_job(mocker, api_user_active):
|
||||
def _get_job(service_id, job_id):
|
||||
job_data['id'] = job_id
|
||||
job_data['service'] = service_id
|
||||
job_data['created_at'] = str(datetime.utcnow())
|
||||
job_data = job_json(service_id, api_user_active, job_id=job_id)
|
||||
return {"data": job_data}
|
||||
|
||||
return mocker.patch('app.job_api_client.get_job', side_effect=_get_job)
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def mock_get_jobs(mocker):
|
||||
def mock_get_jobs(mocker, api_user_active):
|
||||
def _get_jobs(service_id):
|
||||
data = []
|
||||
for i in range(5):
|
||||
job_data = job_json_with_created_by(service_id=service_id)
|
||||
job_data = job_json(service_id, api_user_active)
|
||||
data.append(job_data)
|
||||
return {"data": data}
|
||||
|
||||
@@ -812,7 +805,7 @@ def mock_get_jobs(mocker):
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def mock_get_notifications(mocker):
|
||||
def mock_get_notifications(mocker, api_user_active):
|
||||
def _get_notifications(service_id,
|
||||
job_id=None,
|
||||
page=1,
|
||||
@@ -820,7 +813,10 @@ def mock_get_notifications(mocker):
|
||||
template_type=None,
|
||||
status=None,
|
||||
limit_days=None):
|
||||
return notification_json(service_id)
|
||||
job = None
|
||||
if job_id is not None:
|
||||
job = job_json(service_id, api_user_active, job_id=job_id)
|
||||
return notification_json(service_id, job=job)
|
||||
|
||||
return mocker.patch(
|
||||
'app.notification_api_client.get_notifications_for_service',
|
||||
|
||||
Reference in New Issue
Block a user