darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-03 18:01:08 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1845 from alphagov/use_ft_billing_for_usage_page

Browse Source 
Use ft billing for usage page
This commit is contained in:
Rebecca Law
2018-04-26 09:47:03 +01:00
committed by GitHub
parent 731f29f2f7 51af6b27a0
commit f3dc290bcf
5 changed files with 540 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
app/celery/reporting_tasks.py
View File

@@ -1,29 +1,12 @@
from datetime import datetime, timedelta, time
from app.models import (Notification,
Rate,
NOTIFICATION_CREATED,
NOTIFICATION_TECHNICAL_FAILURE,
KEY_TYPE_TEST,
LetterRate,
FactBilling,
Service,
LETTER_TYPE, SMS_TYPE)
from app import db
from sqlalchemy import func, desc, case
from datetime import datetime, timedelta
from notifications_utils.statsd_decorators import statsd
from app import notify_celery
from flask import current_app
from app.utils import convert_bst_to_utc
def get_rate(non_letter_rates, letter_rates, notification_type, date, crown=None, rate_multiplier=None):
if notification_type == LETTER_TYPE:
return next(r[3] for r in letter_rates if date > r[0] and crown == r[1] and rate_multiplier == r[2])
elif notification_type == SMS_TYPE:
return next(r[2] for r in non_letter_rates if notification_type == r[0] and date > r[1])
else:
return 0
from app.dao.fact_billing_dao import (
fetch_billing_data_for_day,
update_fact_billing
)
@notify_celery.task(name="create-nightly-billing")
@@ -34,91 +17,10 @@ def create_nightly_billing(day_start=None):
if day_start is None:
day_start = datetime.today() - timedelta(days=1)
non_letter_rates = [(r.notification_type, r.valid_from, r.rate) for r in
Rate.query.order_by(desc(Rate.valid_from)).all()]
letter_rates = [(r.start_date, r.crown, r.sheet_count, r.rate) for r in
LetterRate.query.order_by(desc(LetterRate.start_date)).all()]
for i in range(0, 3):
process_day = day_start - timedelta(days=i)
ds = convert_bst_to_utc(datetime.combine(process_day, time.min))
de = convert_bst_to_utc(datetime.combine(process_day + timedelta(days=1), time.min))
transit_data = db.session.query(
Notification.template_id,
Notification.service_id,
Notification.notification_type,
func.coalesce(Notification.sent_by,
case(
[
(Notification.notification_type == 'letter', 'dvla'),
(Notification.notification_type == 'sms', 'unknown'),
(Notification.notification_type == 'email', 'ses')
]),
).label('sent_by'),
func.coalesce(Notification.rate_multiplier, 1).label('rate_multiplier'),
func.coalesce(Notification.international, False).label('international'),
func.sum(Notification.billable_units).label('billable_units'),
func.count().label('notifications_sent'),
Service.crown,
).filter(
Notification.status != NOTIFICATION_CREATED, # at created status, provider information is not available
Notification.status != NOTIFICATION_TECHNICAL_FAILURE,
Notification.key_type != KEY_TYPE_TEST,
Notification.created_at >= ds,
Notification.created_at < de
).group_by(
Notification.template_id,
Notification.service_id,
Notification.notification_type,
'sent_by',
Notification.rate_multiplier,
Notification.international,
Service.crown
).join(
Service
).all()
updated_records = 0
inserted_records = 0
transit_data = fetch_billing_data_for_day(process_day=process_day)
for data in transit_data:
update_count = FactBilling.query.filter(
FactBilling.bst_date == datetime.date(process_day),
FactBilling.template_id == data.template_id,
FactBilling.service_id == data.service_id,
FactBilling.provider == data.sent_by, # This could be zero - this is a bug that needs to be fixed.
FactBilling.rate_multiplier == data.rate_multiplier,
FactBilling.notification_type == data.notification_type,
FactBilling.international == data.international
).update(
{"notifications_sent": data.notifications_sent,
"billable_units": data.billable_units},
synchronize_session=False)
if update_count == 0:
billing_record = FactBilling(
bst_date=process_day,
template_id=data.template_id,
service_id=data.service_id,
notification_type=data.notification_type,
provider=data.sent_by,
rate_multiplier=data.rate_multiplier,
international=data.international,
billable_units=data.billable_units,
notifications_sent=data.notifications_sent,
rate=get_rate(non_letter_rates,
letter_rates,
data.notification_type,
process_day,
data.crown,
data.rate_multiplier)
)
db.session.add(billing_record)
inserted_records += 1
updated_records += update_count
db.session.commit()
current_app.logger.info('ft_billing {} to {}: {} rows updated, {} rows inserted'
.format(ds, de, updated_records, inserted_records))
update_fact_billing(data, process_day)

165
app/dao/fact_billing_dao.py Normal file
View File

@@ -0,0 +1,165 @@
from datetime import datetime, timedelta, time
from flask import current_app
from sqlalchemy import func, case, desc, extract
from app import db
from app.dao.date_util import get_financial_year
from app.models import (
FactBilling,
Notification,
Service,
NOTIFICATION_CREATED,
NOTIFICATION_TECHNICAL_FAILURE,
KEY_TYPE_TEST,
LETTER_TYPE,
SMS_TYPE,
Rate,
LetterRate
)
from app.utils import convert_utc_to_bst, convert_bst_to_utc
def fetch_montly_billing_for_year(service_id, year):
year_start_date, year_end_date = get_financial_year(year)
utcnow = datetime.utcnow()
today = convert_utc_to_bst(utcnow).date()
# if year end date is less than today, we are calculating for data in the past and have no need for deltas.
if year_end_date.date() >= today:
yesterday = today - timedelta(days=1)
for day in [yesterday, today]:
data = fetch_billing_data_for_day(process_day=day, service_id=service_id)
for d in data:
update_fact_billing(data=d, process_day=day)
yearly_data = db.session.query(
extract('month', FactBilling.bst_date).label("Month"),
func.sum(FactBilling.notifications_sent).label("notifications_sent"),
func.sum(FactBilling.billable_units).label("billable_units"),
FactBilling.service_id,
FactBilling.rate,
FactBilling.rate_multiplier,
FactBilling.international
).filter(
FactBilling.service_id == service_id,
FactBilling.bst_date >= year_start_date,
FactBilling.bst_date <= year_end_date
).group_by(
'Month',
FactBilling.service_id,
FactBilling.rate,
FactBilling.rate_multiplier,
FactBilling.international
).all()
return yearly_data
def fetch_billing_data_for_day(process_day, service_id=None):
start_date = convert_bst_to_utc(datetime.combine(process_day, time.min))
end_date = convert_bst_to_utc(datetime.combine(process_day + timedelta(days=1), time.min))
transit_data = db.session.query(
Notification.template_id,
Notification.service_id,
Notification.notification_type,
func.coalesce(Notification.sent_by,
case(
[
(Notification.notification_type == 'letter', 'dvla'),
(Notification.notification_type == 'sms', 'unknown'),
(Notification.notification_type == 'email', 'ses')
]),
).label('sent_by'),
func.coalesce(Notification.rate_multiplier, 1).label('rate_multiplier'),
func.coalesce(Notification.international, False).label('international'),
func.sum(Notification.billable_units).label('billable_units'),
func.count().label('notifications_sent'),
Service.crown,
).filter(
Notification.status != NOTIFICATION_CREATED, # at created status, provider information is not available
Notification.status != NOTIFICATION_TECHNICAL_FAILURE,
Notification.key_type != KEY_TYPE_TEST,
Notification.created_at >= start_date,
Notification.created_at < end_date
).group_by(
Notification.template_id,
Notification.service_id,
Notification.notification_type,
'sent_by',
Notification.rate_multiplier,
Notification.international,
Service.crown
).join(
Service
)
if service_id:
transit_data = transit_data.filter(Notification.service_id == service_id)
return transit_data.all()
def get_rates_for_billing():
non_letter_rates = [(r.notification_type, r.valid_from, r.rate) for r in
Rate.query.order_by(desc(Rate.valid_from)).all()]
letter_rates = [(r.start_date, r.crown, r.sheet_count, r.rate) for r in
LetterRate.query.order_by(desc(LetterRate.start_date)).all()]
return non_letter_rates, letter_rates
def get_rate(non_letter_rates, letter_rates, notification_type, date, crown=None, rate_multiplier=None):
if notification_type == LETTER_TYPE:
return next(r[3] for r in letter_rates if date > r[0] and crown == r[1] and rate_multiplier == r[2])
elif notification_type == SMS_TYPE:
return next(r[2] for r in non_letter_rates if notification_type == r[0] and date > r[1])
else:
return 0
def update_fact_billing(data, process_day):
inserted_records = 0
updated_records = 0
non_letter_rates, letter_rates = get_rates_for_billing()
update_count = FactBilling.query.filter(
FactBilling.bst_date == datetime.date(process_day),
FactBilling.template_id == data.template_id,
FactBilling.service_id == data.service_id,
FactBilling.provider == data.sent_by, # This could be zero - this is a bug that needs to be fixed.
FactBilling.rate_multiplier == data.rate_multiplier,
FactBilling.notification_type == data.notification_type,
FactBilling.international == data.international
).update(
{"notifications_sent": data.notifications_sent,
"billable_units": data.billable_units},
synchronize_session=False)
if update_count == 0:
rate = get_rate(non_letter_rates,
letter_rates,
data.notification_type,
process_day,
data.crown,
data.rate_multiplier)
billing_record = create_billing_record(data, rate, process_day)
db.session.add(billing_record)
inserted_records += 1
updated_records += update_count
db.session.commit()
current_app.logger.info('ft_billing for {}: {} rows updated, {} rows inserted'
.format(process_day, updated_records, inserted_records))
def create_billing_record(data, rate, process_day):
billing_record = FactBilling(
bst_date=process_day,
template_id=data.template_id,
service_id=data.service_id,
notification_type=data.notification_type,
provider=data.sent_by,
rate_multiplier=data.rate_multiplier,
international=data.international,
billable_units=data.billable_units,
notifications_sent=data.notifications_sent,
rate=rate
)
return billing_record

19
tests/app/celery/test_reporting_tasks.py
View File

@@ -1,6 +1,7 @@
from datetime import datetime, timedelta, date
from tests.app.conftest import sample_notification
from app.celery.reporting_tasks import create_nightly_billing, get_rate
from app.celery.reporting_tasks import create_nightly_billing
from app.dao.fact_billing_dao import get_rate
from app.models import (FactBilling,
Notification,
LETTER_TYPE,
@@ -44,7 +45,7 @@ def test_create_nightly_billing_sms_rate_multiplier(
yesterday = datetime.now() - timedelta(days=1)
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
# These are sms notifications
sample_notification(
@@ -94,7 +95,7 @@ def test_create_nightly_billing_different_templates(
mocker):
yesterday = datetime.now() - timedelta(days=1)
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
sample_notification(
notify_db,
@@ -147,7 +148,7 @@ def test_create_nightly_billing_different_sent_by(
mocker):
yesterday = datetime.now() - timedelta(days=1)
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
# These are sms notifications
sample_notification(
@@ -197,7 +198,7 @@ def test_create_nightly_billing_letter(
mocker):
yesterday = datetime.now() - timedelta(days=1)
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
sample_notification(
notify_db,
@@ -234,7 +235,7 @@ def test_create_nightly_billing_null_sent_by_sms(
mocker):
yesterday = datetime.now() - timedelta(days=1)
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
sample_notification(
notify_db,
@@ -272,7 +273,7 @@ def test_create_nightly_billing_consolidate_from_3_days_delta(
sample_template,
mocker):
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
# create records from 11th to 15th
for i in range(0, 5):
@@ -365,7 +366,7 @@ def test_create_nightly_billing_use_BST(
sample_template,
mocker):
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
sample_notification(
notify_db,
@@ -414,7 +415,7 @@ def test_create_nightly_billing_update_when_record_exists(
sample_template,
mocker):
mocker.patch('app.celery.reporting_tasks.get_rate', side_effect=mocker_get_rate)
mocker.patch('app.dao.fact_billing_dao.get_rate', side_effect=mocker_get_rate)
sample_notification(
notify_db,

236
tests/app/dao/test_ft_billing_dao.py Normal file
View File

@@ -0,0 +1,236 @@
from decimal import Decimal
from datetime import datetime, timedelta
from freezegun import freeze_time
from app import db
from app.dao.fact_billing_dao import (
fetch_montly_billing_for_year, fetch_billing_data_for_day, get_rates_for_billing,
get_rate
)
from app.models import FactBilling
from app.utils import convert_utc_to_bst
from tests.app.db import (
create_ft_billing,
create_service,
create_template,
create_notification,
create_rate,
create_letter_rate
)
def test_fetch_billing_data_for_today_includes_data_with_the_right_status(notify_db_session):
service = create_service()
template = create_template(service=service, template_type="email")
for status in ['created', 'technical-failure']:
create_notification(template=template, status=status)
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert results == []
for status in ['delivered', 'sending', 'temporary-failure']:
create_notification(template=template, status=status)
results = fetch_billing_data_for_day(today)
assert len(results) == 1
assert results[0].notifications_sent == 3
def test_fetch_billing_data_for_today_includes_data_with_the_right_key_type(notify_db_session):
service = create_service()
template = create_template(service=service, template_type="email")
for key_type in ['normal', 'test', 'team']:
create_notification(template=template, status='delivered', key_type=key_type)
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert len(results) == 1
assert results[0].notifications_sent == 2
def test_fetch_billing_data_for_today_includes_data_with_the_right_date(notify_db_session):
process_day = datetime(2018, 4, 1, 13, 30, 00)
service = create_service()
template = create_template(service=service, template_type="email")
create_notification(template=template, status='delivered', created_at=process_day)
create_notification(template=template, status='delivered', created_at=datetime(2018, 3, 31, 23, 23, 23))
create_notification(template=template, status='delivered', created_at=datetime(2018, 3, 31, 20, 23, 23))
create_notification(template=template, status='sending', created_at=process_day + timedelta(days=1))
day_under_test = convert_utc_to_bst(process_day)
results = fetch_billing_data_for_day(day_under_test)
assert len(results) == 1
assert results[0].notifications_sent == 2
def test_fetch_billing_data_for_day_is_grouped_by_template_and_notification_type(notify_db_session):
service = create_service()
email_template = create_template(service=service, template_type="email")
sms_template = create_template(service=service, template_type="sms")
create_notification(template=email_template, status='delivered')
create_notification(template=sms_template, status='delivered')
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert len(results) == 2
assert results[0].notifications_sent == 1
assert results[1].notifications_sent == 1
def test_fetch_billing_data_for_day_is_grouped_by_service(notify_db_session):
service_1 = create_service()
service_2 = create_service(service_name='Service 2')
email_template = create_template(service=service_1)
sms_template = create_template(service=service_2)
create_notification(template=email_template, status='delivered')
create_notification(template=sms_template, status='delivered')
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert len(results) == 2
assert results[0].notifications_sent == 1
assert results[1].notifications_sent == 1
def test_fetch_billing_data_for_day_is_grouped_by_provider(notify_db_session):
service = create_service()
template = create_template(service=service)
create_notification(template=template, status='delivered', sent_by='mmg')
create_notification(template=template, status='delivered', sent_by='firetext')
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert len(results) == 2
assert results[0].notifications_sent == 1
assert results[1].notifications_sent == 1
def test_fetch_billing_data_for_day_is_grouped_by_rate_mulitplier(notify_db_session):
service = create_service()
template = create_template(service=service)
create_notification(template=template, status='delivered', rate_multiplier=1)
create_notification(template=template, status='delivered', rate_multiplier=2)
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert len(results) == 2
assert results[0].notifications_sent == 1
assert results[1].notifications_sent == 1
def test_fetch_billing_data_for_day_is_grouped_by_international(notify_db_session):
service = create_service()
template = create_template(service=service)
create_notification(template=template, status='delivered', international=True)
create_notification(template=template, status='delivered', international=False)
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert len(results) == 2
assert results[0].notifications_sent == 1
assert results[1].notifications_sent == 1
def test_fetch_billing_data_for_day_returns_empty_list(notify_db_session):
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(today)
assert results == []
def test_fetch_billing_data_for_day_returns_list_for_given_service(notify_db_session):
service = create_service()
service_2 = create_service(service_name='Service 2')
template = create_template(service=service)
template_2 = create_template(service=service_2)
create_notification(template=template, status='delivered')
create_notification(template=template_2, status='delivered')
today = convert_utc_to_bst(datetime.utcnow())
results = fetch_billing_data_for_day(process_day=today, service_id=service.id)
assert len(results) == 1
assert results[0].service_id == service.id
def test_get_rates_for_billing(notify_db_session):
create_rate(start_date=datetime.utcnow(), value=12, notification_type='email')
create_rate(start_date=datetime.utcnow(), value=22, notification_type='sms')
create_rate(start_date=datetime.utcnow(), value=33, notification_type='email')
create_letter_rate(start_date=datetime.utcnow())
non_letter_rates, letter_rates = get_rates_for_billing()
assert len(non_letter_rates) == 3
assert len(letter_rates) == 1
def test_get_rate(notify_db_session):
create_rate(start_date=datetime.utcnow(), value=1.2, notification_type='email')
create_rate(start_date=datetime.utcnow(), value=2.2, notification_type='sms')
create_rate(start_date=datetime.utcnow(), value=3.3, notification_type='email')
create_letter_rate(start_date=datetime.utcnow(), rate=4.4)
non_letter_rates, letter_rates = get_rates_for_billing()
rate = get_rate(non_letter_rates=non_letter_rates, letter_rates=letter_rates, notification_type='sms',
date=datetime.utcnow())
letter_rate = get_rate(non_letter_rates=non_letter_rates, letter_rates=letter_rates,
notification_type='letter',
crown=True,
rate_multiplier=1,
date=datetime.utcnow())
assert rate == 2.2
assert letter_rate == Decimal('4.4')
def test_fetch_annual_billing_for_year(notify_db_session):
service = create_service()
template = create_template(service=service, template_type="sms")
for i in range(1, 31):
create_ft_billing(bst_date='2018-06-{}'.format(i),
service=service,
template=template,
notification_type='sms',
rate=0.162)
for i in range(1, 32):
create_ft_billing(bst_date='2018-07-{}'.format(i),
service=service,
template=template,
notification_type='sms',
rate=0.158)
results = fetch_montly_billing_for_year(service_id=service.id, year=2018)
assert len(results) == 2
assert results[0][0] == 6.0
assert results[0][1] == 30
assert results[0][2] == Decimal('30')
assert results[0][3] == service.id
assert results[0][4] == Decimal('0.162')
assert results[0][5] == Decimal('1')
assert results[0][6] is False
assert results[1][0] == 7.0
assert results[1][1] == 31
assert results[1][2] == Decimal('31')
assert results[1][3] == service.id
assert results[1][4] == Decimal('0.158')
assert results[1][5] == Decimal('1')
assert results[1][6] is False
@freeze_time('2018-08-01 13:30:00')
def test_fetch_annual_billing_for_year_adds_data_for_today(notify_db_session):
service = create_service()
template = create_template(service=service, template_type="email")
for i in range(1, 32):
create_ft_billing(bst_date='2018-07-{}'.format(i),
service=service,
template=template,
notification_type='email',
rate=0.162)
create_notification(template=template, status='delivered')
assert db.session.query(FactBilling.bst_date).count() == 31
results = fetch_montly_billing_for_year(service_id=service.id,
year=2018)
assert db.session.query(FactBilling.bst_date).count() == 32
assert len(results) == 2

208
tests/app/db.py
View File

@@ -34,6 +34,7 @@ from app.models import (
AnnualBilling,
LetterRate,
InvitedOrganisationUser,
FactBilling
)
from app.dao.users_dao import save_model_user
from app.dao.notifications_dao import (
@@ -65,17 +66,17 @@ def create_user(mobile_number="+447700900986", email="notify@digital.cabinet-off
def create_service(
user=None,
service_name="Sample service",
service_id=None,
restricted=False,
service_permissions=[EMAIL_TYPE, SMS_TYPE],
research_mode=False,
active=True,
email_from=None,
prefix_sms=True,
message_limit=1000,
organisation_type='central'
user=None,
service_name="Sample service",
service_id=None,
restricted=False,
service_permissions=[EMAIL_TYPE, SMS_TYPE],
research_mode=False,
active=True,
email_from=None,
prefix_sms=True,
message_limit=1000,
organisation_type='central'
):
service = Service(
name=service_name,
@@ -96,8 +97,8 @@ def create_service(
def create_service_with_inbound_number(
inbound_number='1234567',
*args, **kwargs
inbound_number='1234567',
*args, **kwargs
):
service = create_service(*args, **kwargs)
@@ -111,8 +112,8 @@ def create_service_with_inbound_number(
def create_service_with_defined_sms_sender(
sms_sender_value='1234567',
*args, **kwargs
sms_sender_value='1234567',
*args, **kwargs
):
service = create_service(*args, **kwargs)
@@ -126,13 +127,13 @@ def create_service_with_defined_sms_sender(
def create_template(
service,
template_type=SMS_TYPE,
template_name=None,
subject='Template subject',
content='Dear Sir/Madam, Hello. Yours Truly, The Government.',
reply_to=None,
hidden=False
service,
template_type=SMS_TYPE,
template_name=None,
subject='Template subject',
content='Dear Sir/Madam, Hello. Yours Truly, The Government.',
reply_to=None,
hidden=False
):
data = {
'name': template_name or '{} Template Name'.format(template_type),
@@ -151,29 +152,29 @@ def create_template(
def create_notification(
template,
job=None,
job_row_number=None,
to_field=None,
status='created',
reference=None,
created_at=None,
sent_at=None,
updated_at=None,
billable_units=1,
personalisation=None,
api_key=None,
key_type=KEY_TYPE_NORMAL,
sent_by=None,
client_reference=None,
rate_multiplier=None,
international=False,
phone_prefix=None,
scheduled_for=None,
normalised_to=None,
one_off=False,
sms_sender_id=None,
reply_to_text=None
template,
job=None,
job_row_number=None,
to_field=None,
status='created',
reference=None,
created_at=None,
sent_at=None,
updated_at=None,
billable_units=1,
personalisation=None,
api_key=None,
key_type=KEY_TYPE_NORMAL,
sent_by=None,
client_reference=None,
rate_multiplier=None,
international=False,
phone_prefix=None,
scheduled_for=None,
normalised_to=None,
one_off=False,
sms_sender_id=None,
reply_to_text=None
):
if created_at is None:
created_at = datetime.utcnow()
@@ -235,13 +236,13 @@ def create_notification(
def create_job(
template,
notification_count=1,
created_at=None,
job_status='pending',
scheduled_for=None,
processing_started=None,
original_file_name='some.csv'
template,
notification_count=1,
created_at=None,
job_status='pending',
scheduled_for=None,
processing_started=None,
original_file_name='some.csv'
):
data = {
'id': uuid.uuid4(),
@@ -272,14 +273,14 @@ def create_service_permission(service_id, permission=EMAIL_TYPE):
def create_inbound_sms(
service,
notify_number=None,
user_number='447700900111',
provider_date=None,
provider_reference=None,
content='Hello',
provider="mmg",
created_at=None
service,
notify_number=None,
user_number='447700900111',
provider_date=None,
provider_reference=None,
content='Hello',
provider="mmg",
created_at=None
):
inbound = InboundSms(
service=service,
@@ -296,9 +297,9 @@ def create_inbound_sms(
def create_service_inbound_api(
service,
url="https://something.com",
bearer_token="some_super_secret",
service,
url="https://something.com",
bearer_token="some_super_secret",
):
service_inbound_api = ServiceInboundApi(service_id=service.id,
url=url,
@@ -310,9 +311,9 @@ def create_service_inbound_api(
def create_service_callback_api(
service,
url="https://something.com",
bearer_token="some_super_secret",
service,
url="https://something.com",
bearer_token="some_super_secret",
):
service_callback_api = ServiceCallbackApi(service_id=service.id,
url=url,
@@ -376,11 +377,11 @@ def create_inbound_number(number, provider='mmg', active=True, service_id=None):
def create_monthly_billing_entry(
service,
start_date,
end_date,
notification_type,
monthly_totals=[]
service,
start_date,
end_date,
notification_type,
monthly_totals=[]
):
entry = MonthlyBilling(
service_id=service.id,
@@ -397,9 +398,9 @@ def create_monthly_billing_entry(
def create_reply_to_email(
service,
email_address,
is_default=True
service,
email_address,
is_default=True
):
data = {
'service': service,
@@ -415,10 +416,10 @@ def create_reply_to_email(
def create_service_sms_sender(
service,
sms_sender,
is_default=True,
inbound_number_id=None
service,
sms_sender,
is_default=True,
inbound_number_id=None
):
data = {
'service_id': service.id,
@@ -435,9 +436,9 @@ def create_service_sms_sender(
def create_letter_contact(
service,
contact_block,
is_default=True
service,
contact_block,
is_default=True
):
data = {
'service': service,
@@ -453,7 +454,7 @@ def create_letter_contact(
def create_annual_billing(
service_id, free_sms_fragment_limit, financial_year_start
service_id, free_sms_fragment_limit, financial_year_start
):
annual_billing = AnnualBilling(
service_id=service_id,
@@ -467,12 +468,12 @@ def create_annual_billing(
def create_letter_rate(
start_date=datetime(2017, 1, 1, 00, 00, 00),
end_date=None,
sheet_count=1,
rate=0.31,
crown=True,
post_class='second'
start_date=datetime(2017, 1, 1, 00, 00, 00),
end_date=None,
sheet_count=1,
rate=0.31,
crown=True,
post_class='second'
):
rate = LetterRate(
start_date=start_date,
@@ -524,3 +525,34 @@ def create_daily_sorted_letter(billing_day=date(2018, 1, 18),
db.session.commit()
return daily_sorted_letter
def create_ft_billing(bst_date,
notification_type,
template=None,
service=None,
provider='test',
rate_multiplier=1,
international=False,
rate=0,
billable_unit=1,
notifications_sent=1
):
if not service:
service = create_service()
if not template:
template = create_template(service=service, template_type=notification_type)
data = FactBilling(bst_date=bst_date,
service_id=service.id,
template_id=template.id,
notification_type=notification_type,
provider=provider,
rate_multiplier=rate_multiplier,
international=international,
rate=rate,
billable_units=billable_unit,
notifications_sent=notifications_sent)
db.session.add(data)
db.session.commit()
return data