Hello all,
I am working on testing a project that uses SQLalchemy with the pytest
testing framework. I have adapted the code from the docs
<http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites>
to pytest as shown.
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
cursor_handler = logging.FileHandler('cursor.log')
cursor_log = logging.getLogger('cursor_log')
cursor_log.addHandler(cursor_handler)
cursor_log.setLevel(logging.DEBUG)
cursor_log.propagate = false
@pytest.fixture(scope='session')
def created_database_path(tmpdir_factory):
db_path = tmpdir_factory.mktemp('ocspdash').join('ocspdash.db')
engine = create_engine(f'sqlite:///{db_path}')
Base.metadata.create_all(engine)
yield db_path
@pytest.fixture(scope='session')
def manager_session(created_database_path):
engine = create_engine(f'sqlite:///{created_database_path}')
@event.listens_for(engine, 'before_cursor_execute')
def receive_before_cursor_execute(connection, cursor, statement, parameters
, context, executemany):
cursor_log.debug(statement)
cursor_log.debug(parameters)
connection = engine.connect()
session_maker = sessionmaker(bind=connection)
session = scoped_session(session_maker)
@event.listens_for(session, '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() normally does
session.expire_all()
session.begin_nested()
transaction = connection.begin()
session.begin_nested()
manager = Manager(
engine=engine,
session=session,
)
yield manager, connection
session.close()
transaction.rollback()
connection.close()
Pytest fixtures are sort of like context managers: everything before the
`yield` is the set up and after the `yield` is the tear down.
I have separated the `create_all` part because SQLalchemy seems to issue
unnecessary COMMITs in the process of creating the tables:
SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
()
SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
()
PRAGMA table_info("authority")
()
PRAGMA table_info("responder")
()
PRAGMA table_info("chain")
()
PRAGMA table_info("location")
()
PRAGMA table_info("result")
()
CREATE TABLE authority (
id INTEGER NOT NULL,
name VARCHAR(255) NOT NULL,
cardinality INTEGER,
last_updated DATETIME DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (id)
)
()
COMMIT
CREATE INDEX ix_authority_name ON authority (name)
()
COMMIT
These commits would prematurely end the transaction created by
`connection.begin()`.
But here is the real problem. Consider the output in `cursor.log` of one
test function:
PRAGMA table_info("authority")
()
PRAGMA table_info("responder")
()
PRAGMA table_info("chain")
()
PRAGMA table_info("location")
()
PRAGMA table_info("result")
()
SAVEPOINT sa_savepoint_1
()
SELECT authority.id AS authority_id, authority.name AS authority_name,
authority.cardinality AS authority_cardinality, authority.last_updated AS
authority_last_updated
FROM authority
WHERE authority.name = ?
('Test Authority',)
INSERT INTO authority (name, cardinality) VALUES (?, ?)
('Test Authority', 1234)
RELEASE SAVEPOINT sa_savepoint_1
()
SAVEPOINT sa_savepoint_2
()
SELECT authority.id AS authority_id, authority.name AS authority_name,
authority.cardinality AS authority_cardinality, authority.last_updated AS
authority_last_updated
FROM authority
WHERE authority.id = ?
(1,)
SELECT authority.id AS authority_id, authority.name AS authority_name,
authority.cardinality AS authority_cardinality, authority.last_updated AS
authority_last_updated
FROM authority
WHERE authority.name = ?
('Test Authority',)
UPDATE authority SET cardinality=?, last_updated=CURRENT_TIMESTAMP WHERE
authority.id = ?
(2345, 1)
RELEASE SAVEPOINT sa_savepoint_2
()
SAVEPOINT sa_savepoint_3
()
SELECT authority.id AS authority_id, authority.name AS authority_name,
authority.cardinality AS authority_cardinality, authority.last_updated AS
authority_last_updated
FROM authority
WHERE authority.id = ?
(1,)
Comparing this to the output in the log from the `sqalchemy.engine` logger:
*snip*
SAVEPOINT sa_savepoint_3
()
SELECT authority.id AS authority_id, authority.name AS authority_name,
authority.cardinality AS authority_cardinality, authority.last_updated AS
authority_last_updated
FROM authority
WHERE authority.id = ?
(1,)
ROLLBACK
The ROLLBACK is missing in the 'before_cursor_execute' events. Why is this?
Adding an extra `engine.execute('ROLLBACK')` at the end raises an
appropriate error about there being no transaction to end.
However, if I open up the test database:
$ sqlite3 test.db
SQLite version 3.19.3 2017-06-27 16:48:08
Enter ".help" for usage hints.
sqlite> select * from authority;
1|Test Authority|2345|2018-05-28 22:59:28
sqlite>
It's still in there!
So here are my questions:
- why does `engine.create_all` issue unnecessary COMMITs?
- why is the final ROLLBACK not showing up in the
'before_cursor_execute' event?
- what am I doing wrong that the transaction doesn't actually get rolled
back?
(I know that `scoped_session` might not be the most appropriate thing to
use here, but that is how the real code works and I want to test in a
similar environment; additionally, when I tried this whole exercise with a
normal `sessionmaker`, all these problems happened in an identical manner.)
Thanks,
Scott Colby
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
