Files
notifications-api/tests/app/organisation/test_rest.py
Pea Tyczynska a74d1b026f Fix calculating remaining free allowance for SMS
The way it was done before, the remainder was incorrect in the
billing report and in the org usage query - it was the sms remainder
left at the start of the report period, not at the end of that period.

This became apparent when we tried to show sms_remainder on the org
usage report, where start date is always the start of the financial year.
We saw that sms sent by services did not reduce their free allowance
remainder according to the report. As a result of this, we had to
temporarily remove of sms_remainder column from the report, until
we fix the bug - it has been fixed now, yay!

I think the bug has snuck in partially because our fixtures for testing
this part of the code are quite complex, so it was
harder to see that numbers don't add up. I have added comments
to the tests to try and make it a bit clearer why the results are
as they are.

I also added comments to the code, and renamed some variables,
to make it easier to understand, as there are quite a few
moving parts in it - subqueries and the like.
I also renamed the fetch_sms_free_allowance_remainder method to
fetch_sms_free_allowance_remainder_until_date so it is clearer
what it does.
2021-12-09 18:58:10 +00:00

898 lines
28 KiB
Python

import uuid
from datetime import datetime
import pytest
from freezegun import freeze_time
from sqlalchemy.exc import SQLAlchemyError
from app.dao.organisation_dao import (
dao_add_service_to_organisation,
dao_add_user_to_organisation,
)
from app.dao.services_dao import dao_archive_service
from app.models import AnnualBilling, Organisation
from tests.app.db import (
create_annual_billing,
create_domain,
create_email_branding,
create_ft_billing,
create_letter_branding,
create_organisation,
create_service,
create_template,
create_user,
)
def test_get_all_organisations(admin_request, notify_db_session):
create_organisation(name='inactive org', active=False, organisation_type='nhs_central')
create_organisation(name='active org', domains=['example.com'])
response = admin_request.get(
'organisation.get_organisations',
_expected_status=200
)
assert len(response) == 2
assert set(response[0].keys()) == set(response[1].keys()) == {
'name',
'id',
'active',
'count_of_live_services',
'domains',
'organisation_type',
}
assert response[0]['name'] == 'active org'
assert response[0]['active'] is True
assert response[0]['count_of_live_services'] == 0
assert response[0]['domains'] == ['example.com']
assert response[0]['organisation_type'] is None
assert response[1]['name'] == 'inactive org'
assert response[1]['active'] is False
assert response[1]['count_of_live_services'] == 0
assert response[1]['domains'] == []
assert response[1]['organisation_type'] == 'nhs_central'
def test_get_organisation_by_id(admin_request, notify_db_session):
org = create_organisation()
response = admin_request.get(
'organisation.get_organisation_by_id',
_expected_status=200,
organisation_id=org.id
)
assert set(response.keys()) == {
'id',
'name',
'active',
'crown',
'organisation_type',
'agreement_signed',
'agreement_signed_at',
'agreement_signed_by_id',
'agreement_signed_version',
'agreement_signed_on_behalf_of_name',
'agreement_signed_on_behalf_of_email_address',
'letter_branding_id',
'email_branding_id',
'domains',
'request_to_go_live_notes',
'count_of_live_services',
'notes',
'billing_contact_names',
'billing_contact_email_addresses',
'billing_reference',
'purchase_order_number'
}
assert response['id'] == str(org.id)
assert response['name'] == 'test_org_1'
assert response['active'] is True
assert response['crown'] is None
assert response['organisation_type'] is None
assert response['agreement_signed'] is None
assert response['agreement_signed_by_id'] is None
assert response['agreement_signed_version'] is None
assert response['letter_branding_id'] is None
assert response['email_branding_id'] is None
assert response['domains'] == []
assert response['request_to_go_live_notes'] is None
assert response['count_of_live_services'] == 0
assert response['agreement_signed_on_behalf_of_name'] is None
assert response['agreement_signed_on_behalf_of_email_address'] is None
assert response['notes'] is None
assert response['billing_contact_names'] is None
assert response['billing_contact_email_addresses'] is None
assert response['billing_reference'] is None
assert response['purchase_order_number'] is None
def test_get_organisation_by_id_returns_domains(admin_request, notify_db_session):
org = create_organisation(domains=[
'foo.gov.uk',
'bar.gov.uk',
])
response = admin_request.get(
'organisation.get_organisation_by_id',
_expected_status=200,
organisation_id=org.id
)
assert set(response['domains']) == {
'foo.gov.uk',
'bar.gov.uk',
}
@pytest.mark.parametrize('domain, expected_status', (
('foo.gov.uk', 200),
('bar.gov.uk', 200),
('oof.gov.uk', 404),
pytest.param(
'rab.gov.uk', 200,
marks=pytest.mark.xfail(raises=AssertionError),
),
(None, 400),
('personally.identifying.information@example.com', 400),
))
def test_get_organisation_by_domain(
admin_request,
notify_db_session,
domain,
expected_status
):
org = create_organisation()
other_org = create_organisation('Other organisation')
create_domain('foo.gov.uk', org.id)
create_domain('bar.gov.uk', org.id)
create_domain('rab.gov.uk', other_org.id)
response = admin_request.get(
'organisation.get_organisation_by_domain',
_expected_status=expected_status,
domain=domain,
)
if expected_status == 200:
assert response['id'] == str(org.id)
else:
assert response['result'] == 'error'
@pytest.mark.parametrize('crown', [True, False])
def test_post_create_organisation(admin_request, notify_db_session, crown):
data = {
'name': 'test organisation',
'active': True,
'crown': crown,
'organisation_type': 'local',
}
response = admin_request.post(
'organisation.create_organisation',
_data=data,
_expected_status=201
)
organisation = Organisation.query.all()
assert data['name'] == response['name']
assert data['active'] == response['active']
assert data['crown'] == response['crown']
assert data['organisation_type'] == response['organisation_type']
assert len(organisation) == 1
def test_post_create_organisation_existing_name_raises_400(admin_request, sample_organisation):
data = {
'name': sample_organisation.name,
'active': True,
'crown': True,
'organisation_type': 'central',
}
response = admin_request.post(
'organisation.create_organisation',
_data=data,
_expected_status=400
)
organisation = Organisation.query.all()
assert len(organisation) == 1
assert response['message'] == 'Organisation name already exists'
@pytest.mark.parametrize('data, expected_error', (
({
'active': False,
'crown': True,
'organisation_type': 'central',
}, 'name is a required property'),
({
'active': False,
'name': 'Service name',
'organisation_type': 'central',
}, 'crown is a required property'),
({
'active': False,
'name': 'Service name',
'crown': True,
}, 'organisation_type is a required property'),
({
'active': False,
'name': 'Service name',
'crown': None,
'organisation_type': 'central',
}, 'crown None is not of type boolean'),
({
'active': False,
'name': 'Service name',
'crown': False,
'organisation_type': 'foo',
}, (
'organisation_type foo is not one of '
'[central, local, nhs_central, nhs_local, nhs_gp, emergency_service, school_or_college, other]'
)),
))
def test_post_create_organisation_with_missing_data_gives_validation_error(
admin_request,
notify_db_session,
data,
expected_error,
):
response = admin_request.post(
'organisation.create_organisation',
_data=data,
_expected_status=400
)
assert len(response['errors']) == 1
assert response['errors'][0]['error'] == 'ValidationError'
assert response['errors'][0]['message'] == expected_error
@pytest.mark.parametrize('crown', (
None, True, False
))
def test_post_update_organisation_updates_fields(
admin_request,
notify_db_session,
crown,
):
org = create_organisation()
data = {
'name': 'new organisation name',
'active': False,
'crown': crown,
'organisation_type': 'central',
}
assert org.crown is None
admin_request.post(
'organisation.update_organisation',
_data=data,
organisation_id=org.id,
_expected_status=204
)
organisation = Organisation.query.all()
assert len(organisation) == 1
assert organisation[0].id == org.id
assert organisation[0].name == data['name']
assert organisation[0].active == data['active']
assert organisation[0].crown == crown
assert organisation[0].domains == []
assert organisation[0].organisation_type == 'central'
@pytest.mark.parametrize('domain_list', (
['example.com'],
['example.com', 'example.org', 'example.net'],
[],
))
def test_post_update_organisation_updates_domains(
admin_request,
notify_db_session,
domain_list,
):
org = create_organisation(name='test_org_2')
data = {
'domains': domain_list,
}
admin_request.post(
'organisation.update_organisation',
_data=data,
organisation_id=org.id,
_expected_status=204
)
organisation = Organisation.query.all()
assert len(organisation) == 1
assert [
domain.domain for domain in organisation[0].domains
] == domain_list
def test_update_other_organisation_attributes_doesnt_clear_domains(
admin_request,
notify_db_session,
):
org = create_organisation(name='test_org_2')
create_domain('example.gov.uk', org.id)
admin_request.post(
'organisation.update_organisation',
_data={
'crown': True,
},
organisation_id=org.id,
_expected_status=204
)
assert [
domain.domain for domain in org.domains
] == [
'example.gov.uk'
]
def test_update_organisation_default_branding(
admin_request,
notify_db_session,
):
org = create_organisation(name='Test Organisation')
email_branding = create_email_branding()
letter_branding = create_letter_branding()
assert org.email_branding is None
assert org.letter_branding is None
admin_request.post(
'organisation.update_organisation',
_data={
'email_branding_id': str(email_branding.id),
'letter_branding_id': str(letter_branding.id),
},
organisation_id=org.id,
_expected_status=204
)
assert org.email_branding == email_branding
assert org.letter_branding == letter_branding
def test_post_update_organisation_raises_400_on_existing_org_name(
admin_request, sample_organisation):
org = create_organisation()
data = {
'name': sample_organisation.name,
'active': False
}
response = admin_request.post(
'organisation.update_organisation',
_data=data,
organisation_id=org.id,
_expected_status=400
)
assert response['message'] == 'Organisation name already exists'
def test_post_update_organisation_gives_404_status_if_org_does_not_exist(admin_request, notify_db_session):
data = {'name': 'new organisation name'}
admin_request.post(
'organisation.update_organisation',
_data=data,
organisation_id='31d42ce6-3dac-45a7-95cb-94423d5ca03c',
_expected_status=404
)
organisation = Organisation.query.all()
assert not organisation
def test_post_update_organisation_returns_400_if_domain_is_duplicate(admin_request, notify_db_session):
org = create_organisation()
org2 = create_organisation(name='Second org')
create_domain('same.com', org.id)
data = {'domains': ['new.com', 'same.com']}
response = admin_request.post(
'organisation.update_organisation',
_data=data,
organisation_id=org2.id,
_expected_status=400
)
assert response['message'] == 'Domain already exists'
def test_post_update_organisation_set_mou_doesnt_email_if_no_signed_by(
sample_organisation,
admin_request,
mocker
):
queue_mock = mocker.patch('app.organisation.rest.send_notification_to_queue')
data = {'agreement_signed': True}
admin_request.post(
'organisation.update_organisation',
_data=data,
organisation_id=sample_organisation.id,
_expected_status=204
)
assert queue_mock.called is False
@pytest.mark.parametrize('on_behalf_of_name, on_behalf_of_email_address, templates_and_recipients', [
(
None,
None,
{
'MOU_SIGNER_RECEIPT_TEMPLATE_ID': 'notify@digital.cabinet-office.gov.uk',
}
),
(
'Important Person',
'important@person.com',
{
'MOU_SIGNED_ON_BEHALF_ON_BEHALF_RECEIPT_TEMPLATE_ID': 'important@person.com',
'MOU_SIGNED_ON_BEHALF_SIGNER_RECEIPT_TEMPLATE_ID': 'notify@digital.cabinet-office.gov.uk',
}
),
])
def test_post_update_organisation_set_mou_emails_signed_by(
sample_organisation,
admin_request,
mou_signed_templates,
mocker,
sample_user,
on_behalf_of_name,
on_behalf_of_email_address,
templates_and_recipients
):
queue_mock = mocker.patch('app.organisation.rest.send_notification_to_queue')
sample_organisation.agreement_signed_on_behalf_of_name = on_behalf_of_name
sample_organisation.agreement_signed_on_behalf_of_email_address = on_behalf_of_email_address
admin_request.post(
'organisation.update_organisation',
_data={'agreement_signed': True, 'agreement_signed_by_id': str(sample_user.id)},
organisation_id=sample_organisation.id,
_expected_status=204
)
notifications = [x[0][0] for x in queue_mock.call_args_list]
assert {n.template.name: n.to for n in notifications} == templates_and_recipients
for n in notifications:
# we pass in the same personalisation for all templates (though some templates don't use all fields)
assert n.personalisation == {
'mou_link': 'http://localhost:6012/agreement/non-crown.pdf',
'org_name': 'sample organisation',
'org_dashboard_link': 'http://localhost:6012/organisations/{}'.format(sample_organisation.id),
'signed_by_name': 'Test User',
'on_behalf_of_name': on_behalf_of_name
}
def test_post_link_service_to_organisation(admin_request, sample_service):
data = {
'service_id': str(sample_service.id)
}
organisation = create_organisation(organisation_type='central')
admin_request.post(
'organisation.link_service_to_organisation',
_data=data,
organisation_id=organisation.id,
_expected_status=204
)
assert len(organisation.services) == 1
assert sample_service.organisation_type == 'central'
def test_post_link_service_to_organisation_inserts_annual_billing(admin_request, sample_service):
data = {
'service_id': str(sample_service.id)
}
organisation = create_organisation(organisation_type='central')
assert len(organisation.services) == 0
assert len(AnnualBilling.query.all()) == 0
admin_request.post(
'organisation.link_service_to_organisation',
_data=data,
organisation_id=organisation.id,
_expected_status=204
)
annual_billing = AnnualBilling.query.all()
assert len(annual_billing) == 1
assert annual_billing[0].free_sms_fragment_limit == 150000
def test_post_link_service_to_organisation_rollback_service_if_annual_billing_update_fails(
admin_request, sample_service, mocker
):
mocker.patch('app.dao.annual_billing_dao.dao_create_or_update_annual_billing_for_year',
side_effect=SQLAlchemyError)
data = {
'service_id': str(sample_service.id)
}
assert not sample_service.organisation_type
organisation = create_organisation(organisation_type='central')
assert len(organisation.services) == 0
assert len(AnnualBilling.query.all()) == 0
with pytest.raises(expected_exception=SQLAlchemyError):
admin_request.post(
'organisation.link_service_to_organisation',
_data=data,
organisation_id=organisation.id
)
assert not sample_service.organisation_type
assert len(organisation.services) == 0
assert len(AnnualBilling.query.all()) == 0
def test_post_link_service_to_another_org(
admin_request, sample_service, sample_organisation):
data = {
'service_id': str(sample_service.id)
}
assert len(sample_organisation.services) == 0
assert not sample_service.organisation_type
admin_request.post(
'organisation.link_service_to_organisation',
_data=data,
organisation_id=sample_organisation.id,
_expected_status=204
)
assert len(sample_organisation.services) == 1
assert not sample_service.organisation_type
new_org = create_organisation(organisation_type='central')
admin_request.post(
'organisation.link_service_to_organisation',
_data=data,
organisation_id=new_org.id,
_expected_status=204
)
assert not sample_organisation.services
assert len(new_org.services) == 1
assert sample_service.organisation_type == 'central'
annual_billing = AnnualBilling.query.all()
assert len(annual_billing) == 1
assert annual_billing[0].free_sms_fragment_limit == 150000
def test_post_link_service_to_organisation_nonexistent_organisation(
admin_request, sample_service, fake_uuid):
data = {
'service_id': str(sample_service.id)
}
admin_request.post(
'organisation.link_service_to_organisation',
_data=data,
organisation_id=fake_uuid,
_expected_status=404
)
def test_post_link_service_to_organisation_nonexistent_service(
admin_request, sample_organisation, fake_uuid):
data = {
'service_id': fake_uuid
}
admin_request.post(
'organisation.link_service_to_organisation',
_data=data,
organisation_id=str(sample_organisation.id),
_expected_status=404
)
def test_post_link_service_to_organisation_missing_payload(
admin_request, sample_organisation, fake_uuid):
admin_request.post(
'organisation.link_service_to_organisation',
organisation_id=str(sample_organisation.id),
_expected_status=400
)
def test_rest_get_organisation_services(
admin_request, sample_organisation, sample_service):
dao_add_service_to_organisation(sample_service, sample_organisation.id)
response = admin_request.get(
'organisation.get_organisation_services',
organisation_id=str(sample_organisation.id),
_expected_status=200
)
assert response == [sample_service.serialize_for_org_dashboard()]
def test_rest_get_organisation_services_is_ordered_by_name(
admin_request, sample_organisation, sample_service):
service_2 = create_service(service_name='service 2')
service_1 = create_service(service_name='service 1')
dao_add_service_to_organisation(service_1, sample_organisation.id)
dao_add_service_to_organisation(service_2, sample_organisation.id)
dao_add_service_to_organisation(sample_service, sample_organisation.id)
response = admin_request.get(
'organisation.get_organisation_services',
organisation_id=str(sample_organisation.id),
_expected_status=200
)
assert response[0]['name'] == sample_service.name
assert response[1]['name'] == service_1.name
assert response[2]['name'] == service_2.name
def test_rest_get_organisation_services_inactive_services_at_end(
admin_request, sample_organisation):
inactive_service = create_service(service_name='inactive service', active=False)
service = create_service()
inactive_service_1 = create_service(service_name='inactive service 1', active=False)
dao_add_service_to_organisation(inactive_service, sample_organisation.id)
dao_add_service_to_organisation(service, sample_organisation.id)
dao_add_service_to_organisation(inactive_service_1, sample_organisation.id)
response = admin_request.get(
'organisation.get_organisation_services',
organisation_id=str(sample_organisation.id),
_expected_status=200
)
assert response[0]['name'] == service.name
assert response[1]['name'] == inactive_service.name
assert response[2]['name'] == inactive_service_1.name
def test_add_user_to_organisation_returns_added_user(admin_request, sample_organisation, sample_user):
response = admin_request.post(
'organisation.add_user_to_organisation',
organisation_id=str(sample_organisation.id),
user_id=str(sample_user.id),
_expected_status=200
)
assert response['data']['id'] == str(sample_user.id)
assert len(response['data']['organisations']) == 1
assert response['data']['organisations'][0] == str(sample_organisation.id)
def test_add_user_to_organisation_returns_404_if_user_does_not_exist(admin_request, sample_organisation):
admin_request.post(
'organisation.add_user_to_organisation',
organisation_id=str(sample_organisation.id),
user_id=str(uuid.uuid4()),
_expected_status=404
)
def test_get_organisation_users_returns_users_for_organisation(admin_request, sample_organisation):
first = create_user(email='first@invited.com')
second = create_user(email='another@invited.com')
dao_add_user_to_organisation(organisation_id=sample_organisation.id, user_id=first.id)
dao_add_user_to_organisation(organisation_id=sample_organisation.id, user_id=second.id)
response = admin_request.get(
'organisation.get_organisation_users',
organisation_id=sample_organisation.id,
_expected_status=200
)
assert len(response['data']) == 2
assert response['data'][0]['id'] == str(first.id)
def test_is_organisation_name_unique_returns_200_if_unique(admin_request, notify_db, notify_db_session):
organisation = create_organisation(name='unique')
response = admin_request.get(
'organisation.is_organisation_name_unique',
_expected_status=200,
org_id=organisation.id,
name='something'
)
assert response == {"result": True}
@pytest.mark.parametrize('name', ["UNIQUE", "Unique.", "**uniQUE**"])
def test_is_organisation_name_unique_returns_200_and_name_capitalized_or_punctuation_added(
admin_request,
notify_db,
notify_db_session,
name
):
organisation = create_organisation(name='unique')
response = admin_request.get(
'organisation.is_organisation_name_unique',
_expected_status=200,
org_id=organisation.id,
name=name
)
assert response == {"result": True}
@pytest.mark.parametrize('name', ["UNIQUE", "Unique"])
def test_is_organisation_name_unique_returns_200_and_false_with_same_name_and_different_case_of_other_organisation(
admin_request,
notify_db,
notify_db_session,
name
):
create_organisation(name='unique')
different_organisation_id = '111aa111-2222-bbbb-aaaa-111111111111'
response = admin_request.get(
'organisation.is_organisation_name_unique',
_expected_status=200,
org_id=different_organisation_id,
name=name
)
assert response == {"result": False}
def test_is_organisation_name_unique_returns_200_and_false_if_name_exists_for_a_different_organisation(
admin_request,
notify_db,
notify_db_session
):
create_organisation(name='existing name')
different_organisation_id = '111aa111-2222-bbbb-aaaa-111111111111'
response = admin_request.get(
'organisation.is_organisation_name_unique',
_expected_status=200,
org_id=different_organisation_id,
name='existing name'
)
assert response == {"result": False}
def test_is_organisation_name_unique_returns_200_and_true_if_name_exists_for_the_same_organisation(
admin_request,
notify_db,
notify_db_session
):
organisation = create_organisation(name='unique')
response = admin_request.get(
'organisation.is_organisation_name_unique',
_expected_status=200,
org_id=organisation.id,
name='unique'
)
assert response == {"result": True}
def test_is_organisation_name_unique_returns_400_when_name_does_not_exist(admin_request):
response = admin_request.get(
'organisation.is_organisation_name_unique',
_expected_status=400
)
assert response["message"][0]["org_id"] == ["Can't be empty"]
assert response["message"][1]["name"] == ["Can't be empty"]
@freeze_time('2020-02-24 13:30')
def test_get_organisation_services_usage(admin_request, notify_db_session):
org = create_organisation(name='Organisation without live services')
service = create_service()
template = create_template(service=service)
dao_add_service_to_organisation(service=service, organisation_id=org.id)
create_annual_billing(service_id=service.id, free_sms_fragment_limit=10, financial_year_start=2019)
create_ft_billing(bst_date=datetime.utcnow().date(), template=template, billable_unit=19, rate=0.060,
notifications_sent=19)
response = admin_request.get(
'organisation.get_organisation_services_usage',
organisation_id=org.id,
**{"year": 2019}
)
assert len(response) == 1
assert len(response['services']) == 1
service_usage = response['services'][0]
assert service_usage['service_id'] == str(service.id)
assert service_usage['service_name'] == service.name
assert service_usage['chargeable_billable_sms'] == 9.0
assert service_usage['emails_sent'] == 0
assert service_usage['free_sms_limit'] == 10
assert service_usage['letter_cost'] == 0
assert service_usage['sms_billable_units'] == 19
assert service_usage['sms_remainder'] == 0
assert service_usage['sms_cost'] == 0.54
@freeze_time('2020-02-24 13:30')
def test_get_organisation_services_usage_sort_active_first(admin_request, notify_db_session):
org = create_organisation(name='Organisation without live services')
service = create_service(service_name='live service')
archived_service = create_service(service_name='archived_service')
template = create_template(service=service)
dao_add_service_to_organisation(service=service, organisation_id=org.id)
dao_add_service_to_organisation(service=archived_service, organisation_id=org.id)
create_annual_billing(service_id=service.id, free_sms_fragment_limit=10, financial_year_start=2019)
create_ft_billing(bst_date=datetime.utcnow().date(), template=template, billable_unit=19, rate=0.060,
notifications_sent=19)
response = admin_request.get(
'organisation.get_organisation_services_usage',
organisation_id=org.id,
**{"year": 2019}
)
assert len(response) == 1
assert len(response['services']) == 2
first_service = response['services'][0]
assert first_service['service_id'] == str(archived_service.id)
assert first_service['service_name'] == archived_service.name
assert first_service['active'] is True
last_service = response['services'][1]
assert last_service['service_id'] == str(service.id)
assert last_service['service_name'] == service.name
assert last_service['active'] is True
dao_archive_service(service_id=archived_service.id)
response_after_archive = admin_request.get(
'organisation.get_organisation_services_usage',
organisation_id=org.id,
**{"year": 2019}
)
first_service = response_after_archive['services'][0]
assert first_service['service_id'] == str(service.id)
assert first_service['service_name'] == service.name
assert first_service['active'] is True
last_service = response_after_archive['services'][1]
assert last_service['service_id'] == str(archived_service.id)
assert last_service['service_name'] == archived_service.name
assert last_service['active'] is False
def test_get_organisation_services_usage_returns_400_if_year_is_invalid(admin_request):
response = admin_request.get(
'organisation.get_organisation_services_usage',
organisation_id=uuid.uuid4(),
**{"year": 'not-a-valid-year'},
_expected_status=400
)
assert response['message'] == 'No valid year provided'
def test_get_organisation_services_usage_returns_400_if_year_is_empty(admin_request):
response = admin_request.get(
'organisation.get_organisation_services_usage',
organisation_id=uuid.uuid4(),
_expected_status=400
)
assert response['message'] == 'No valid year provided'