Adding more tests to the accounts model

lemur/extensions.py

@@ -4,41 +4,7 @@
     :license: Apache, see LICENSE for more details.
 """
 
-from flask.ext.sqlalchemy import SQLAlchemy, SignallingSession, SessionBase
+from flask.ext.sqlalchemy import SQLAlchemy
-
-
-class _SignallingSession(SignallingSession):
-    """A subclass of `SignallingSession` that allows for `binds` to be specified
-    in the `options` keyword arguments.
-
-    """
-    def __init__(self, db, autocommit=False, autoflush=True, **options):
-        self.app = db.get_app()
-        self._model_changes = {}
-        self.emit_modification_signals = \
-            self.app.config['SQLALCHEMY_TRACK_MODIFICATIONS']
-
-        bind = options.pop('bind', None)
-        if bind is None:
-            bind = db.engine
-
-        binds = options.pop('binds', None)
-        if binds is None:
-            binds = db.get_binds(self.app)
-
-        SessionBase.__init__(self,
-                             autocommit=autocommit,
-                             autoflush=autoflush,
-                             bind=bind,
-                             binds=binds,
-                             **options)
-
-
-class _SQLAlchemy(SQLAlchemy):
-    """A subclass of `SQLAlchemy` that uses `_SignallingSession`."""
-    def create_session(self, options):
-        return _SignallingSession(self, **options)
-
 db = SQLAlchemy()
 
 from flask.ext.migrate import Migrate

lemur/tests/conftest.py

@@ -1,13 +1,9 @@
 import pytest
 
-from flask import current_app
-
 from lemur import create_app
-
+from lemur.database import db as _db
-from flask.ext.sqlalchemy import SignallingSession
+from lemur.users import service as user_service
-from flask.ext.principal import Identity, identity_changed
+from lemur.roles import service as role_service
-
-from sqlalchemy import event
 
 
 def pytest_addoption(parser):
@@ -38,6 +34,7 @@ def app():
     Uses application factory `create_app`.
     """
     app = create_app()
+    app.config['TESTING'] = True
 
     ctx = app.app_context()
     ctx.push()
@@ -47,66 +44,45 @@ def app():
     ctx.pop()
 
 
-@pytest.yield_fixture(scope="function")
-def unauth_client(app):
-    with app.test_client() as client:
-        yield client
-
-
-@pytest.yield_fixture(scope="function")
-def auth_client(app):
-    with app.test_client() as client:
-        yield client
-
-
-@pytest.yield_fixture(scope="function")
-def admin_client(app):
-    with app.test_client() as client:
-        yield client
-
-
 
 @pytest.yield_fixture(scope="session")
-def database(app):
+def db(app, request):
-    app.db.create_all()
+    _db.drop_all()
+    _db.create_all()
 
-    yield app.db
+    _db.app = app
 
-    app.db.drop_all()
+    yield _db
+
+    _db.drop_all()
 
 
 @pytest.yield_fixture(scope="function")
-def session(database):
+def session(db, request):
     """
     Creates a new database session with (with working transaction)
     for test duration.
     """
-    connection = database.engine.connect()
+    db.session.begin_nested()
-    transaction = connection.begin()
-
-    options = dict(bind=connection)
-    session = database.create_scoped_session(options=options)
-
-    # then each time that SAVEPOINT ends, reopen it
-    @event.listens_for(SignallingSession, "after_transaction_end")
-    def restart_savepoint(session, transaction):
-        if transaction.nested and not transaction._parent.nested:
-
-            # ensure that state is expired the way
-            # session.commit() at the top level normally does
-            # (optional step)
-            session.expire_all()
-
-            session.begin_nested()
-
-    # pushing new Flask application context for multiple-thread
-    # tests to work
-
-    database.session = session
-
     yield session
+    db.session.rollback()
+
+
+@pytest.yield_fixture(scope="session")
+def default_user(db):
+    user = user_service.create('user', 'test', 'user@example.com', True, None, [])
+    yield user
+
+
+@pytest.yield_fixture(scope="session")
+def admin_user(db):
+    admin_role = role_service.create('admin')
+    admin = user_service.create('admin', 'admin', 'admin@example.com', True, None, [admin_role])
+    yield admin
+
+
+@pytest.yield_fixture(scope="function")
+def client(app):
+    with app.test_client() as client:
+        yield client
 
-    # the code after the yield statement works as a teardown
-    transaction.rollback()
-    connection.close()
-    session.remove()

lemur/tests/test_accounts.py

@@ -1,46 +1,132 @@
-
-import pytest
 from lemur.accounts.service import *
-from lemur.exceptions import DuplicateError
-
 from lemur.accounts.views import *
 
-#def test_crud(session):
+from json import dumps
-#    account = create('111111', 'account1')
-#    assert account.id > 0
-#
-#    account = update(account.id, 11111, 'account2')
-#    assert account.label == 'account2'
-#
-#    assert len(get_all()) == 1
-#
-#    delete(1)
-#    assert len(get_all()) == 0
-#
-
-#def test_duplicate(session):
-#    account = create('111111', 'account1')
-#    assert account.id > 0
-#
-#    with pytest.raises(DuplicateError):
-#        account = create('111111', 'account1')
 
 
-def test_basic_user_views(client):
+def test_crud(session):
-    pass
+    account = create('111111', 'account1')
+    assert account.id > 0
+
+    account = update(account.id, 11111, 'account2')
+    assert account.label == 'account2'
+
+    assert len(get_all()) == 1
+
+    delete(1)
+    assert len(get_all()) == 0
 
 
-def test_admin_user_views(client):
+def test_account_get(client):
-    pass
-
-def test_unauthenticated_views(client):
     assert client.get(api.url_for(Accounts, account_id=1)).status_code == 401
+
+
+def test_account_post(client):
     assert client.post(api.url_for(Accounts, account_id=1), {}).status_code == 405
+
+
+def test_account_put(client):
     assert client.put(api.url_for(Accounts, account_id=1), {}).status_code == 401
+
+
+def test_account_delete(client):
     assert client.delete(api.url_for(Accounts, account_id=1)).status_code == 401
+
+
+def test_account_patch(client):
     assert client.patch(api.url_for(Accounts, account_id=1), {}).status_code == 405
 
-VALID_TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyMzMzNjksInN1YiI6MSwiZXhwIjoxNTIxNTQ2OTY5fQ.1qCi0Ip7mzKbjNh0tVd3_eJOrae3rNa_9MCVdA4WtQI'
 
-def test_auth_account_get(auth_client):
+VALID_USER_HEADER_TOKEN = {
-        assert auth_client.get(api.url_for(Accounts, account_id=1), headers={'Authorization': 'Basic ' + VALID_TOKEN}).status_code == 200
+    'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyMzMzNjksInN1YiI6MSwiZXhwIjoxNTIxNTQ2OTY5fQ.1qCi0Ip7mzKbjNh0tVd3_eJOrae3rNa_9MCVdA4WtQI'}
+
+def test_auth_account_get(client, default_user):
+    assert client.get(api.url_for(Accounts, account_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 200
+
+
+def test_auth_account_post_(client, default_user):
+    assert client.post(api.url_for(Accounts, account_id=1), {}, headers=VALID_USER_HEADER_TOKEN).status_code == 405
+
+
+def test_auth_account_put(client, default_user):
+    assert client.put(api.url_for(Accounts, account_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 403
+
+
+def test_auth_account_delete(client, default_user):
+    assert client.delete(api.url_for(Accounts, account_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 403
+
+
+def test_auth_account_patch(client, default_user):
+    assert client.patch(api.url_for(Accounts, account_id=1), {}, headers=VALID_USER_HEADER_TOKEN).status_code == 405
+
+
+VALID_ADMIN_HEADER_TOKEN = {
+    'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyNTAyMTgsInN1YiI6MiwiZXhwIjoxNTIxNTYzODE4fQ.6mbq4-Ro6K5MmuNiTJBB153RDhlM5LGJBjI7GBKkfqA'}
+
+def test_admin_account_get(client, admin_user):
+    assert client.get(api.url_for(Accounts, account_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 200
+
+
+def test_admin_account_post(client, admin_user):
+    assert client.post(api.url_for(Accounts, account_id=1), {}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405
+
+
+def test_admin_account_put(client, admin_user):
+    assert client.put(api.url_for(Accounts, account_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 400
+
+
+def test_admin_account_delete(client, admin_user):
+    assert client.delete(api.url_for(Accounts, account_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 500
+
+
+def test_admin_account_patch(client, admin_user):
+    assert client.patch(api.url_for(Accounts, account_id=1), {}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405
+
+
+def test_accounts_get(client):
+    assert client.get(api.url_for(AccountsList)).status_code == 401
+
+
+def test_accounts_post(client):
+    assert client.post(api.url_for(AccountsList), {}).status_code == 401
+
+
+def test_accounts_put(client):
+    assert client.put(api.url_for(AccountsList), {}).status_code == 405
+
+
+def test_accounts_delete(client):
+    assert client.delete(api.url_for(AccountsList)).status_code == 405
+
+
+def test_accounts_patch(client):
+    assert client.patch(api.url_for(AccountsList), {}).status_code == 405
+
+
+def test_auth_accounts_get(client, default_user):
+    assert client.get(api.url_for(AccountsList), headers=VALID_USER_HEADER_TOKEN).status_code == 200
+
+
+def test_auth_accounts_post(client, default_user):
+    assert client.post(api.url_for(AccountsList), {}, headers=VALID_USER_HEADER_TOKEN).status_code == 403
+
+
+def test_admin_accounts_get(client, admin_user):
+    resp = client.get(api.url_for(AccountsList), headers=VALID_ADMIN_HEADER_TOKEN)
+    assert resp.status_code == 200
+    assert resp.json == {'items': [], 'total': 0}
+
+
+def test_admin_accounts_crud(client, admin_user):
+    assert client.post(api.url_for(AccountsList), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 400
+    data = {'accountNumber': 111, 'label': 'test', 'comments': 'test'}
+    resp = client.post(api.url_for(AccountsList), data=dumps(data), content_type='application/json',  headers=VALID_ADMIN_HEADER_TOKEN)
+    assert resp.status_code == 200
+    assert client.get(api.url_for(Accounts, account_id=resp.json['id']), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 200
+    resp = client.get(api.url_for(AccountsList), headers=VALID_ADMIN_HEADER_TOKEN)
+    assert resp.status_code == 200
+    assert resp.json == {'items': [{'accountNumber': 111, 'label': 'test', 'comments': 'test', 'id': 2}], 'total': 1}
+    assert client.delete(api.url_for(Accounts, account_id=2), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 200
+    resp = client.get(api.url_for(AccountsList), headers=VALID_ADMIN_HEADER_TOKEN)
+    assert resp.status_code == 200
+    assert resp.json == {'items': [], 'total': 0}