Infra
/
lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Adding more tests to the accounts model

This commit is contained in:
Kevin Glisson 2015-06-25 18:05:52 -07:00
parent 75e5bdfa55
commit 2a3fac11e4
3 changed files with 152 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
lemur/extensions.py
View File

@ -4,41 +4,7 @@
:license: Apache, see LICENSE for more details.
"""
from flask.ext.sqlalchemy import SQLAlchemy, SignallingSession, SessionBase
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)
from flask.ext.sqlalchemy import SQLAlchemy
db = SQLAlchemy()
from flask.ext.migrate import Migrate

90
lemur/tests/conftest.py
View File

@ -1,13 +1,9 @@
import pytest
from flask import current_app
from lemur import create_app
from flask.ext.sqlalchemy import SignallingSession
from flask.ext.principal import Identity, identity_changed
from sqlalchemy import event
from lemur.database import db as _db
from lemur.users import service as user_service
from lemur.roles import service as role_service
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):
app.db.create_all()
def db(app, request):
_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()
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
db.session.begin_nested()
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()

150
lemur/tests/test_accounts.py
View File

@ -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):
# 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')
from json import dumps
def test_basic_user_views(client):
pass
def test_crud(session):
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):
pass
def test_unauthenticated_views(client):
def test_account_get(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):
assert auth_client.get(api.url_for(Accounts, account_id=1), headers={'Authorization': 'Basic ' + VALID_TOKEN}).status_code == 200
VALID_USER_HEADER_TOKEN = {
'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}