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 sqlalchemy+unsubscr...@googlegroups.com. To post to this group, send email to sqlalchemy@googlegroups.com. Visit this group at https://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.