Upgrades cic-notify test suite.

apps/cic-notify/.coveragerc

@@ -0,0 +1,7 @@
+[report]
+omit =
+	venv/*
+	scripts/*
+	cic_notify/db/migrations/*
+	cic_notify/runnable/*
+	cic_notify/version.py

apps/cic-notify/test_requirements.txt

@@ -1,5 +1,7 @@
-pytest~=6.0.1
-pytest-celery~=0.0.0a1
-pytest-mock~=3.3.1
-pysqlite3~=0.4.3
-pytest-cov==2.10.1
+pytest==6.2.5
+pytest-celery~=0.0.0
+pytest-mock==3.6.1
+pysqlite3~=0.4.6
+pytest-cov==3.0.0
+pytest-alembic==0.7.0
+requests-mock==1.9.3

apps/cic-notify/tests/cic_notify/db/migrations/test_migrations.py

@@ -0,0 +1,28 @@
+import pytest
+
+
+def test_single_head_revision(alembic_runner):
+    heads = alembic_runner.heads
+    head_count = len(heads)
+    assert head_count == 1
+
+
+def test_upgrade(alembic_runner):
+    try:
+        alembic_runner.migrate_up_to("head")
+    except RuntimeError:
+        pytest.fail('Failed to upgrade to the head revision.')
+
+
+def test_up_down_consistency(alembic_runner):
+    try:
+        for revision in alembic_runner.history.revisions:
+            alembic_runner.migrate_up_to(revision)
+    except RuntimeError:
+        pytest.fail('Failed to upgrade through each revision individually.')
+
+    try:
+        for revision in reversed(alembic_runner.history.revisions):
+            alembic_runner.migrate_down_to(revision)
+    except RuntimeError:
+        pytest.fail('Failed to downgrade through each revision individually.')

apps/cic-notify/tests/cic_notify/db/models/test_notification.py

@@ -0,0 +1,27 @@
+# standard imports
+
+# external imports
+from faker import Faker
+from faker_e164.providers import E164Provider
+
+# local imports
+from cic_notify.db.enum import NotificationStatusEnum, NotificationTransportEnum
+from cic_notify.db.models.notification import Notification
+
+
+# test imports
+from tests.helpers.phone import phone_number
+
+
+def test_notification(init_database):
+    message = 'Hello world'
+    recipient = phone_number()
+    notification = Notification(NotificationTransportEnum.SMS, recipient, message)
+    init_database.add(notification)
+    init_database.commit()
+
+    notification = init_database.query(Notification).get(1)
+    assert notification.status == NotificationStatusEnum.UNKNOWN
+    assert notification.recipient == recipient
+    assert notification.message == message
+    assert notification.transport == NotificationTransportEnum.SMS

apps/cic-notify/tests/cic_notify/db/test_db.py

@@ -0,0 +1,38 @@
+# standard imports
+import os
+
+# third-party imports
+
+# local imports
+from cic_notify.db import dsn_from_config
+
+
+def test_dsn_from_config(load_config):
+    """
+    """
+    # test dsn for other db formats
+    overrides = {
+        'DATABASE_PASSWORD': 'password',
+        'DATABASE_DRIVER': 'psycopg2',
+        'DATABASE_ENGINE': 'postgresql'
+    }
+    load_config.dict_override(dct=overrides, dct_description='Override values to test different db formats.')
+
+    scheme = f'{load_config.get("DATABASE_ENGINE")}+{load_config.get("DATABASE_DRIVER")}'
+
+    dsn = dsn_from_config(load_config)
+    assert dsn == f"{scheme}://{load_config.get('DATABASE_USER')}:{load_config.get('DATABASE_PASSWORD')}@{load_config.get('DATABASE_HOST')}:{load_config.get('DATABASE_PORT')}/{load_config.get('DATABASE_NAME')}"
+
+    # undoes overrides to revert engine and drivers to sqlite
+    overrides = {
+        'DATABASE_PASSWORD': '',
+        'DATABASE_DRIVER': 'pysqlite',
+        'DATABASE_ENGINE': 'sqlite'
+    }
+    load_config.dict_override(dct=overrides, dct_description='Override values to test different db formats.')
+
+    # test dsn for sqlite engine
+    dsn = dsn_from_config(load_config)
+    scheme = f'{load_config.get("DATABASE_ENGINE")}+{load_config.get("DATABASE_DRIVER")}'
+    assert dsn == f'{scheme}:///{load_config.get("DATABASE_NAME")}'
+

apps/cic-notify/tests/cic_notify/tasks/sms/test_africastalking_tasks.py

@@ -0,0 +1,73 @@
+# standard imports
+import os
+
+# external imports
+import pytest
+import requests_mock
+
+
+# local imports
+from cic_notify.error import NotInitializedError, AlreadyInitializedError, NotificationSendError
+from cic_notify.tasks.sms.africastalking import AfricasTalkingNotifier
+
+# test imports
+from tests.helpers.phone import phone_number
+
+
+def test_africas_talking_notifier(africastalking_response, caplog):
+    with pytest.raises(NotInitializedError) as error:
+        AfricasTalkingNotifier()
+    assert str(error.value) == ''
+
+    api_key = os.urandom(24).hex()
+    sender_id = 'bar'
+    username = 'sandbox'
+    AfricasTalkingNotifier.initialize(username, api_key, sender_id)
+    africastalking_notifier = AfricasTalkingNotifier()
+    assert africastalking_notifier.sender_id == sender_id
+    assert africastalking_notifier.initiated is True
+
+    with pytest.raises(AlreadyInitializedError) as error:
+        AfricasTalkingNotifier.initialize(username, api_key, sender_id)
+    assert str(error.value) == ''
+
+    with requests_mock.Mocker(real_http=False) as request_mocker:
+        message = 'Hello world.'
+        recipient = phone_number()
+        africastalking_response.get('SMSMessageData').get('Recipients')[0]['number'] = recipient
+        request_mocker.register_uri(method='POST',
+                                    headers={'content-type': 'application/json'},
+                                    json=africastalking_response,
+                                    url='https://api.sandbox.africastalking.com/version1/messaging',
+                                    status_code=200)
+        africastalking_notifier.send(message, recipient)
+        assert f'Africastalking response sender-id {africastalking_response}' in caplog.text
+        africastalking_notifier.sender_id = None
+        africastalking_notifier.send(message, recipient)
+        assert f'africastalking response no-sender-id {africastalking_response}' in caplog.text
+        with pytest.raises(NotificationSendError) as error:
+            status = 'InvalidPhoneNumber'
+            status_code = 403
+            africastalking_response.get('SMSMessageData').get('Recipients')[0]['status'] = status
+            africastalking_response.get('SMSMessageData').get('Recipients')[0]['statusCode'] = status_code
+
+            request_mocker.register_uri(method='POST',
+                                        headers={'content-type': 'application/json'},
+                                        json=africastalking_response,
+                                        url='https://api.sandbox.africastalking.com/version1/messaging',
+                                        status_code=200)
+            africastalking_notifier.send(message, recipient)
+        assert str(error.value) == f'Sending notification failed due to: {status}'
+        with pytest.raises(NotificationSendError) as error:
+            recipients = []
+            status = 'InsufficientBalance'
+            africastalking_response.get('SMSMessageData')['Recipients'] = recipients
+            africastalking_response.get('SMSMessageData')['Message'] = status
+
+            request_mocker.register_uri(method='POST',
+                                        headers={'content-type': 'application/json'},
+                                        json=africastalking_response,
+                                        url='https://api.sandbox.africastalking.com/version1/messaging',
+                                        status_code=200)
+            africastalking_notifier.send(message, recipient)
+        assert str(error.value) == f'Unexpected number of recipients: {len(recipients)}. Status: {status}'

apps/cic-notify/tests/cic_notify/tasks/sms/test_db_tasks.py

@@ -0,0 +1,26 @@
+# standard imports
+
+# external imports
+import celery
+
+# local imports
+from cic_notify.db.enum import NotificationStatusEnum, NotificationTransportEnum
+from cic_notify.db.models.notification import Notification
+
+# test imports
+from tests.helpers.phone import phone_number
+
+
+def test_persist_notification(celery_session_worker, init_database):
+    message = 'Hello world.'
+    recipient = phone_number()
+    s_persist_notification = celery.signature(
+        'cic_notify.tasks.sms.db.persist_notification', (message, recipient)
+    )
+    s_persist_notification.apply_async().get()
+
+    notification = Notification.session.query(Notification).filter_by(recipient=recipient).first()
+    assert notification.status == NotificationStatusEnum.UNKNOWN
+    assert notification.recipient == recipient
+    assert notification.message == message
+    assert notification.transport == NotificationTransportEnum.SMS

apps/cic-notify/tests/cic_notify/tasks/sms/test_log_tasks.py

@@ -0,0 +1,19 @@
+# standard imports
+
+# external imports
+import celery
+
+# local imports
+
+# test imports
+from tests.helpers.phone import phone_number
+
+
+def test_log(caplog, celery_session_worker):
+    message = 'Hello world.'
+    recipient = phone_number()
+    s_log = celery.signature(
+        'cic_notify.tasks.sms.log.log', [message, recipient]
+    )
+    s_log.apply_async().get()
+    assert f'message to {recipient}: {message}' in caplog.text

apps/cic-notify/tests/conftest.py

@@ -1,31 +1,13 @@
 # standard imports
-import sys
-import os
-import pytest
 import logging
 
 # third party imports
-import confini
-
-script_dir = os.path.dirname(os.path.realpath(__file__))
-root_dir = os.path.dirname(script_dir)
-sys.path.insert(0, root_dir)
 
 # local imports
-from cic_notify.db.models.base import SessionBase
-#from transport.notification import AfricastalkingNotification
 
-# fixtures
-from tests.fixtures_config import *
-from tests.fixtures_celery import *
-from tests.fixtures_database import *
+# test imports
 
-logg = logging.getLogger()
-
-
-#@pytest.fixture(scope='session')
-#def africastalking_notification(
-#        load_config,
-#        ):
-#    return AfricastalkingNotificationTransport(load_config)
-#
+from .fixtures.celery import *
+from .fixtures.config import *
+from .fixtures.database import *
+from .fixtures.result import *

apps/cic-notify/tests/fixtures/celery.py

@@ -37,12 +37,6 @@ def celery_config():
     shutil.rmtree(rq)
 
 
-@pytest.fixture(scope='session')
-def celery_worker_parameters():
-    return {
-#            'queues': ('cic-notify'),
-            }
-
 @pytest.fixture(scope='session')
 def celery_enable_logging():
     return True

apps/cic-notify/tests/fixtures/config.py

@@ -0,0 +1,32 @@
+# standard imports
+import os
+import logging
+
+# external imports
+import pytest
+from confini import Config
+
+logg = logging.getLogger(__file__)
+
+
+fixtures_dir = os.path.dirname(__file__)
+root_directory = os.path.dirname(os.path.dirname(fixtures_dir))
+
+
+@pytest.fixture(scope='session')
+def alembic_config():
+    migrations_directory = os.path.join(root_directory, 'cic_notify', 'db', 'migrations', 'default')
+    file = os.path.join(migrations_directory, 'alembic.ini')
+    return {
+        'file': file,
+        'script_location': migrations_directory
+    }
+
+
+@pytest.fixture(scope='session')
+def load_config():
+    config_directory = os.path.join(root_directory, '.config/test')
+    config = Config(default_dir=config_directory)
+    config.process()
+    logg.debug('config loaded\n{}'.format(config))
+    return config

apps/cic-notify/tests/fixtures/database.py

@@ -0,0 +1,54 @@
+# standard imports
+import os
+
+# third-party imports
+import pytest
+import alembic
+from alembic.config import Config as AlembicConfig
+
+# local imports
+from cic_notify.db import dsn_from_config
+from cic_notify.db.models.base import SessionBase, create_engine
+
+from .config import root_directory
+
+
+@pytest.fixture(scope='session')
+def alembic_engine(load_config):
+    data_source_name = dsn_from_config(load_config)
+    return create_engine(data_source_name)
+
+
+@pytest.fixture(scope='session')
+def database_engine(load_config):
+    if load_config.get('DATABASE_ENGINE') == 'sqlite':
+        try:
+            os.unlink(load_config.get('DATABASE_NAME'))
+        except FileNotFoundError:
+            pass
+    dsn = dsn_from_config(load_config)
+    SessionBase.connect(dsn)
+    return dsn
+
+
+@pytest.fixture(scope='function')
+def init_database(load_config, database_engine):
+    db_directory = os.path.join(root_directory, 'cic_notify', 'db')
+    migrations_directory = os.path.join(db_directory, 'migrations', load_config.get('DATABASE_ENGINE'))
+    if not os.path.isdir(migrations_directory):
+        migrations_directory = os.path.join(db_directory, 'migrations', 'default')
+
+    session = SessionBase.create_session()
+
+    alembic_config = AlembicConfig(os.path.join(migrations_directory, 'alembic.ini'))
+    alembic_config.set_main_option('sqlalchemy.url', database_engine)
+    alembic_config.set_main_option('script_location', migrations_directory)
+
+    alembic.command.downgrade(alembic_config, 'base')
+    alembic.command.upgrade(alembic_config, 'head')
+
+    yield session
+    session.commit()
+    session.close()
+
+

apps/cic-notify/tests/fixtures/result.py

@@ -0,0 +1,24 @@
+# standard imports
+
+# external imports
+import pytest
+
+
+# local imports
+
+# test imports
+
+@pytest.fixture(scope="function")
+def africastalking_response():
+    return {
+        "SMSMessageData": {
+            "Message": "Sent to 1/1 Total Cost: KES 0.8000",
+            "Recipients": [{
+                "statusCode": 101,
+                "number": "+254711XXXYYY",
+                "status": "Success",
+                "cost": "KES 0.8000",
+                "messageId": "ATPid_SampleTxnId123"
+            }]
+        }
+    }

apps/cic-notify/tests/fixtures_config.py

@@ -1,20 +0,0 @@
-# standard imports
-import os
-import logging
-
-# third-party imports
-import pytest
-import confini
-
-script_dir = os.path.dirname(os.path.realpath(__file__))
-root_dir = os.path.dirname(script_dir)
-logg = logging.getLogger(__file__)
-
-
-@pytest.fixture(scope='session')
-def load_config():
-    config_dir = os.path.join(root_dir, '.config/test')
-    conf = confini.Config(config_dir, 'CICTEST')
-    conf.process()
-    logg.debug('config {}'.format(conf))
-    return conf

apps/cic-notify/tests/fixtures_database.py

@@ -1,48 +0,0 @@
-# standard imports
-import os
-
-# third-party imports
-import pytest
-import alembic
-from alembic.config import Config as AlembicConfig
-
-# local imports
-from cic_notify.db import SessionBase
-from cic_notify.db import dsn_from_config
-
-
-@pytest.fixture(scope='session')
-def database_engine(
-    load_config,
-        ):
-    dsn = dsn_from_config(load_config)
-    SessionBase.connect(dsn)
-    return dsn
-
-
-@pytest.fixture(scope='function')
-def init_database(
-        load_config,
-        database_engine,
-        ):
-
-    rootdir = os.path.dirname(os.path.dirname(__file__))
-    dbdir = os.path.join(rootdir, 'cic_notify', 'db')
-    migrationsdir = os.path.join(dbdir, 'migrations', load_config.get('DATABASE_ENGINE'))
-    if not os.path.isdir(migrationsdir):
-        migrationsdir = os.path.join(dbdir, 'migrations', 'default')
-
-    session = SessionBase.create_session()
-
-    ac = AlembicConfig(os.path.join(migrationsdir, 'alembic.ini'))
-    ac.set_main_option('sqlalchemy.url', database_engine)
-    ac.set_main_option('script_location', migrationsdir)
-
-    alembic.command.downgrade(ac, 'base')
-    alembic.command.upgrade(ac, 'head')
-
-    yield session
-    session.commit()
-    session.close()
-
-

apps/cic-notify/tests/helpers/phone.py

@@ -0,0 +1,16 @@
+# standard imports
+
+# external imports
+from faker import Faker
+from faker_e164.providers import E164Provider
+
+# local imports
+
+# test imports
+
+fake = Faker()
+fake.add_provider(E164Provider)
+
+
+def phone_number() -> str:
+    return fake.e164('KE')

apps/cic-notify/tests/test_sms.py

@@ -1,34 +0,0 @@
-# standard imports
-import json
-
-# third party imports
-import pytest
-import celery
-
-# local imports
-from cic_notify.tasks.sms import db
-from cic_notify.tasks.sms import log
-
-def test_log_notification(
-    celery_session_worker,
-        ):
-
-    recipient = '+25412121212'
-    content = 'bar'
-    s_log = celery.signature('cic_notify.tasks.sms.log.log')
-    t = s_log.apply_async(args=[recipient, content])
-
-    r = t.get()
-
-
-def test_db_notification(
-    init_database,
-    celery_session_worker,
-        ):
-
-    recipient = '+25412121213'
-    content = 'foo'
-    s_db = celery.signature('cic_notify.tasks.sms.db.persist_notification')
-    t = s_db.apply_async(args=[recipient, content])
-
-    r = t.get()