In a testing suite I have a fixture that drop all the tables in an engine, then start fresh and create all the tables. After this fixture logic, my test case runs, using the newly created table. The fixture and the test case are run in the MainThread, while the database consumer is a web application server run in another thread. However, I keep getting: *sqlite3.OperationalError: no such table: **** I've checked that they are using the same in-memory engine, but different connections(this is correct). And I've checked that the fixture does run before the consumer thread starts running. What could be possible cause?
My code is as below: import os import pytest import cherrypy class DAL: def __init__(self, path="database", filename=None, conn_string=None, echo=False): if filename is None and conn_string is None: conn_string = "sqlite:///:memory:" elif conn_string is not None: conn_string = conn_string else: conn_string = f'sqlite:///{os.path.abspath(path)}/{filename}' self.conn_string = conn_string engine = create_engine(conn_string, echo=echo) Session_Factory = sessionmaker(bind=engine) self.Session = sqlalchemy.orm.scoped_session(Session_Factory) def __str__(self): return f"<DAL>object: {self.conn_string}, at {hex(id(self))}" def get_a_dbsession(self): opened_db = self.Session() return opened_db def __enter__(self): return self.get_a_dbsession() def __exit__(self, exception_type, exception_value, exception_traceback): opened_db = self.Session() try: opened_db.commit() except: opened_db.rollback() raise finally: self.Session.remove() else: opened_db.close() raise exception_type def create_schema(self): SchemaBase.metadata.create_all(self.Session().connection().engine) class SAEnginePlugin(cherrypy.process.plugins.SimplePlugin): def __init__(self, bus, dal): """ The plugin is registered to the CherryPy engine. """ cherrypy.process.plugins.SimplePlugin.__init__(self, bus) self.dal = dal def start(self): self.bus.subscribe("bind-session", self.bind) def stop(self): self.bus.unsubscribe("bind-session", self.bind) if self.dal: del self.dal def bind(self): """ Whenever this plugin receives the 'bind-session' message, it applies this method and bind the received session to the engine. """ # self.dal.Session.configure(bind=self.dal.engine) session = self.dal.get_a_dbsession() return session class SATool(cherrypy.Tool): def __init__(self): """ This tool binds a session to the engine each time a requests starts and commits/rollbacks whenever the request terminates. """ cherrypy.Tool.__init__(self, 'on_start_resource', self.bind_session, priority=20) def _setup(self): cherrypy.Tool._setup(self) cherrypy.request.hooks.attach('on_end_resource', self.close_session, priority=80) def bind_session(self): """ Attaches a session to the request's scope by requesting the SA plugin to bind a session to the SA engine. """ session = cherrypy.engine.publish('bind-session').pop() cherrypy.request.db = session def close_session(self): """ Commits the current transaction or rollbacks if an error occurs. In all cases, the current session is unbound and therefore not usable any longer. """ if not hasattr(cherrypy.request, 'db'): return try: cherrypy.request.db.commit() except: cherrypy.request.db.rollback() raise finally: cherrypy.request.db.close() cherrypy.request.db = None # Register the SQLAlchemy tool cherrypy.tools.db = SATool() class UnitServer: ... @cherrypy.expose @cherrypy.tools.json_in() def list_filtered_entries(self): ... queryOBJ = cherrypy.request.db.query(classmodel_obj) ... ############# main module code below ############: # mocking 'db': dal = database.DAL() # configure cherrypy: SAEnginePlugin(cherrypy.engine, dal).subscribe() @pytest.fixture(autouse=True) # automatically run before every test case def mocked_dal(request): # first, clean the database by dropping all tables: database.SchemaBase.metadata.drop_all(dal.Session().connection().engine) # second, create the schema from blank: dal.create_schema() # third, insert some dummy data record: ... db.commit() class TestMyUnitServer(cherrypy.test.helper.CPWebCase): @staticmethod def setup_server(): ... server_app = UnitServer() cherrypy.tree.mount(server_app, '', {'/': {'tools.db.on': True}}) def test_list_filtered_entries_allentries(self): ... self.getPage('/list_filtered_entries', headers=[("Accept", "application/json"), ('Content-type', 'application/json'), ('Content-Length', str(len(json.dumps(query_params)))), ("Connection", "keep-alive"), ("Cache-Control", "max-age=0")], body=serialized_query_params, method="POST") self.assertStatus('200 OK') -- 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 view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/b21a6648-a568-47ca-bd0a-31883284e0d1n%40googlegroups.com.