Added dao, test framework and tests for dao.

app/__init__.py

@@ -10,7 +10,6 @@ from utils import logging
 
 db = SQLAlchemy()
 
-
 api_user = LocalProxy(lambda: _request_ctx_stack.top.api_user)
 
 

app/main/dao/services_dao.py

@@ -6,7 +6,7 @@ from app import db
 from app.models import Service
 
 
-def create_new_service(service_name,
+def create_service(service_name,
                    user,
                    limit=1000,
                    active=False,
@@ -22,7 +22,10 @@ def create_new_service(service_name,
     return service.id
 
 
-def get_services(user, service_id=None):
+def get_services(service_id=None, user_id=None):
+    # TODO need better mapping from function params to sql query.
     if service_id:
-        return Service.query.filter_by(user=user, service_id=service_id).one()
+        return Service.query.filter_by(id=service_id).one()
-    return Service.query.filter_by(user=user).all()
+    elif user_id:
+        return Service.query.filter(Service.users.any(id=user_id)).all()
+    return Service.query.all()

app/main/dao/users_dao.py

@@ -6,7 +6,7 @@ from app import db
 from app.models import User
 
 
-def create_new_user(email_address):
+def create_user(email_address):
     user = User(email_address=email_address,
                 created_at=datetime.now())
     db.session.add(user)
@@ -16,5 +16,5 @@ def create_new_user(email_address):
 
 def get_users(user_id=None):
     if user_id:
-        return User.query.filter_by(user_id=user_id).one()
+        return User.query.filter_by(id=user_id).one()
     return User.query.filter_by().all()

config.py

@@ -15,7 +15,7 @@ class Development(Config):
 
 class Test(Config):
     DEBUG = True
-    SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/notification_api_test'
+    SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/test_notification_api'
 
 
 class Live(Config):

tests/app/conftest.py

@@ -0,0 +1,22 @@
+import pytest
+from app.main.dao.users_dao import (create_user, get_users)
+from app.main.dao.services_dao import (create_service, get_services)
+
+
+@pytest.fixture(scope='function')
+def sample_user(notify_db,
+                notify_db_session,
+                email="notify@digital.cabinet-office.gov.uk"):
+    user_id = create_user(email)
+    return get_users(user_id=user_id)
+
+
+@pytest.fixture(scope='function')
+def sample_service(notify_db,
+                   notify_db_session,
+                   service_name="Sample service",
+                   user=None):
+    if user is None:
+        user = sample_user(notify_db, notify_db_session)
+    service_id = create_service(service_name, user)
+    return get_services(service_id=service_id)

tests/app/main/dao/test_services_dao.py

@@ -1,8 +1,38 @@
+from app.main.dao.services_dao import (create_service, get_services)
+from tests.app.conftest import sample_service as create_sample_service
+from app.models import Service
 
 
-def test_create_service(notify_api):
+def test_create_service(notify_api, notify_db, notify_db_session, sample_user):
-    pass
+    assert Service.query.count() == 0
+    service_name = 'Sample Service'
+    service_id = create_service(service_name, sample_user)
+    assert Service.query.count() == 1
+    assert Service.query.first().name == service_name
+    assert Service.query.first().id == service_id
 
 
-def test_get_all_services(notify_api):
+def test_get_services(notify_api, notify_db, notify_db_session, sample_user):
-    pass
+    sample_service = create_sample_service(notify_db,
+                                           notify_db_session,
+                                           user=sample_user)
+    assert Service.query.count() == 1
+    assert len(get_services()) == 1
+    service_name = "Another service"
+    sample_service = create_sample_service(notify_db,
+                                           notify_db_session,
+                                           service_name=service_name,
+                                           user=sample_user)
+    assert Service.query.count() == 2
+    assert len(get_services()) == 2
+
+
+def test_get_user_service(notify_api, notify_db, notify_db_session, sample_user):
+    assert Service.query.count() == 0
+    service_name = "Random service"
+    sample_service = create_sample_service(notify_db,
+                                           notify_db_session,
+                                           service_name=service_name,
+                                           user=sample_user)
+    assert get_services(service_id=sample_service.id).name == service_name
+    assert Service.query.count() == 1

tests/app/main/dao/test_users_dao.py

@@ -0,0 +1,30 @@
+from app.main.dao.users_dao import (create_user, get_users)
+from tests.app.conftest import sample_user as create_sample_user
+from app.models import User
+
+
+def test_create_user(notify_api, notify_db, notify_db_session):
+    email = 'notify@digital.cabinet-office.gov.uk'
+    user_id = create_user(email)
+    assert User.query.count() == 1
+    assert User.query.first().email_address == email
+    assert User.query.filter_by(id=user_id).one()
+
+
+def test_get_all_users(notify_api, notify_db, notify_db_session, sample_user):
+    assert User.query.count() == 1
+    assert len(get_users()) == 1
+    email = "another.notify@digital.cabinet-office.gov.uk"
+    another_user = create_sample_user(notify_db,
+                                      notify_db_session,
+                                      email=email)
+    assert User.query.count() == 2
+    assert len(get_users()) == 2
+
+
+def test_get_user(notify_api, notify_db, notify_db_session):
+    email = "another.notify@digital.cabinet-office.gov.uk"
+    another_user = create_sample_user(notify_db,
+                                      notify_db_session,
+                                      email=email)
+    assert get_users(user_id=another_user.id).email_address == email

tests/conftest.py

@@ -6,8 +6,9 @@ from alembic.command import upgrade
 from alembic.config import Config
 from flask.ext.migrate import Migrate, MigrateCommand
 from flask.ext.script import Manager
-
+from sqlalchemy.schema import MetaData
 from app import create_app, db
+from app import models
 
 
 @pytest.fixture(scope='session')
@@ -33,6 +34,10 @@ def notify_db(notify_api, request):
     config.set_main_option("script_location", ALEMBIC_CONFIG)
 
     with notify_api.app_context():
+        # TODO this next line shouldn't be needed,
+        # but cannot work out the import order to
+        # remove it.
+        db.create_all()
         upgrade(config, 'head')
 
     def teardown():
@@ -44,6 +49,18 @@ def notify_db(notify_api, request):
     request.addfinalizer(teardown)
 
 
+@pytest.fixture(scope='function')
+def notify_db_session(request):
+    def teardown():
+        db.session.remove()
+        for tbl in reversed(meta.sorted_tables):
+            if tbl.fullname not in ['roles']:
+                db.engine.execute(tbl.delete())
+
+    meta = MetaData(bind=db.engine, reflect=True)
+    request.addfinalizer(teardown)
+
+
 @pytest.fixture(scope='function')
 def notify_config(notify_api):
     notify_api.config['NOTIFY_API_ENVIRONMENT'] = 'test'