Adding some structure for authenticated tests
This commit is contained in:
parent
37669b906c
commit
0285422654
|
@ -181,7 +181,7 @@ class Accounts(AuthenticatedResource):
|
|||
@marshal_items(FIELDS)
|
||||
def put(self, account_id):
|
||||
"""
|
||||
.. http:post:: /accounts/1
|
||||
.. http:put:: /accounts/1
|
||||
|
||||
Updates an account
|
||||
|
||||
|
|
|
@ -96,9 +96,8 @@ def login_required(f):
|
|||
response.status_code = 401
|
||||
return response
|
||||
|
||||
token = request.headers.get('Authorization').split()[1]
|
||||
|
||||
try:
|
||||
token = request.headers.get('Authorization').split()[1]
|
||||
payload = jwt.decode(token, current_app.config['TOKEN_SECRET'])
|
||||
except jwt.DecodeError:
|
||||
return dict(message='Token is invalid'), 403
|
||||
|
|
|
@ -1,9 +1,11 @@
|
|||
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 flask.ext.principal import Identity, identity_changed
|
||||
|
||||
from sqlalchemy import event
|
||||
|
||||
|
@ -45,26 +47,45 @@ def app():
|
|||
ctx.pop()
|
||||
|
||||
|
||||
@pytest.yield_fixture(scope="session")
|
||||
def db():
|
||||
_db.create_all()
|
||||
|
||||
yield _db
|
||||
|
||||
_db.drop_all()
|
||||
@pytest.yield_fixture(scope="function")
|
||||
def unauth_client(app):
|
||||
with app.test_client() as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.yield_fixture(scope="function")
|
||||
def session(app, db):
|
||||
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()
|
||||
|
||||
yield app.db
|
||||
|
||||
app.db.drop_all()
|
||||
|
||||
|
||||
@pytest.yield_fixture(scope="function")
|
||||
def session(database):
|
||||
"""
|
||||
Creates a new database session with (with working transaction)
|
||||
for test duration.
|
||||
"""
|
||||
connection = _db.engine.connect()
|
||||
connection = database.engine.connect()
|
||||
transaction = connection.begin()
|
||||
|
||||
options = dict(bind=connection)
|
||||
session = _db.create_scoped_session(options=options)
|
||||
session = database.create_scoped_session(options=options)
|
||||
|
||||
# then each time that SAVEPOINT ends, reopen it
|
||||
@event.listens_for(SignallingSession, "after_transaction_end")
|
||||
|
@ -81,7 +102,7 @@ def session(app, db):
|
|||
# pushing new Flask application context for multiple-thread
|
||||
# tests to work
|
||||
|
||||
_db.session = session
|
||||
database.session = session
|
||||
|
||||
yield session
|
||||
|
||||
|
|
|
@ -40,14 +40,7 @@ def test_unauthenticated_views(client):
|
|||
assert client.delete(api.url_for(Accounts, account_id=1)).status_code == 401
|
||||
assert client.patch(api.url_for(Accounts, account_id=1), {}).status_code == 405
|
||||
|
||||
assert client.get(api.url_for(AccountsList)).status_code == 401
|
||||
assert client.post(api.url_for(AccountsList), {}).status_code == 401
|
||||
assert client.put(api.url_for(AccountsList), {}).status_code == 405
|
||||
assert client.delete(api.url_for(AccountsList)).status_code == 405
|
||||
assert client.patch(api.url_for(Accounts), {}).status_code == 405
|
||||
VALID_TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyMzMzNjksInN1YiI6MSwiZXhwIjoxNTIxNTQ2OTY5fQ.1qCi0Ip7mzKbjNh0tVd3_eJOrae3rNa_9MCVdA4WtQI'
|
||||
|
||||
assert client.get(api.url_for(CertificateAccounts, certificate_id=1)).status_code == 401
|
||||
assert client.post(api.url_for(CertificateAccounts), {}).status_code == 405
|
||||
assert client.put(api.url_for(CertificateAccounts), {}).status_code == 405
|
||||
assert client.delete(api.url_for(CertificateAccounts)).status_code == 405
|
||||
assert client.patch(api.url_for(CertificateAccounts), {}).status_code == 405
|
||||
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
|
Loading…
Reference in New Issue