darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-11 09:27:56 -04:00
Code Issues Packages Projects Releases Wiki Activity

Remove references to monthly_billing table from api

Browse Source
This commit is contained in:
Pea Tyczynska
2018-07-23 15:14:37 +01:00
parent c0f309a2a6
commit ca2b350a99
10 changed files with 15 additions and 1040 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
app/billing/rest.py
View File

@@ -1,4 +1,3 @@
import json
from datetime import datetime
from flask import Blueprint, jsonify, request
@@ -15,12 +14,8 @@ from app.dao.annual_billing_dao import (
dao_update_annual_billing_for_future_years
)
from app.dao.date_util import get_current_financial_year_start_year
from app.dao.date_util import get_months_for_financial_year
from app.dao.fact_billing_dao import fetch_monthly_billing_for_year, fetch_billing_totals_for_year
from app.dao.monthly_billing_dao import (
get_billing_data_for_financial_year,
get_monthly_billing_by_notification_type
)
from app.errors import InvalidRequest
from app.errors import register_errors
from app.models import SMS_TYPE, EMAIL_TYPE, LETTER_TYPE
@@ -60,41 +55,6 @@ def get_yearly_billing_usage_summary_from_ft_billing(service_id):
return jsonify(data)
@billing_blueprint.route('/monthly-usage')
def get_yearly_usage_by_month(service_id):
try:
year = int(request.args.get('year'))
results = []
for month in get_months_for_financial_year(year):
letter_billing_for_month = get_monthly_billing_by_notification_type(service_id, month, LETTER_TYPE)
if letter_billing_for_month:
results.extend(_transform_billing_for_month_letters(letter_billing_for_month))
billing_for_month = get_monthly_billing_by_notification_type(service_id, month, SMS_TYPE)
if billing_for_month:
results.append(_transform_billing_for_month_sms(billing_for_month))
return jsonify(results)
except TypeError:
return jsonify(result='error', message='No valid year provided'), 400
@billing_blueprint.route('/yearly-usage-summary')
def get_yearly_billing_usage_summary(service_id):
try:
year = int(request.args.get('year'))
billing_data = get_billing_data_for_financial_year(service_id, year)
notification_types = [EMAIL_TYPE, LETTER_TYPE, SMS_TYPE]
response = [
_get_total_billable_units_and_rate_for_notification_type(billing_data, notification_type)
for notification_type in notification_types
]
return json.dumps(response)
except TypeError:
return jsonify(result='error', message='No valid year provided'), 400
def _get_total_billable_units_and_rate_for_notification_type(billing_data, noti_type):
total_sent = 0
rate = 0

8
app/celery/scheduled_tasks.py
View File

@@ -19,7 +19,6 @@ from app.celery.service_callback_tasks import (
)
from app.celery.tasks import process_job
from app.config import QueueNames, TaskNames
from app.dao.date_util import get_month_start_and_end_date_in_utc
from app.dao.inbound_sms_dao import delete_inbound_sms_created_more_than_a_week_ago
from app.dao.invited_org_user_dao import delete_org_invitations_created_more_than_two_days_ago
from app.dao.invited_user_dao import delete_invitations_created_more_than_two_days_ago
@@ -29,10 +28,6 @@ from app.dao.jobs_dao import (
dao_get_jobs_older_than_limited_by
)
from app.dao.jobs_dao import dao_update_job
from app.dao.monthly_billing_dao import (
get_service_ids_that_need_billing_populated,
create_or_update_monthly_billing
)
from app.dao.notifications_dao import (
dao_timeout_notifications,
is_delivery_slow_for_provider,
@@ -67,9 +62,6 @@ from app.models import (
)
from app.notifications.process_notifications import send_notification_to_queue
from app.performance_platform import total_sent_notifications, processing_time
from app.utils import (
convert_utc_to_bst
)
from app.v2.errors import JobIncompleteError

3
app/commands.py
View File

@@ -13,7 +13,6 @@ from sqlalchemy import func
from notifications_utils.statsd_decorators import statsd
from app import db, DATETIME_FORMAT, encryption, redis_store
from app.billing.rest import get_yearly_usage_by_month, get_yearly_usage_by_monthly_from_ft_billing
from app.celery.scheduled_tasks import send_total_sent_notifications_to_performance_platform
from app.celery.service_callback_tasks import send_delivery_status_to_service
from app.celery.letters_pdf_tasks import create_letters_pdf
@@ -33,7 +32,7 @@ from app.dao.services_dao import (
dao_fetch_service_by_id
)
from app.dao.users_dao import (delete_model_user, delete_user_verify_codes)
from app.models import PROVIDERS, User, SMS_TYPE, EMAIL_TYPE, Notification
from app.models import PROVIDERS, User, Notification
from app.performance_platform.processing_time import (send_processing_time_for_start_and_end)
from app.utils import (
cache_key_for_service_template_usage_per_day,

11
app/dao/fact_billing_dao.py
View File

@@ -200,6 +200,17 @@ def get_rates_for_billing():
return non_letter_rates, letter_rates
def get_service_ids_that_need_billing_populated(start_date, end_date):
return db.session.query(
NotificationHistory.service_id
).filter(
NotificationHistory.created_at >= start_date,
NotificationHistory.created_at <= end_date,
NotificationHistory.notification_type.in_([SMS_TYPE, EMAIL_TYPE, LETTER_TYPE]),
NotificationHistory.billable_units != 0
).distinct().all()
def get_rate(non_letter_rates, letter_rates, notification_type, date, crown=None, letter_page_count=None):
if notification_type == LETTER_TYPE:
return next(r[3] for r in letter_rates if date > r[0] and crown == r[1] and letter_page_count == r[2])

128
app/dao/monthly_billing_dao.py
View File

@@ -1,128 +0,0 @@
from datetime import datetime
from notifications_utils.statsd_decorators import statsd
from app import db
from app.dao.dao_utils import transactional
from app.dao.date_util import get_month_start_and_end_date_in_utc, get_financial_year
from app.dao.notification_usage_dao import get_billing_data_for_month
from app.models import (
SMS_TYPE,
EMAIL_TYPE,
LETTER_TYPE,
MonthlyBilling,
NotificationHistory
)
from app.utils import convert_utc_to_bst
def get_service_ids_that_need_billing_populated(start_date, end_date):
return db.session.query(
NotificationHistory.service_id
).filter(
NotificationHistory.created_at >= start_date,
NotificationHistory.created_at <= end_date,
NotificationHistory.notification_type.in_([SMS_TYPE, EMAIL_TYPE, LETTER_TYPE]),
NotificationHistory.billable_units != 0
).distinct().all()
@statsd(namespace="dao")
def create_or_update_monthly_billing(service_id, billing_month):
start_date, end_date = get_month_start_and_end_date_in_utc(billing_month)
_update_monthly_billing(service_id, start_date, end_date, SMS_TYPE)
_update_monthly_billing(service_id, start_date, end_date, EMAIL_TYPE)
_update_monthly_billing(service_id, start_date, end_date, LETTER_TYPE)
def _monthly_billing_data_to_json(billing_data):
results = []
if billing_data:
# total cost must take into account the free allowance.
# might be a good idea to capture free allowance in this table
results = [{
"billing_units": x.billing_units,
"rate_multiplier": x.rate_multiplier,
"international": x.international,
"rate": x.rate,
"total_cost": (x.billing_units * x.rate_multiplier) * x.rate
} for x in billing_data]
return results
@statsd(namespace="dao")
@transactional
def _update_monthly_billing(service_id, start_date, end_date, notification_type):
billing_data = get_billing_data_for_month(
service_id=service_id,
start_date=start_date,
end_date=end_date,
notification_type=notification_type
)
monthly_totals = _monthly_billing_data_to_json(billing_data)
row = get_monthly_billing_entry(service_id, start_date, notification_type)
if row:
row.monthly_totals = monthly_totals
row.updated_at = datetime.utcnow()
else:
row = MonthlyBilling(
service_id=service_id,
notification_type=notification_type,
monthly_totals=monthly_totals,
start_date=start_date,
end_date=end_date
)
db.session.add(row)
def get_monthly_billing_entry(service_id, start_date, notification_type):
entry = MonthlyBilling.query.filter_by(
service_id=service_id,
start_date=start_date,
notification_type=notification_type
).first()
return entry
@statsd(namespace="dao")
def get_yearly_billing_data_for_date_range(
service_id, start_date, end_date, notification_types
):
results = db.session.query(
MonthlyBilling.notification_type,
MonthlyBilling.monthly_totals,
MonthlyBilling.start_date,
).filter(
MonthlyBilling.service_id == service_id,
MonthlyBilling.start_date >= start_date,
MonthlyBilling.end_date <= end_date,
MonthlyBilling.notification_type.in_(notification_types)
).order_by(
MonthlyBilling.start_date,
MonthlyBilling.notification_type,
).all()
return results
@statsd(namespace="dao")
def get_monthly_billing_by_notification_type(service_id, billing_month, notification_type):
billing_month_in_bst = convert_utc_to_bst(billing_month)
start_date, _ = get_month_start_and_end_date_in_utc(billing_month_in_bst)
return get_monthly_billing_entry(service_id, start_date, notification_type)
@statsd(namespace="dao")
def get_billing_data_for_financial_year(service_id, year, notification_types=[SMS_TYPE, EMAIL_TYPE, LETTER_TYPE]):
now = convert_utc_to_bst(datetime.utcnow())
start_date, end_date = get_financial_year(year)
if start_date <= now <= end_date:
# Update totals to the latest so we include data for today
create_or_update_monthly_billing(service_id=service_id, billing_month=now)
results = get_yearly_billing_data_for_date_range(
service_id, start_date, end_date, notification_types
)
return results

1
app/schemas.py
View File

@@ -232,7 +232,6 @@ class ServiceSchema(BaseSchema):
'service_provider_stats',
'service_notification_stats',
'service_sms_senders',
'monthly_billing',
'reply_to_email_addresses',
'letter_contacts',
'complaints',

316
tests/app/billing/test_billing.py
View File

@@ -4,18 +4,12 @@ import json
import pytest
from app.billing.rest import _transform_billing_for_month_sms
from app.dao.monthly_billing_dao import (
create_or_update_monthly_billing,
get_monthly_billing_by_notification_type,
)
from app.models import SMS_TYPE, EMAIL_TYPE, LETTER_TYPE, FactBilling
from app.models import FactBilling
from app.dao.date_util import get_current_financial_year_start_year, get_month_start_and_end_date_in_utc
from app.dao.annual_billing_dao import dao_get_free_sms_fragment_limit_for_year
from tests.app.db import (
create_notification,
create_rate,
create_monthly_billing_entry,
create_annual_billing,
create_template,
create_service,
@@ -36,253 +30,6 @@ def _assert_dict_equals(actual, expected_dict):
assert actual == expected_dict
def test_get_yearly_billing_summary_returns_correct_breakdown(client, sample_template):
create_rate(start_date=IN_MAY_2016 - timedelta(days=1), value=0.12, notification_type=SMS_TYPE)
create_notification(
template=sample_template, created_at=IN_MAY_2016,
billable_units=1, rate_multiplier=2, status='delivered'
)
create_notification(
template=sample_template, created_at=IN_JUN_2016,
billable_units=2, rate_multiplier=3, status='delivered'
)
letter_template = create_template(service=sample_template.service, template_type=LETTER_TYPE)
create_notification(template=letter_template, created_at=IN_MAY_2016, status='delivered', billable_units=1)
create_notification(template=letter_template, created_at=IN_JUN_2016, status='delivered', billable_units=1)
create_or_update_monthly_billing(service_id=sample_template.service_id, billing_month=IN_MAY_2016)
create_or_update_monthly_billing(service_id=sample_template.service_id, billing_month=IN_JUN_2016)
response = client.get(
'/service/{}/billing/yearly-usage-summary?year=2016'.format(sample_template.service.id),
headers=[create_authorization_header()]
)
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json) == 3
_assert_dict_equals(resp_json[0], {
'notification_type': EMAIL_TYPE,
'billing_units': 0,
'rate': 0,
'letter_total': 0
})
_assert_dict_equals(resp_json[1], {
'notification_type': LETTER_TYPE,
'billing_units': 2,
'rate': 0,
'letter_total': 0.66
})
_assert_dict_equals(resp_json[2], {
'notification_type': SMS_TYPE,
'billing_units': 8,
'rate': 0.12,
'letter_total': 0
})
def test_get_yearly_billing_usage_breakdown_returns_400_if_missing_year(client, sample_service):
response = client.get(
'/service/{}/billing/yearly-usage-summary'.format(sample_service.id),
headers=[create_authorization_header()]
)
assert response.status_code == 400
assert json.loads(response.get_data(as_text=True)) == {
'message': 'No valid year provided', 'result': 'error'
}
def test_get_yearly_usage_by_month_returns_400_if_missing_year(client, sample_service):
response = client.get(
'/service/{}/billing/monthly-usage'.format(sample_service.id),
headers=[create_authorization_header()]
)
assert response.status_code == 400
assert json.loads(response.get_data(as_text=True)) == {
'message': 'No valid year provided', 'result': 'error'
}
def test_get_yearly_usage_by_month_returns_empty_list_if_no_usage(client, sample_template):
create_rate(start_date=IN_MAY_2016 - timedelta(days=1), value=0.12, notification_type=SMS_TYPE)
response = client.get(
'/service/{}/billing/monthly-usage?year=2016'.format(sample_template.service.id),
headers=[create_authorization_header()]
)
assert response.status_code == 200
results = json.loads(response.get_data(as_text=True))
assert results == []
def test_get_yearly_usage_by_month_returns_correctly(client, sample_template):
create_rate(start_date=IN_MAY_2016 - timedelta(days=1), value=0.12, notification_type=SMS_TYPE)
create_notification(
template=sample_template, created_at=IN_MAY_2016,
billable_units=1, rate_multiplier=2, status='delivered'
)
create_notification(
template=sample_template, created_at=IN_JUN_2016,
billable_units=2, rate_multiplier=3, status='delivered'
)
create_or_update_monthly_billing(service_id=sample_template.service_id, billing_month=IN_MAY_2016)
create_or_update_monthly_billing(service_id=sample_template.service_id, billing_month=IN_JUN_2016)
response = client.get(
'/service/{}/billing/monthly-usage?year=2016'.format(sample_template.service.id),
headers=[create_authorization_header()]
)
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
_assert_dict_equals(resp_json[0], {
'billing_units': 0,
'month': 'May',
'notification_type': LETTER_TYPE,
'rate': 0
})
_assert_dict_equals(resp_json[1], {
'billing_units': 2,
'month': 'May',
'notification_type': SMS_TYPE,
'rate': 0.12
})
_assert_dict_equals(resp_json[2], {
'billing_units': 0,
'month': 'June',
'notification_type': LETTER_TYPE,
'rate': 0
})
_assert_dict_equals(resp_json[3], {
'billing_units': 6,
'month': 'June',
'notification_type': SMS_TYPE,
'rate': 0.12
})
def test_transform_billing_for_month_returns_empty_if_no_monthly_totals(sample_service):
create_monthly_billing_entry(
service=sample_service,
monthly_totals=[],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
transformed_billing_data = _transform_billing_for_month_sms(get_monthly_billing_by_notification_type(
sample_service.id, APR_2016_MONTH_START, SMS_TYPE
))
_assert_dict_equals(transformed_billing_data, {
'notification_type': SMS_TYPE,
'billing_units': 0,
'month': 'April',
'rate': 0,
})
def test_transform_billing_for_month_formats_monthly_totals_correctly(sample_service):
create_monthly_billing_entry(
service=sample_service,
monthly_totals=[{
"billing_units": 12,
"rate": 0.0158,
"rate_multiplier": 5,
"total_cost": 2.1804,
"international": False
}],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
transformed_billing_data = _transform_billing_for_month_sms(get_monthly_billing_by_notification_type(
sample_service.id, APR_2016_MONTH_START, SMS_TYPE
))
_assert_dict_equals(transformed_billing_data, {
'notification_type': SMS_TYPE,
'billing_units': 60,
'month': 'April',
'rate': 0.0158,
})
def test_transform_billing_sums_billable_units(sample_service):
create_monthly_billing_entry(
service=sample_service,
monthly_totals=[{
'billing_units': 1321,
'international': False,
'month': 'May',
'notification_type': SMS_TYPE,
'rate': 0.12,
'rate_multiplier': 1
}, {
'billing_units': 1,
'international': False,
'month': 'May',
'notification_type': SMS_TYPE,
'rate': 0.12,
'rate_multiplier': 1
}],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
transformed_billing_data = _transform_billing_for_month_sms(get_monthly_billing_by_notification_type(
sample_service.id, APR_2016_MONTH_START, SMS_TYPE
))
_assert_dict_equals(transformed_billing_data, {
'notification_type': SMS_TYPE,
'billing_units': 1322,
'month': 'April',
'rate': 0.12,
})
def test_transform_billing_calculates_with_different_rate_multipliers(sample_service):
create_monthly_billing_entry(
service=sample_service,
monthly_totals=[{
'billing_units': 1321,
'international': False,
'month': 'May',
'notification_type': SMS_TYPE,
'rate': 0.12,
'rate_multiplier': 1
}, {
'billing_units': 1,
'international': False,
'month': 'May',
'notification_type': SMS_TYPE,
'rate': 0.12,
'rate_multiplier': 3
}],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
transformed_billing_data = _transform_billing_for_month_sms(get_monthly_billing_by_notification_type(
sample_service.id, APR_2016_MONTH_START, SMS_TYPE
))
_assert_dict_equals(transformed_billing_data, {
'notification_type': SMS_TYPE,
'billing_units': 1324,
'month': 'April',
'rate': 0.12,
})
def test_create_update_free_sms_fragment_limit_invalid_schema(client, sample_service):
response = client.post('service/{}/billing/free-sms-fragment-limit'.format(sample_service.id),
@@ -480,21 +227,6 @@ def test_get_yearly_usage_by_monthly_from_ft_billing(client, notify_db_session):
assert len(ft_email) == 0
def test_compare_ft_billing_to_monthly_billing(client, notify_db_session):
service = set_up_yearly_data()
monthly_billing_response = client.get('/service/{}/billing/monthly-usage?year=2016'.format(service.id),
headers=[create_authorization_header()])
ft_billing_response = client.get('service/{}/billing/ft-monthly-usage?year=2016'.format(service.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
monthly_billing_json_resp = json.loads(monthly_billing_response.get_data(as_text=True))
ft_billing_json_resp = json.loads(ft_billing_response.get_data(as_text=True))
assert monthly_billing_json_resp == ft_billing_json_resp
def set_up_yearly_data():
service = create_service()
sms_template = create_template(service=service, template_type="sms")
@@ -527,55 +259,9 @@ def set_up_yearly_data():
notification_type='letter',
rate=0.33)
start_date, end_date = get_month_start_and_end_date_in_utc(datetime(2016, int(mon), 1))
create_monthly_billing_entry(service=service, start_date=start_date,
end_date=end_date,
notification_type='sms',
monthly_totals=[
{"rate": 0.0162, "international": False,
"rate_multiplier": 1, "billing_units": int(d),
"total_cost": 0.0162 * int(d)},
{"rate": 0.0162, "international": False,
"rate_multiplier": 2, "billing_units": int(d),
"total_cost": 0.0162 * int(d)}]
)
create_monthly_billing_entry(service=service, start_date=start_date,
end_date=end_date,
notification_type='email',
monthly_totals=[
{"rate": 0, "international": False,
"rate_multiplier": 1, "billing_units": int(d),
"total_cost": 0}]
)
create_monthly_billing_entry(service=service, start_date=start_date,
end_date=end_date,
notification_type='letter',
monthly_totals=[
{"rate": 0.33, "international": False,
"rate_multiplier": 1, "billing_units": int(d),
"total_cost": 0.33 * int(d)}]
)
return service
def test_get_yearly_billing_usage_summary_from_ft_billing_compare_to_monthly_billing(
client, notify_db_session
):
service = set_up_yearly_data()
monthly_billing_response = client.get('/service/{}/billing/yearly-usage-summary?year=2016'.format(service.id),
headers=[create_authorization_header()])
ft_billing_response = client.get('service/{}/billing/ft-yearly-usage-summary?year=2016'.format(service.id),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
monthly_billing_json_resp = json.loads(monthly_billing_response.get_data(as_text=True))
ft_billing_json_resp = json.loads(ft_billing_response.get_data(as_text=True))
assert len(monthly_billing_json_resp) == 3
assert len(ft_billing_json_resp) == 3
for i in range(0, 3):
assert sorted(monthly_billing_json_resp[i]) == sorted(ft_billing_json_resp[i])
def test_get_yearly_billing_usage_summary_from_ft_billing_returns_400_if_missing_year(client, sample_service):
response = client.get(
'/service/{}/billing/ft-yearly-usage-summary'.format(sample_service.id),

3
tests/app/celery/test_scheduled_tasks.py
View File

@@ -61,8 +61,7 @@ from app.utils import get_london_midnight_in_utc
from app.celery.service_callback_tasks import create_delivery_status_callback_data
from app.v2.errors import JobIncompleteError
from tests.app.db import (
create_notification, create_service, create_template, create_job, create_rate,
create_service_callback_api
create_notification, create_service, create_template, create_job, create_service_callback_api
)
from tests.app.conftest import (

521
tests/app/dao/test_monthly_billing.py
View File

@@ -1,521 +0,0 @@
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta
from freezegun import freeze_time
from functools import partial
from app.dao.monthly_billing_dao import (
create_or_update_monthly_billing,
get_monthly_billing_entry,
get_monthly_billing_by_notification_type,
get_service_ids_that_need_billing_populated,
get_billing_data_for_financial_year
)
from app.models import MonthlyBilling, SMS_TYPE, EMAIL_TYPE
from tests.app.conftest import sample_letter_template
from tests.app.db import (
create_notification,
create_rate,
create_service,
create_template,
create_monthly_billing_entry,
)
FEB_2016_MONTH_START = datetime(2016, 2, 1)
FEB_2016_MONTH_END = datetime(2016, 2, 29, 23, 59, 59, 99999)
MAR_2016_MONTH_START = datetime(2016, 3, 1)
MAR_2016_MONTH_END = datetime(2016, 3, 31, 22, 59, 59, 99999)
APR_2016_MONTH_START = datetime(2016, 3, 31, 23, 00, 00)
APR_2016_MONTH_END = datetime(2016, 4, 30, 22, 59, 59, 99999)
MAY_2016_MONTH_START = datetime(2016, 5, 31, 23, 00, 00)
MAY_2016_MONTH_END = MAY_2016_MONTH_START + relativedelta(months=1, seconds=-1)
APR_2017_MONTH_START = datetime(2017, 3, 31, 23, 00, 00)
APR_2017_MONTH_END = datetime(2017, 4, 30, 23, 59, 59, 99999)
JAN_2017_MONTH_START = datetime(2017, 1, 1)
JAN_2017_MONTH_END = datetime(2017, 1, 31, 23, 59, 59, 99999)
FEB_2017 = datetime(2017, 2, 15)
APR_2016 = datetime(2016, 4, 10)
NO_BILLING_DATA = {
"billing_units": 0,
"rate_multiplier": 1,
"international": False,
"rate": 0,
"total_cost": 0
}
def _assert_monthly_billing(monthly_billing, service_id, notification_type, month_start, month_end):
assert monthly_billing.service_id == service_id
assert monthly_billing.notification_type == notification_type
assert monthly_billing.start_date == month_start
assert monthly_billing.end_date == month_end
def _assert_monthly_billing_totals(monthly_billing_totals, expected_dict):
assert sorted(monthly_billing_totals.keys()) == sorted(expected_dict.keys())
assert sorted(monthly_billing_totals.values()) == sorted(expected_dict.values())
def test_get_monthly_billing_by_notification_type_returns_correct_totals(notify_db, notify_db_session):
service = create_service(service_name="Service One")
create_monthly_billing_entry(
service=service,
monthly_totals=[{
"billing_units": 12,
"rate": 0.0158,
"rate_multiplier": 5,
"total_cost": 2.1804,
"international": False
}],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
monthly_billing_data = get_monthly_billing_by_notification_type(service.id, APR_2016, SMS_TYPE)
_assert_monthly_billing(
monthly_billing_data, service.id, 'sms', APR_2016_MONTH_START, APR_2016_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing_data.monthly_totals[0], {
"billing_units": 12,
"rate_multiplier": 5,
"international": False,
"rate": 0.0158,
"total_cost": 2.1804
})
def test_get_monthly_billing_by_notification_type_filters_by_type(notify_db, notify_db_session):
service = create_service(service_name="Service One")
create_monthly_billing_entry(
service=service,
monthly_totals=[{
"billing_units": 138,
"rate": 0.0158,
"rate_multiplier": 1,
"total_cost": 2.1804,
"international": None
}],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
create_monthly_billing_entry(
service=service,
monthly_totals=[],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=EMAIL_TYPE
)
monthly_billing_data = get_monthly_billing_by_notification_type(service.id, APR_2016, EMAIL_TYPE)
_assert_monthly_billing(
monthly_billing_data, service.id, 'email', APR_2016_MONTH_START, APR_2016_MONTH_END
)
assert monthly_billing_data.monthly_totals == []
def test_get_monthly_billing_by_notification_type_normalises_start_date(notify_db, notify_db_session):
service = create_service(service_name="Service One")
create_monthly_billing_entry(
service=service,
monthly_totals=[{
"billing_units": 321,
"rate": 0.0158,
"rate_multiplier": 1,
"total_cost": 2.1804,
"international": None
}],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
monthly_billing_data = get_monthly_billing_by_notification_type(service.id, APR_2016 + timedelta(days=5), SMS_TYPE)
assert monthly_billing_data.start_date == APR_2016_MONTH_START
assert monthly_billing_data.monthly_totals[0]['billing_units'] == 321
def test_add_monthly_billing_for_single_month_populates_correctly(
sample_template, sample_email_template
):
create_rate(start_date=JAN_2017_MONTH_START, value=0.0158, notification_type=SMS_TYPE)
letter_template = sample_letter_template(sample_template.service)
create_notification(
template=sample_template, created_at=JAN_2017_MONTH_START,
billable_units=1, rate_multiplier=2, status='delivered'
)
create_notification(template=sample_email_template, created_at=JAN_2017_MONTH_START,
status='delivered')
create_notification(template=letter_template, created_at=JAN_2017_MONTH_START, status='delivered')
create_or_update_monthly_billing(
service_id=sample_template.service_id,
billing_month=JAN_2017_MONTH_START
)
monthly_billing = MonthlyBilling.query.order_by(MonthlyBilling.notification_type).all()
assert len(monthly_billing) == 3
_assert_monthly_billing(
monthly_billing[0], sample_template.service.id, 'email', JAN_2017_MONTH_START, JAN_2017_MONTH_END
)
_assert_monthly_billing_totals(
monthly_billing[0].monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 1,
"international": False,
"rate": 0.0,
"total_cost": 0
})
_assert_monthly_billing(
monthly_billing[1], sample_template.service.id, 'sms', JAN_2017_MONTH_START, JAN_2017_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing[1].monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 2,
"international": False,
"rate": 0.0158,
"total_cost": 1 * 2 * 0.0158
})
_assert_monthly_billing(
monthly_billing[2], sample_template.service.id, 'letter', JAN_2017_MONTH_START, JAN_2017_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing[2].monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 1,
"international": False,
"rate": 0.33,
"total_cost": 1 * 0.33
})
def test_add_monthly_billing_for_multiple_months_populate_correctly(
sample_template
):
create_rate(start_date=FEB_2016_MONTH_START - timedelta(days=1), value=0.12, notification_type=SMS_TYPE)
create_notification(
template=sample_template, created_at=FEB_2016_MONTH_START,
billable_units=1, rate_multiplier=2, status='delivered'
)
create_notification(
template=sample_template, created_at=MAR_2016_MONTH_START,
billable_units=2, rate_multiplier=3, status='delivered'
)
create_or_update_monthly_billing(service_id=sample_template.service_id, billing_month=FEB_2016_MONTH_START)
create_or_update_monthly_billing(service_id=sample_template.service_id, billing_month=MAR_2016_MONTH_START)
monthly_billing = MonthlyBilling.query.order_by(
MonthlyBilling.notification_type,
MonthlyBilling.start_date
).all()
assert len(monthly_billing) == 6
_assert_monthly_billing(
monthly_billing[0], sample_template.service.id, 'email', FEB_2016_MONTH_START, FEB_2016_MONTH_END
)
assert monthly_billing[0].monthly_totals == []
_assert_monthly_billing(
monthly_billing[1], sample_template.service.id, 'email', MAR_2016_MONTH_START, MAR_2016_MONTH_END
)
assert monthly_billing[1].monthly_totals == []
_assert_monthly_billing(
monthly_billing[2], sample_template.service.id, 'sms', FEB_2016_MONTH_START, FEB_2016_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing[2].monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 2,
"international": False,
"rate": 0.12,
"total_cost": 0.24
})
_assert_monthly_billing(
monthly_billing[3], sample_template.service.id, 'sms', MAR_2016_MONTH_START, MAR_2016_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing[3].monthly_totals[0], {
"billing_units": 2,
"rate_multiplier": 3,
"international": False,
"rate": 0.12,
"total_cost": 0.72
})
_assert_monthly_billing(
monthly_billing[4], sample_template.service.id, 'letter', FEB_2016_MONTH_START, FEB_2016_MONTH_END
)
assert monthly_billing[4].monthly_totals == []
_assert_monthly_billing(
monthly_billing[5], sample_template.service.id, 'letter', MAR_2016_MONTH_START, MAR_2016_MONTH_END
)
assert monthly_billing[5].monthly_totals == []
def test_add_monthly_billing_with_multiple_rates_populate_correctly(
sample_template, sample_email_template
):
letter_template = sample_letter_template(sample_template.service)
create_rate(start_date=JAN_2017_MONTH_START, value=0.0158, notification_type=SMS_TYPE)
create_rate(start_date=JAN_2017_MONTH_START + timedelta(days=5), value=0.123, notification_type=SMS_TYPE)
create_notification(template=sample_template, created_at=JAN_2017_MONTH_START, billable_units=1, status='delivered')
create_notification(
template=sample_template, created_at=JAN_2017_MONTH_START + timedelta(days=6),
billable_units=2, status='delivered'
)
create_notification(template=sample_email_template, created_at=JAN_2017_MONTH_START, status='delivered')
create_notification(template=letter_template, created_at=JAN_2017_MONTH_START, status='delivered',
billable_units=1)
create_or_update_monthly_billing(service_id=sample_template.service_id, billing_month=JAN_2017_MONTH_START)
monthly_billing = MonthlyBilling.query.order_by(MonthlyBilling.notification_type).all()
assert len(monthly_billing) == 3
_assert_monthly_billing(
monthly_billing[0], sample_template.service.id, 'email', JAN_2017_MONTH_START, JAN_2017_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing[0].monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 1,
"international": False,
"rate": 0.0,
"total_cost": 0.0
})
_assert_monthly_billing(
monthly_billing[1], sample_template.service.id, 'sms', JAN_2017_MONTH_START, JAN_2017_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing[1].monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 1,
"international": False,
"rate": 0.0158,
"total_cost": 0.0158
})
_assert_monthly_billing_totals(monthly_billing[1].monthly_totals[1], {
"billing_units": 2,
"rate_multiplier": 1,
"international": False,
"rate": 0.123,
"total_cost": 0.246
})
_assert_monthly_billing(
monthly_billing[2], sample_template.service.id, 'letter', JAN_2017_MONTH_START, JAN_2017_MONTH_END
)
_assert_monthly_billing_totals(monthly_billing[2].monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 1,
"international": False,
"rate": 0.33,
"total_cost": 0.33
})
def test_update_monthly_billing_overwrites_old_totals(sample_template):
create_rate(APR_2016_MONTH_START, 0.123, SMS_TYPE)
create_notification(template=sample_template, created_at=APR_2016_MONTH_START, billable_units=1, status='delivered')
create_or_update_monthly_billing(sample_template.service_id, APR_2016_MONTH_END)
first_update = get_monthly_billing_by_notification_type(sample_template.service_id, APR_2016_MONTH_START, SMS_TYPE)
_assert_monthly_billing(
first_update, sample_template.service.id, 'sms', APR_2016_MONTH_START, APR_2016_MONTH_END
)
_assert_monthly_billing_totals(first_update.monthly_totals[0], {
"billing_units": 1,
"rate_multiplier": 1,
"international": False,
"rate": 0.123,
"total_cost": 0.123
})
first_updated_at = first_update.updated_at
with freeze_time(APR_2016_MONTH_START + timedelta(days=3)):
create_notification(template=sample_template, billable_units=2, status='delivered')
create_or_update_monthly_billing(sample_template.service_id, APR_2016_MONTH_END)
second_update = get_monthly_billing_by_notification_type(sample_template.service_id, APR_2016_MONTH_START, SMS_TYPE)
_assert_monthly_billing_totals(second_update.monthly_totals[0], {
"billing_units": 3,
"rate_multiplier": 1,
"international": False,
"rate": 0.123,
"total_cost": 0.369
})
assert second_update.updated_at == APR_2016_MONTH_START + timedelta(days=3)
assert first_updated_at != second_update.updated_at
def test_get_service_ids_that_need_billing_populated_return_correctly(notify_db_session):
service_1 = create_service(service_name="Service One")
template_1 = create_template(service=service_1)
service_2 = create_service(service_name="Service Two")
template_2 = create_template(service=service_2)
create_notification(template=template_1, created_at=datetime(2017, 6, 30, 13, 30), status='delivered')
create_notification(template=template_1, created_at=datetime(2017, 7, 1, 14, 30), status='delivered')
create_notification(template=template_2, created_at=datetime(2017, 7, 15, 13, 30))
create_notification(template=template_2, created_at=datetime(2017, 7, 31, 13, 30))
services = get_service_ids_that_need_billing_populated(
start_date=datetime(2017, 7, 1), end_date=datetime(2017, 7, 16)
)
expected_services = [service_1.id, service_2.id]
assert sorted([x.service_id for x in services]) == sorted(expected_services)
def test_get_monthly_billing_entry_filters_by_service(notify_db, notify_db_session):
service_1 = create_service(service_name="Service One")
service_2 = create_service(service_name="Service Two")
now = datetime.utcnow()
create_monthly_billing_entry(
service=service_1,
monthly_totals=[],
start_date=now,
end_date=now + timedelta(days=30),
notification_type=SMS_TYPE
)
create_monthly_billing_entry(
service=service_2,
monthly_totals=[],
start_date=now,
end_date=now + timedelta(days=30),
notification_type=SMS_TYPE
)
entry = get_monthly_billing_entry(service_2.id, now, SMS_TYPE)
assert entry.start_date == now
assert entry.service_id == service_2.id
def test_get_yearly_billing_data_for_year_returns_within_year_only(
sample_template
):
monthly_billing_entry = partial(
create_monthly_billing_entry, service=sample_template.service, notification_type=SMS_TYPE
)
monthly_billing_entry(start_date=FEB_2016_MONTH_START, end_date=FEB_2016_MONTH_END)
monthly_billing_entry(
monthly_totals=[{
"billing_units": 138,
"rate": 0.0158,
"rate_multiplier": 1,
"total_cost": 2.1804,
"international": None
}],
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
monthly_billing_entry(start_date=APR_2017_MONTH_START, end_date=APR_2017_MONTH_END)
billing_data = get_billing_data_for_financial_year(sample_template.service.id, 2016, [SMS_TYPE])
assert len(billing_data) == 1
assert billing_data[0].monthly_totals[0]['billing_units'] == 138
def test_get_yearly_billing_data_for_year_returns_multiple_notification_types(sample_template):
monthly_billing_entry = partial(
create_monthly_billing_entry, service=sample_template.service,
start_date=APR_2016_MONTH_START, end_date=APR_2016_MONTH_END
)
monthly_billing_entry(
notification_type=SMS_TYPE, monthly_totals=[]
)
monthly_billing_entry(
notification_type=EMAIL_TYPE,
monthly_totals=[{
"billing_units": 2,
"rate": 1.3,
"rate_multiplier": 3,
"total_cost": 2.1804,
"international": False
}]
)
billing_data = get_billing_data_for_financial_year(
service_id=sample_template.service.id,
year=2016,
notification_types=[SMS_TYPE, EMAIL_TYPE]
)
assert len(billing_data) == 2
assert billing_data[0].notification_type == EMAIL_TYPE
assert billing_data[0].monthly_totals[0]['billing_units'] == 2
assert billing_data[1].notification_type == SMS_TYPE
@freeze_time("2016-04-21 11:00:00")
def test_get_yearly_billing_data_for_year_includes_current_day_totals(sample_template):
create_rate(start_date=FEB_2016_MONTH_START, value=0.0158, notification_type=SMS_TYPE)
create_monthly_billing_entry(
service=sample_template.service,
start_date=APR_2016_MONTH_START,
end_date=APR_2016_MONTH_END,
notification_type=SMS_TYPE
)
billing_data = get_billing_data_for_financial_year(
service_id=sample_template.service.id,
year=2016,
notification_types=[SMS_TYPE]
)
assert len(billing_data) == 1
assert billing_data[0].notification_type == SMS_TYPE
assert billing_data[0].monthly_totals == []
create_notification(
template=sample_template,
created_at=datetime.utcnow(),
sent_at=datetime.utcnow(),
status='sending',
billable_units=3
)
billing_data = get_billing_data_for_financial_year(
service_id=sample_template.service.id,
year=2016,
notification_types=[SMS_TYPE]
)
assert billing_data[0].monthly_totals[0]['billing_units'] == 3
@freeze_time("2017-06-16 13:00:00")
def test_get_billing_data_for_financial_year_updated_monthly_billing_if_today_is_in_current_year(
sample_service,
mocker
):
mock = mocker.patch("app.dao.monthly_billing_dao.create_or_update_monthly_billing")
get_billing_data_for_financial_year(sample_service.id, 2016)
mock.assert_not_called()

22
tests/app/db.py
View File

@@ -14,7 +14,6 @@ from app.models import (
InboundSms,
InboundNumber,
Job,
MonthlyBilling,
Notification,
EmailBranding,
Organisation,
@@ -382,27 +381,6 @@ def create_inbound_number(number, provider='mmg', active=True, service_id=None):
return inbound_number
def create_monthly_billing_entry(
service,
start_date,
end_date,
notification_type,
monthly_totals=[]
):
entry = MonthlyBilling(
service_id=service.id,
notification_type=notification_type,
monthly_totals=monthly_totals,
start_date=start_date,
end_date=end_date
)
db.session.add(entry)
db.session.commit()
return entry
def create_reply_to_email(
service,
email_address,