From 2a3fac11e48573f50437d78a1db5b1bc83433c38 Mon Sep 17 00:00:00 2001 From: Kevin Glisson Date: Thu, 25 Jun 2015 18:05:52 -0700 Subject: [PATCH] Adding more tests to the accounts model --- lemur/extensions.py | 36 +-------- lemur/tests/conftest.py | 90 ++++++++------------- lemur/tests/test_accounts.py | 150 +++++++++++++++++++++++++++-------- 3 files changed, 152 insertions(+), 124 deletions(-) diff --git a/lemur/extensions.py b/lemur/extensions.py index 101432e8..07101c4d 100644 --- a/lemur/extensions.py +++ b/lemur/extensions.py @@ -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 diff --git a/lemur/tests/conftest.py b/lemur/tests/conftest.py index 2d680850..254a7e27 100644 --- a/lemur/tests/conftest.py +++ b/lemur/tests/conftest.py @@ -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() diff --git a/lemur/tests/test_accounts.py b/lemur/tests/test_accounts.py index 3d1de94d..08239afb 100644 --- a/lemur/tests/test_accounts.py +++ b/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): -# 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 \ No newline at end of file +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}