remove usage of notify_db fixture in unit tests

* notify_db fixture creates the database connection and ensures the test
  db exists and has migrations applied etc. It will run once per session
  (test run).
* notify_db_session fixture runs after your test finishes and deletes
  all non static (eg type table) data.

In unit tests that hit the database (ie: most of them), 99% of the time
we will need to use notify_db_session to ensure everything is reset. The
only time we don't need to use it is when we're querying things such as
"ensure get X works when database is empty". This is such a low
percentage of tests that it's easier for us to just use
notify_db_session every time, and ensure that all our tests run much
more consistently, at the cost of a small bit of performance when
running tests.

We used to use notify_db to access the session object for manually
adding, committing, etc. To dissuade usage of that fixture I've moved
that to the `notify_db_session`. I've then removed all uses of notify_db
that I could find in the codebase.

As a note, if you're writing a test that uses a `sample_x` fixture, all
of those fixtures rely on notify_db_session so you'll get the teardown
functionality for free. If you're just calling eg `create_x` db.py
functions, then you'll need to make you add notify_db_session fixture to
your test, even if you aren't manually accessing the session.
This commit is contained in:
Leo Hemsted
2022-05-03 17:00:51 +01:00
parent 867e8fbce3
commit 6181c60f75
33 changed files with 151 additions and 162 deletions

View File

@@ -401,7 +401,7 @@ def test_save_notification_and_increment_job(sample_template, sample_job, mmg_pr
assert Notification.query.count() == 2
def test_save_notification_and_increment_correct_job(notify_db, notify_db_session, sample_template, mmg_provider):
def test_save_notification_and_increment_correct_job(sample_template, mmg_provider):
job_1 = create_job(sample_template)
job_2 = create_job(sample_template)
@@ -459,7 +459,7 @@ def test_get_notification_by_id_when_notification_exists(sample_notification):
assert sample_notification == notification_from_db
def test_get_notification_by_id_when_notification_does_not_exist(notify_db, fake_uuid):
def test_get_notification_by_id_when_notification_does_not_exist(notify_db_session, fake_uuid):
notification_from_db = get_notification_by_id(fake_uuid)
assert notification_from_db is None
@@ -738,7 +738,6 @@ def test_should_not_count_pages_when_given_a_flag(sample_user, sample_template):
def test_get_notifications_created_by_api_or_csv_are_returned_correctly_excluding_test_key_notifications(
notify_db,
notify_db_session,
sample_service,
sample_job,
@@ -1529,7 +1528,7 @@ def test_dao_update_notifications_by_reference_updates_history_no_notifications_
assert updated_history_count == 2
def test_dao_update_notifications_by_reference_returns_zero_when_no_notifications_to_update(notify_db):
def test_dao_update_notifications_by_reference_returns_zero_when_no_notifications_to_update(notify_db_session):
updated_count, updated_history_count = dao_update_notifications_by_reference(
references=['ref'],
update_dict={
@@ -1574,14 +1573,14 @@ def test_dao_update_notifications_by_reference_updates_history_when_one_of_two_n
assert NotificationHistory.query.get(notification1.id).status == 'returned-letter'
def test_dao_get_notification_by_reference_with_one_match_returns_notification(sample_letter_template, notify_db):
def test_dao_get_notification_by_reference_with_one_match_returns_notification(sample_letter_template):
create_notification(template=sample_letter_template, reference='REF1')
notification = dao_get_notification_by_reference('REF1')
assert notification.reference == 'REF1'
def test_dao_get_notification_by_reference_with_multiple_matches_raises_error(sample_letter_template, notify_db):
def test_dao_get_notification_by_reference_with_multiple_matches_raises_error(sample_letter_template):
create_notification(template=sample_letter_template, reference='REF1')
create_notification(template=sample_letter_template, reference='REF1')
@@ -1589,7 +1588,7 @@ def test_dao_get_notification_by_reference_with_multiple_matches_raises_error(sa
dao_get_notification_by_reference('REF1')
def test_dao_get_notification_by_reference_with_no_matches_raises_error(notify_db):
def test_dao_get_notification_by_reference_with_no_matches_raises_error(notify_db_session):
with pytest.raises(SQLAlchemyError):
dao_get_notification_by_reference('REF1')
@@ -1613,7 +1612,7 @@ def test_dao_get_notification_or_history_by_reference_with_multiple_matches_rais
dao_get_notification_or_history_by_reference('REF1')
def test_dao_get_notification_or_history_by_reference_with_no_matches_raises_error(notify_db):
def test_dao_get_notification_or_history_by_reference_with_no_matches_raises_error(notify_db_session):
with pytest.raises(SQLAlchemyError):
dao_get_notification_or_history_by_reference('REF1')

View File

@@ -56,7 +56,7 @@ def test_get_api_key_should_raise_exception_when_api_key_does_not_exist(sample_s
get_model_api_keys(sample_service.id, id=fake_uuid)
def test_should_return_api_key_for_service(notify_api, notify_db, notify_db_session, sample_api_key):
def test_should_return_api_key_for_service(notify_api, notify_db_session, sample_api_key):
api_key = get_model_api_keys(service_id=sample_api_key.service_id, id=sample_api_key.id)
assert api_key == sample_api_key

View File

@@ -120,7 +120,7 @@ def test_fetch_count_of_complaints(sample_email_notification):
assert count_of_complaints == 3
def test_fetch_count_of_complaints_returns_zero(notify_db):
def test_fetch_count_of_complaints_returns_zero(notify_db_session):
count_of_complaints = fetch_count_of_complaints(start_date=datetime(2018, 6, 7),
end_date=datetime(2018, 6, 7))
assert count_of_complaints == 0

View File

@@ -8,7 +8,7 @@ from app.models import DailySortedLetter
from tests.app.db import create_daily_sorted_letter
def test_dao_get_daily_sorted_letter_by_billing_day(notify_db, notify_db_session):
def test_dao_get_daily_sorted_letter_by_billing_day(notify_db_session):
billing_day = date(2018, 2, 1)
other_day = date(2017, 9, 8)
@@ -18,7 +18,7 @@ def test_dao_get_daily_sorted_letter_by_billing_day(notify_db, notify_db_session
assert not dao_get_daily_sorted_letter_by_billing_day(other_day)
def test_dao_create_or_update_daily_sorted_letter_creates_a_new_entry(notify_db, notify_db_session):
def test_dao_create_or_update_daily_sorted_letter_creates_a_new_entry(notify_db_session):
billing_day = date(2018, 2, 1)
dsl = DailySortedLetter(billing_day=billing_day,
file_name="Notify-201802011234.rs.txt",
@@ -35,7 +35,6 @@ def test_dao_create_or_update_daily_sorted_letter_creates_a_new_entry(notify_db,
def test_dao_create_or_update_daily_sorted_letter_updates_an_existing_entry(
notify_db,
notify_db_session
):
create_daily_sorted_letter(billing_day=date(2018, 1, 18),

View File

@@ -8,7 +8,7 @@ from app.models import EmailBranding
from tests.app.db import create_email_branding
def test_get_email_branding_options_gets_all_email_branding(notify_db, notify_db_session):
def test_get_email_branding_options_gets_all_email_branding(notify_db_session):
email_branding_1 = create_email_branding(name='test_email_branding_1')
email_branding_2 = create_email_branding(name='test_email_branding_2')
@@ -19,7 +19,7 @@ def test_get_email_branding_options_gets_all_email_branding(notify_db, notify_db
assert email_branding_2 == email_branding[1]
def test_get_email_branding_by_id_gets_correct_email_branding(notify_db, notify_db_session):
def test_get_email_branding_by_id_gets_correct_email_branding(notify_db_session):
email_branding = create_email_branding()
email_branding_from_db = dao_get_email_branding_by_id(email_branding.id)
@@ -27,7 +27,7 @@ def test_get_email_branding_by_id_gets_correct_email_branding(notify_db, notify_
assert email_branding_from_db == email_branding
def test_get_email_branding_by_name_gets_correct_email_branding(notify_db, notify_db_session):
def test_get_email_branding_by_name_gets_correct_email_branding(notify_db_session):
email_branding = create_email_branding(name="Crystal Gems")
email_branding_from_db = dao_get_email_branding_by_name("Crystal Gems")
@@ -35,7 +35,7 @@ def test_get_email_branding_by_name_gets_correct_email_branding(notify_db, notif
assert email_branding_from_db == email_branding
def test_update_email_branding(notify_db, notify_db_session):
def test_update_email_branding(notify_db_session):
updated_name = 'new name'
create_email_branding()
@@ -52,7 +52,7 @@ def test_update_email_branding(notify_db, notify_db_session):
assert email_branding[0].name == updated_name
def test_email_branding_has_no_domain(notify_db, notify_db_session):
def test_email_branding_has_no_domain(notify_db_session):
create_email_branding()
email_branding = EmailBranding.query.all()
assert not hasattr(email_branding, 'domain')

View File

@@ -3,7 +3,7 @@ from app.dao.events_dao import dao_create_event
from app.models import Event
def test_create_event(notify_db, notify_db_session):
def test_create_event(notify_db_session):
assert Event.query.count() == 0
data = {
'event_type': 'sucessful_login',

View File

@@ -13,14 +13,14 @@ from app.models import InboundNumber
from tests.app.db import create_inbound_number, create_service
def test_get_inbound_numbers(notify_db, notify_db_session, sample_inbound_numbers):
def test_get_inbound_numbers(notify_db_session, sample_inbound_numbers):
res = dao_get_inbound_numbers()
assert len(res) == len(sample_inbound_numbers)
assert res == sample_inbound_numbers
def test_get_available_inbound_numbers(notify_db, notify_db_session):
def test_get_available_inbound_numbers(notify_db_session):
inbound_number = create_inbound_number(number='1')
res = dao_get_available_inbound_numbers()
@@ -29,7 +29,7 @@ def test_get_available_inbound_numbers(notify_db, notify_db_session):
assert res[0] == inbound_number
def test_set_service_id_on_inbound_number(notify_db, notify_db_session, sample_inbound_numbers):
def test_set_service_id_on_inbound_number(notify_db_session, sample_inbound_numbers):
service = create_service(service_name='test service')
numbers = dao_get_available_inbound_numbers()
@@ -42,7 +42,7 @@ def test_set_service_id_on_inbound_number(notify_db, notify_db_session, sample_i
def test_after_setting_service_id_that_inbound_number_is_unavailable(
notify_db, notify_db_session, sample_inbound_numbers):
notify_db_session, sample_inbound_numbers):
service = create_service(service_name='test service')
numbers = dao_get_available_inbound_numbers()
@@ -55,7 +55,7 @@ def test_after_setting_service_id_that_inbound_number_is_unavailable(
assert len(res) == 0
def test_setting_a_service_twice_will_raise_an_error(notify_db, notify_db_session):
def test_setting_a_service_twice_will_raise_an_error(notify_db_session):
create_inbound_number(number='1')
create_inbound_number(number='2')
service = create_service(service_name='test service')
@@ -70,7 +70,7 @@ def test_setting_a_service_twice_will_raise_an_error(notify_db, notify_db_sessio
@pytest.mark.parametrize("active", [True, False])
def test_set_inbound_number_active_flag(notify_db, notify_db_session, sample_service, active):
def test_set_inbound_number_active_flag(notify_db_session, sample_service, active):
inbound_number = create_inbound_number(number='1')
dao_set_inbound_number_to_service(sample_service.id, inbound_number)

View File

@@ -16,7 +16,7 @@ from app.models import InvitedUser
from tests.app.db import create_invited_user
def test_create_invited_user(notify_db, notify_db_session, sample_service):
def test_create_invited_user(notify_db_session, sample_service):
assert InvitedUser.query.count() == 0
email_address = 'invited_user@service.gov.uk'
invite_from = sample_service.users[0]
@@ -43,8 +43,6 @@ def test_create_invited_user(notify_db, notify_db_session, sample_service):
def test_create_invited_user_sets_default_folder_permissions_of_empty_list(
notify_db,
notify_db_session,
sample_service,
):
assert InvitedUser.query.count() == 0
@@ -64,17 +62,17 @@ def test_create_invited_user_sets_default_folder_permissions_of_empty_list(
assert invited_user.folder_permissions == []
def test_get_invited_user_by_service_and_id(notify_db, notify_db_session, sample_invited_user):
def test_get_invited_user_by_service_and_id(notify_db_session, sample_invited_user):
from_db = get_invited_user_by_service_and_id(sample_invited_user.service.id, sample_invited_user.id)
assert from_db == sample_invited_user
def test_get_invited_user_by_id(notify_db, notify_db_session, sample_invited_user):
def test_get_invited_user_by_id(notify_db_session, sample_invited_user):
from_db = get_invited_user_by_id(sample_invited_user.id)
assert from_db == sample_invited_user
def test_get_unknown_invited_user_returns_none(notify_db, notify_db_session, sample_service):
def test_get_unknown_invited_user_returns_none(notify_db_session, sample_service):
unknown_id = uuid.uuid4()
with pytest.raises(NoResultFound) as e:
@@ -82,7 +80,7 @@ def test_get_unknown_invited_user_returns_none(notify_db, notify_db_session, sam
assert 'No row was found when one was required' in str(e.value)
def test_get_invited_users_for_service(notify_db, notify_db_session, sample_service):
def test_get_invited_users_for_service(notify_db_session, sample_service):
invites = []
for i in range(0, 5):
email = 'invited_user_{}@service.gov.uk'.format(i)
@@ -96,12 +94,12 @@ def test_get_invited_users_for_service(notify_db, notify_db_session, sample_serv
assert invite in all_from_db
def test_get_invited_users_for_service_that_has_no_invites(notify_db, notify_db_session, sample_service):
def test_get_invited_users_for_service_that_has_no_invites(notify_db_session, sample_service):
invites = get_invited_users_for_service(sample_service.id)
assert len(invites) == 0
def test_save_invited_user_sets_status_to_cancelled(notify_db, notify_db_session, sample_invited_user):
def test_save_invited_user_sets_status_to_cancelled(notify_db_session, sample_invited_user):
assert InvitedUser.query.count() == 1
saved = InvitedUser.query.get(sample_invited_user.id)
assert saved.status == 'pending'

View File

@@ -154,7 +154,7 @@ def test_get_jobs_for_service_with_limit_days_edge_case(sample_template):
assert just_before_midnight_job not in jobs_limit_days
def test_get_jobs_for_service_in_processed_at_then_created_at_order(notify_db, notify_db_session, sample_template):
def test_get_jobs_for_service_in_processed_at_then_created_at_order(notify_db_session, sample_template):
from_hour = partial(datetime, 2001, 1, 1)
created_jobs = [
@@ -269,7 +269,7 @@ def test_should_get_jobs_seven_days_old(sample_template):
assert jobs[0].id == job_to_delete.id
def test_get_jobs_for_service_is_paginated(notify_db, notify_db_session, sample_service, sample_template):
def test_get_jobs_for_service_is_paginated(notify_db_session, sample_service, sample_template):
with freeze_time('2015-01-01T00:00:00') as the_time:
for _ in range(10):
the_time.tick(timedelta(hours=1))

View File

@@ -38,7 +38,7 @@ def test_dao_get_all_letter_branding(notify_db_session):
assert len(results) == 2
def test_dao_get_all_letter_branding_returns_empty_list_if_no_brands_exist(notify_db):
def test_dao_get_all_letter_branding_returns_empty_list_if_no_brands_exist(notify_db_session):
assert dao_get_all_letter_branding() == []

View File

@@ -45,7 +45,7 @@ def test_save_service_callback_api(sample_service):
assert versioned.version == 1
def test_save_service_callback_api_fails_if_service_does_not_exist(notify_db, notify_db_session):
def test_save_service_callback_api_fails_if_service_does_not_exist(notify_db_session):
service_callback_api = ServiceCallbackApi(
service_id=uuid.uuid4(),
url="https://some_service/callback_endpoint",

View File

@@ -29,7 +29,7 @@ def test_add_and_commit_guest_list_contacts_saves_data(sample_service):
assert db_contents[0].id == guest_list.id
def test_remove_service_guest_list_only_removes_for_my_service(notify_db, notify_db_session):
def test_remove_service_guest_list_only_removes_for_my_service(notify_db_session):
service_1 = create_service(service_name="service 1")
service_2 = create_service(service_name="service 2")
dao_add_and_commit_guest_list_contacts([
@@ -43,10 +43,10 @@ def test_remove_service_guest_list_only_removes_for_my_service(notify_db, notify
assert len(service_2.guest_list) == 1
def test_remove_service_guest_list_does_not_commit(notify_db, sample_service_guest_list):
def test_remove_service_guest_list_does_not_commit(notify_db_session, sample_service_guest_list):
dao_remove_service_guest_list(sample_service_guest_list.service_id)
# since dao_remove_service_guest_list doesn't commit, we can still rollback its changes
notify_db.session.rollback()
notify_db_session.rollback()
assert ServiceGuestList.query.count() == 1

View File

@@ -45,7 +45,7 @@ def test_save_service_inbound_api(sample_service):
assert versioned.version == 1
def test_save_service_inbound_api_fails_if_service_does_not_exist(notify_db, notify_db_session):
def test_save_service_inbound_api_fails_if_service_does_not_exist(notify_db_session):
service_inbound_api = ServiceInboundApi(
service_id=uuid.uuid4(),
url="https://some_service/inbound_messages",

View File

@@ -15,7 +15,7 @@ from tests.app.db import create_service, create_service_permission
@pytest.fixture(scope='function')
def service_without_permissions(notify_db, notify_db_session):
def service_without_permissions(notify_db_session):
return create_service(service_permissions=[])

View File

@@ -505,7 +505,7 @@ def test_dao_fetch_live_services_data(sample_user):
]
def test_get_service_by_id_returns_none_if_no_service(notify_db):
def test_get_service_by_id_returns_none_if_no_service(notify_db_session):
with pytest.raises(NoResultFound) as e:
dao_fetch_service_by_id(str(uuid.uuid4()))
assert 'No row was found when one was required' in str(e.value)
@@ -1021,7 +1021,7 @@ def test_dao_fetch_todays_stats_for_all_services_only_includes_today(notify_db_s
assert stats['failed'] == 1
def test_dao_fetch_todays_stats_for_all_services_groups_correctly(notify_db, notify_db_session):
def test_dao_fetch_todays_stats_for_all_services_groups_correctly(notify_db_session):
service1 = create_service(service_name='service 1', email_from='service.1')
service2 = create_service(service_name='service 2', email_from='service.2')
template_sms = create_template(service=service1)

View File

@@ -159,7 +159,7 @@ def test_update_user_attribute(client, sample_user, user_attribute, user_value):
@freeze_time('2020-01-24T12:00:00')
def test_update_user_password(notify_api, notify_db, notify_db_session, sample_user):
def test_update_user_password(notify_api, notify_db_session, sample_user):
sample_user.password_changed_at = datetime.utcnow() - timedelta(days=1)
password = 'newpassword'
assert not sample_user.check_password(password)