Cic notify build

apps/cic-notify/tests/conftest.py

@@ -0,0 +1,31 @@
+# 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 *
+
+logg = logging.getLogger()
+
+
+#@pytest.fixture(scope='session')
+#def africastalking_notification(
+#        load_config,
+#        ):
+#    return AfricastalkingNotificationTransport(load_config)
+#

apps/cic-notify/tests/fixtures_celery.py

@@ -0,0 +1,48 @@
+# third-party imports
+import pytest
+import tempfile
+import logging
+import shutil
+
+logg = logging.getLogger(__file__)
+
+
+# celery fixtures
+@pytest.fixture(scope='session')
+def celery_includes():
+    return [
+        'cic_notify.tasks.sms',
+    ]
+
+
+@pytest.fixture(scope='session')
+def celery_config():
+    bq = tempfile.mkdtemp()
+    bp = tempfile.mkdtemp()
+    rq = tempfile.mkdtemp()
+    logg.debug('celery broker queue {} processed {}'.format(bq, bp))
+    logg.debug('celery backend store {}'.format(rq))
+    yield {
+        'broker_url': 'filesystem://',
+        'broker_transport_options': {
+            'data_folder_in': bq,
+            'data_folder_out': bq,
+            'data_folder_processed': bp,
+            },
+        'result_backend': 'file://{}'.format(rq),
+            }
+    logg.debug('cleaning up celery filesystem backend files {} {} {}'.format(bq, bp, rq))
+    shutil.rmtree(bq)
+    shutil.rmtree(bp)
+    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,20 @@
+# 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

@@ -0,0 +1,48 @@
+# 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/test_sms.py

@@ -0,0 +1,34 @@
+# 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()