我有一个烧瓶应用程序并试图用pytest测试它。问题是如何让基础为每个测试功能创建一个新的？这就是我认为应该如何运作的方式。
这是我的模特。
models.朋友
class User(UserMixin, db.Model): # type: ignore
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), index=True, unique=True)
password_hash = db.Column(db.String(128))
def set_password(self, password: str) -> None:
self.password_hash = generate_password_hash(password) # type: ignore
def check_password(self, password) -> None:
return check_password_hash(self.password_hash, password) # type:ignore
class Employee(db.Model): # type: ignore
query_class = EmployeeSearch
__tablename__ = 'employee'
id = db.Column(db.Integer, primary_key=True)
first_name = db.Column(db.String(60))
patronymic = db.Column(db.String(60))
last_name = db.Column(db.String(60))
position = db.Column(db.String(60))
office = db.Column(db.String)
birthday = db.Column(db.Date)
sex = db.Column(db.String)
我试图在每个函数后删除会话并删除更改。
conf test.朋友
import os
import pytest
from backend.auth import db as _db
from backend.auth import create_app
from backend.auth.models import User
class TestConfig(object):
TESTING = True
WTF_CSRF_ENABLED = False
SECRET_KEY = os.environ.get('SECRET_KEY', 'JustaSecret')
SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:admin@localhost/FlaskTest'
SQLALCHEMY_TRACK_MODIFICATIONS = False
DEBUG = True
@pytest.yield_fixture(scope='session')
def app(request):
flask_app = create_app(TestConfig)
# Flask provides a way to test your application by exposing the Werkzeug test Client
# and handling the context locals for you.
testing_client = flask_app.test_client()
# Establish an application context before running the tests.
ctx = flask_app.app_context()
ctx.push()
def teardown():
ctx.pop()
request.addfinalizer(teardown)
return testing_client
@pytest.fixture(scope='session')
def db(app, request):
# Create the database and the database table
def teardown():
_db.drop_all()
_db.app = app
_db.create_all()
user = User(username='admin')
user.set_password('12345')
_db.session.add(user)
_db.session.commit()
# Commit the changes for the users
request.addfinalizer(teardown)
return _db
@pytest.fixture(scope='function')
def session(db, request):
"""Creates a new database session for a test."""
# connect to the database
connection = db.engine.connect()
# begin a non-ORM transaction
transaction = connection.begin()
# bind an individual session to the connection
options = dict(bind=connection, binds={})
session = db.create_scoped_session(options=options)
# overload the default session with the session above
db.session = session
def teardown():
session.close()
# rollback - everything that happened with the
# session above (including calls to commit())
# is rolled back.
transaction.rollback()
# return connection to the Engine
connection.close()
session.remove()
request.addfinalizer(teardown)
return session
但是，如果我尝试多次呼叫会话。
test_employee.朋友
class TestEmployee:
def test_create_employee_get(self, app, session):
app.post('/login', data=dict(username='admin', password='12345'),
content_type='application/x-www-form-urlencoded', follow_redirects=True)
response = app.get('/create')
assert response.status_code == 200
assert b'Create Employee' in response.data
assert b'first name' in response.data
assert b'Date of birth' in response.data
assert b'sex' in response.data
def test_create_employee_post(self, app, session):
app.post('/login', data=dict(username='admin', password='12345'),
content_type='application/x-www-form-urlencoded', follow_redirects=True)
response = app.post('/create', data=dict(
first_name='Another', patronymic='Test', last_name='Just',
position='Check', office='Москва', birthday='1990-01-01', sex='man'),
content_type='application/x-www-form-urlencoded', follow_redirects=True)
assert response.status_code == 200
每个函数后都不会删除数据库。
____________________________________________________________________ ERROR at setup of TestEmployee.test_create_employee_get ____________________________________________________________________
[gw2] darwin -- Python 3.6.6 /Users/vladimir/UsetechTelegramBot/venv/bin/python
self = <sqlalchemy.engine.base.Connection object at 0x106737b70>
dialect = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x10671ac50>
constructor = <bound method DefaultExecutionContext._init_compiled of <class 'sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2'>>
statement = 'INSERT INTO "user" (username, password_hash) VALUES (%(username)s, %(password_hash)s) RETURNING "user".id'
parameters = {'password_hash': 'pbkdf2:sha256:50000$Ek5K9otp$eea50131bd908bfb87427fe975f0f04e2e6607dd002919da77d1ad322a95078d', 'username': 'admin'}
args = (<sqlalchemy.dialects.postgresql.psycopg2.PGCompiler_psycopg2 object at 0x106737860>, [{'password_hash': 'pbkdf2:sha256:50000$Ek5K9otp$eea50131bd908bfb87427fe975f0f04e2e6607dd002919da77d1ad322a95078d', 'username': 'admin'}])
conn = <sqlalchemy.pool._ConnectionFairy object at 0x10677b4a8>
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x106737898>
def _execute_context(self, dialect, constructor,
statement, parameters,
*args):
"""Create an :class:`.ExecutionContext` and execute, returning
a :class:`.ResultProxy`."""
try:
try:
conn = self.__connection
except AttributeError:
# escape "except AttributeError" before revalidating
# to prevent misleading stacktraces in Py3K
conn = None
if conn is None:
conn = self._revalidate_connection()
context = constructor(dialect, self, conn, *args)
except BaseException as e:
self._handle_dbapi_exception(
e,
util.text_type(statement), parameters,
None, None)
if context.compiled:
context.pre_exec()
cursor, statement, parameters = context.cursor, \
context.statement, \
context.parameters
if not context.executemany:
parameters = parameters[0]
if self._has_events or self.engine._has_events:
for fn in self.dispatch.before_cursor_execute:
statement, parameters = \
fn(self, cursor, statement, parameters,
context, context.executemany)
if self._echo:
self.engine.logger.info(statement)
self.engine.logger.info(
"%r",
sql_util._repr_params(parameters, batches=10)
)
evt_handled = False
try:
if context.executemany:
if self.dialect._has_events:
for fn in self.dialect.dispatch.do_executemany:
if fn(cursor, statement, parameters, context):
evt_handled = True
break
if not evt_handled:
self.dialect.do_executemany(
cursor,
statement,
parameters,
context)
elif not parameters and context.no_parameters:
if self.dialect._has_events:
for fn in self.dialect.dispatch.do_execute_no_params:
if fn(cursor, statement, context):
evt_handled = True
break
if not evt_handled:
self.dialect.do_execute_no_params(
cursor,
statement,
context)
else:
if self.dialect._has_events:
for fn in self.dialect.dispatch.do_execute:
if fn(cursor, statement, parameters, context):
evt_handled = True
break
if not evt_handled:
self.dialect.do_execute(
cursor,
statement,
parameters,
> context)
venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py:1193:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x10671ac50>
cursor = <cursor object at 0x1065deed0; closed: -1>
statement = 'INSERT INTO "user" (username, password_hash) VALUES (%(username)s, %(password_hash)s) RETURNING "user".id'
parameters = {'password_hash': 'pbkdf2:sha256:50000$Ek5K9otp$eea50131bd908bfb87427fe975f0f04e2e6607dd002919da77d1ad322a95078d', 'username': 'admin'}
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x106737898>
def do_execute(self, cursor, statement, parameters, context=None):
> cursor.execute(statement, parameters)
E psycopg2.IntegrityError: duplicate key value violates unique constraint "ix_user_username"
E DETAIL: Key (username)=(admin) already exists.
venv/lib/python3.6/site-packages/sqlalchemy/engine/default.py:509: IntegrityError
The above exception was the direct cause of the following exception:
app = <FlaskClient <Flask 'backend.auth'>>
request = <SubRequest 'db' for <Function 'test_create_employee_get'>>
@pytest.fixture(scope='session')
def db(app, request):
# Create the database and the database table
def teardown():
_db.drop_all()
_db.app = app
_db.create_all()
user = User(username='admin')
user.set_password('12345')
_db.session.add(user)
> _db.session.commit()
tests/conftest.py:64:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv/lib/python3.6/site-packages/sqlalchemy/orm/scoping.py:153: in do
return getattr(self.registry(), name)(*args, **kwargs)
venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py:943: in commit
self.transaction.commit()
venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py:467: in commit
self._prepare_impl()
venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py:447: in _prepare_impl
self.session.flush()
venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py:2254: in flush
self._flush(objects)
venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py:2380: in _flush
transaction.rollback(_capture_exception=True)
venv/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py:66: in __exit__
compat.reraise(exc_type, exc_value, exc_tb)
venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py:249: in reraise
raise value
venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py:2344: in _flush
flush_context.execute()
venv/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py:391: in execute
rec.execute(self)
venv/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py:556: in execute
uow
venv/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py:181: in save_obj
mapper, table, insert)
venv/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py:866: in _emit_insert_statements
execute(statement, params)
venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py:948: in execute
return meth(self, multiparams, params)
venv/lib/python3.6/site-packages/sqlalchemy/sql/elements.py:269: in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py:1060: in _execute_clauseelement
compiled_sql, distilled_params
venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py:1200: in _execute_context
context)
venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py:1413: in _handle_dbapi_exception
exc_info
venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py:265: in raise_from_cause
reraise(type(exception), exception, tb=exc_tb, cause=cause)
venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py:248: in reraise
raise value.with_traceback(tb)
venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py:1193: in _execute_context
context)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x10671ac50>
cursor = <cursor object at 0x1065deed0; closed: -1>
statement = 'INSERT INTO "user" (username, password_hash) VALUES (%(username)s, %(password_hash)s) RETURNING "user".id'
parameters = {'password_hash': 'pbkdf2:sha256:50000$Ek5K9otp$eea50131bd908bfb87427fe975f0f04e2e6607dd002919da77d1ad322a95078d', 'username': 'admin'}
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x106737898>
def do_execute(self, cursor, statement, parameters, context=None):
> cursor.execute(statement, parameters)
E sqlalchemy.exc.IntegrityError: (psycopg2.IntegrityError) duplicate key value violates unique constraint "ix_user_username"
E DETAIL: Key (username)=(admin) already exists.
E [SQL: 'INSERT INTO "user" (username, password_hash) VALUES (%(username)s, %(password_hash)s) RETURNING "user".id'] [parameters: {'username': 'admin', 'password_hash': 'pbkdf2:sha256:50000$Ek5K9otp$eea50131bd908bfb87427fe975f0f04e2e6607dd002919da77d1ad322a95078d'}] (Background on this error at: http://sqlalche.me/e/gkpj)
venv/lib/python3.6/site-packages/sqlalchemy/engine/default.py:509: IntegrityError
更改为此，一切正常。
@pytest.fixture(scope='session')
def db(app):
_db.app = app
_db.create_all()
return _db