more

tests/app/dao/test_services_dao.py

@@ -28,6 +28,7 @@ from app.dao.services_dao import (
     dao_fetch_all_services_by_user,
     dao_fetch_live_services_data,
     dao_fetch_service_by_id,
+    dao_fetch_service_by_id_with_api_keys,
     dao_fetch_service_by_inbound_number,
     dao_fetch_todays_stats_for_all_services,
     dao_fetch_todays_stats_for_service,
@@ -133,6 +134,39 @@ def test_create_service_with_organization(notify_db_session):
     assert service.organization == organization
 
 
+def test_fetch_service_by_id_with_api_keys(notify_db_session):
+    user = create_user(email='local.authority@local-authority.gov.uk')
+    organization = create_organization(
+        name='Some local authority', organization_type='state', domains=['local-authority.gov.uk'])
+    assert Service.query.count() == 0
+    service = Service(name="service_name",
+                      email_from="email_from",
+                      message_limit=1000,
+                      restricted=False,
+                      organization_type='federal',
+                      created_by=user)
+    dao_create_service(service, user)
+    assert Service.query.count() == 1
+    service_db = Service.query.one()
+    organization = Organization.query.get(organization.id)
+    assert service_db.name == "service_name"
+    assert service_db.id == service.id
+    assert service_db.email_from == 'email_from'
+    assert service_db.research_mode is False
+    assert service_db.prefix_sms is True
+    assert service.active is True
+    assert user in service_db.users
+    assert service_db.organization_type == 'state'
+    assert service.organization_id == organization.id
+    assert service.organization == organization
+
+    service = dao_fetch_service_by_id_with_api_keys(service.id, False)
+    assert service is not None
+    assert service.api_keys is not None
+    service = dao_fetch_service_by_id_with_api_keys(service.id, True)
+    assert service is not None
+
+
 def test_cannot_create_two_services_with_same_name(notify_db_session):
     user = create_user()
     assert Service.query.count() == 0

tests/app/test_commands.py

@@ -1,3 +1,6 @@
+import datetime
+import os
+
 import pytest
 
 from app.commands import (
@@ -5,10 +8,171 @@ from app.commands import (
     create_test_user,
     insert_inbound_numbers_from_file,
     populate_annual_billing_with_defaults,
+    populate_annual_billing_with_the_previous_years_allowance,
+    populate_organization_agreement_details_from_file,
+    populate_organizations_from_file,
+    purge_functional_test_data,
+    update_jobs_archived_flag,
 )
 from app.dao.inbound_numbers_dao import dao_get_available_inbound_numbers
-from app.models import AnnualBilling, Template, User
-from tests.app.db import create_annual_billing, create_service
+from app.models import AnnualBilling, Job, Organization, Service, Template, User
+from tests.app.db import (
+    create_annual_billing,
+    create_job,
+    create_organization,
+    create_service,
+    create_template,
+)
+
+
+def test_purge_functional_test_data(notify_db_session, notify_api):
+    print("ENTER test_purge_functional_test_data")
+
+    user_count = User.query.count()
+    print(f"INITIAL user count {user_count}")
+    # run the command
+    notify_api.test_cli_runner().invoke(
+        create_test_user, [
+            '--email', 'somebody+7af2cdb0-7cbc-44dc-a5d0-f817fc6ee94e@fake.gov',
+            '--mobile_number', '202-555-5555',
+            '--password', 'correct horse battery staple',
+            '--name', 'Fake Personson',
+            # '--auth_type', 'sms_auth',  # this is the default
+            # '--state', 'active',  # this is the default
+            # '--admin', 'False',  # this is the default
+        ]
+    )
+
+    user_count = User.query.count()
+    print(f"AFTER create_test_user {user_count}")
+    assert user_count == 1
+
+    users = User.query.all()
+    for user in users:
+        print(f"USER: {user.email_address} {user.id} {user.name}")
+
+    # run the command
+    print("INVOKING purge_functional_test_data")
+    notify_api.test_cli_runner().invoke(purge_functional_test_data, ['-u', 'somebody'])
+    print("IT WAS INVOKED")
+    # there should be one more user
+    assert User.query.count() == 0
+
+
+def test_purge_functional_test_data_bad_mobile(notify_db_session, notify_api):
+
+    user_count = User.query.count()
+    assert user_count == 0
+    # run the command
+    x = notify_api.test_cli_runner().invoke(
+        create_test_user, [
+            '--email', 'somebody+7af2cdb0-7cbc-44dc-a5d0-f817fc6ee94e@fake.gov',
+            '--mobile_number', '555-555-55554444',
+            '--password', 'correct horse battery staple',
+            '--name', 'Fake Personson',
+            # '--auth_type', 'sms_auth',  # this is the default
+            # '--state', 'active',  # this is the default
+            # '--admin', 'False',  # this is the default
+        ]
+    )
+    print(f"X = {x}")
+    # The bad mobile phone number results in a bad parameter error, leading to a system exit 2 and no entry made in db
+    assert "SystemExit(2)" in str(x)
+    user_count = User.query.count()
+    assert user_count == 0
+
+
+def test_update_jobs_archived_flag(notify_db_session, notify_api):
+    print("ENTER test_update_jobs_archived_flag")
+
+    service_count = Service.query.count()
+    assert service_count == 0
+
+    service = create_service()
+    service_count = Service.query.count()
+    assert service_count == 1
+
+    sms_template = create_template(service=service, template_type='sms')
+    create_job(sms_template)
+
+    # run the command
+    one_hour_past = datetime.datetime.utcnow()
+    one_hour_future = datetime.datetime.utcnow() + datetime.timedelta(days=1)
+
+    one_hour_past = one_hour_past.strftime("%Y-%m-%d")
+    one_hour_future = one_hour_future.strftime("%Y-%m-%d")
+    print(f"PAST {one_hour_past} FUTURE = {one_hour_future}")
+
+    archived_jobs = Job.query.filter(Job.archived is True).count()
+    assert archived_jobs == 0
+
+    x = notify_api.test_cli_runner().invoke(
+        update_jobs_archived_flag, [
+            '-e', one_hour_future,
+            '-s', one_hour_past,
+        ]
+    )
+    print(f"X = {x}")
+    jobs = Job.query.all()
+    assert len(jobs) == 1
+    for job in jobs:
+        assert job.archived is True
+
+
+def test_populate_organizations_from_file(notify_db_session, notify_api):
+
+    org_count = Organization.query.count()
+    assert org_count == 0
+
+    file_name = "./tests/app/orgs1.csv"
+    text = "name|blah|blah|blah|||\n" \
+           "foo|Federal|True|'foo.gov'|||\n"
+    f = open(file_name, "a")
+    f.write(text)
+    f.close()
+    x = notify_api.test_cli_runner().invoke(
+        populate_organizations_from_file, [
+            '-f', file_name
+        ]
+    )
+
+    os.remove(file_name)
+    print(f"X = {x}")
+
+    org_count = Organization.query.count()
+    assert org_count == 1
+
+
+def test_populate_organization_agreement_details_from_file(notify_db_session, notify_api):
+    file_name = "./tests/app/orgs.csv"
+
+    org_count = Organization.query.count()
+    assert org_count == 0
+    create_organization()
+    org_count = Organization.query.count()
+    assert org_count == 1
+
+    org = Organization.query.one()
+    org.agreement_signed = True
+    notify_db_session.commit()
+
+    text = "id,agreement_signed_version,agreement_signed_on_behalf_of_name,agreement_signed_at\n" \
+           f"{org.id},1,bob,'2023-01-01 00:00:00'\n"
+    f = open(file_name, "a")
+    f.write(text)
+    f.close()
+    x = notify_api.test_cli_runner().invoke(
+        populate_organization_agreement_details_from_file, [
+            '-f', file_name
+        ]
+    )
+    print(f"X = {x}")
+
+    org_count = Organization.query.count()
+    assert org_count == 1
+    org = Organization.query.one()
+    assert org.agreement_signed_on_behalf_of_name == 'bob'
+    os.remove(file_name)
 
 
 def test_create_test_user_command(notify_db_session, notify_api):
@@ -73,6 +237,39 @@ def test_populate_annual_billing_with_defaults(
     assert results[0].free_sms_fragment_limit == expected_allowance
 
 
+@pytest.mark.parametrize("organization_type, expected_allowance",
+                         [('federal', 40000),
+                          ('state', 40000)])
+def test_populate_annual_billing_with_the_previous_years_allowance(
+        notify_db_session, notify_api, organization_type, expected_allowance
+):
+    service = create_service(service_name=organization_type, organization_type=organization_type)
+
+    notify_api.test_cli_runner().invoke(
+        populate_annual_billing_with_defaults, ['-y', 2022]
+    )
+
+    results = AnnualBilling.query.filter(
+        AnnualBilling.financial_year_start == 2022,
+        AnnualBilling.service_id == service.id
+    ).all()
+
+    assert len(results) == 1
+    assert results[0].free_sms_fragment_limit == expected_allowance
+
+    notify_api.test_cli_runner().invoke(
+        populate_annual_billing_with_the_previous_years_allowance, ['-y', 2023]
+    )
+
+    results = AnnualBilling.query.filter(
+        AnnualBilling.financial_year_start == 2023,
+        AnnualBilling.service_id == service.id
+    ).all()
+
+    assert len(results) == 1
+    assert results[0].free_sms_fragment_limit == expected_allowance
+
+
 def test_populate_annual_billing_with_defaults_sets_free_allowance_to_zero_if_previous_year_is_zero(
         notify_db_session, notify_api
 ):