1import os 2 3from ...engine import url as sa_url 4from ...testing.provision import create_db 5from ...testing.provision import drop_db 6from ...testing.provision import follower_url_from_main 7from ...testing.provision import log 8from ...testing.provision import post_configure_engine 9from ...testing.provision import run_reap_dbs 10from ...testing.provision import temp_table_keyword_args 11 12 13@follower_url_from_main.for_db("sqlite") 14def _sqlite_follower_url_from_main(url, ident): 15 url = sa_url.make_url(url) 16 if not url.database or url.database == ":memory:": 17 return url 18 else: 19 return sa_url.make_url("sqlite:///%s.db" % ident) 20 21 22@post_configure_engine.for_db("sqlite") 23def _sqlite_post_configure_engine(url, engine, follower_ident): 24 from sqlalchemy import event 25 26 @event.listens_for(engine, "connect") 27 def connect(dbapi_connection, connection_record): 28 # use file DBs in all cases, memory acts kind of strangely 29 # as an attached 30 if not follower_ident: 31 # note this test_schema.db gets created for all test runs. 32 # there's not any dedicated cleanup step for it. it in some 33 # ways corresponds to the "test.test_schema" schema that's 34 # expected to be already present, so for now it just stays 35 # in a given checkout directory. 36 dbapi_connection.execute( 37 'ATTACH DATABASE "test_schema.db" AS test_schema' 38 ) 39 else: 40 dbapi_connection.execute( 41 'ATTACH DATABASE "%s_test_schema.db" AS test_schema' 42 % follower_ident 43 ) 44 45 46@create_db.for_db("sqlite") 47def _sqlite_create_db(cfg, eng, ident): 48 pass 49 50 51@drop_db.for_db("sqlite") 52def _sqlite_drop_db(cfg, eng, ident): 53 for path in ["%s.db" % ident, "%s_test_schema.db" % ident]: 54 if os.path.exists(path): 55 log.info("deleting SQLite database file: %s" % path) 56 os.remove(path) 57 58 59@temp_table_keyword_args.for_db("sqlite") 60def _sqlite_temp_table_keyword_args(cfg, eng): 61 return {"prefixes": ["TEMPORARY"]} 62 63 64@run_reap_dbs.for_db("sqlite") 65def _reap_sqlite_dbs(url, idents): 66 log.info("db reaper connecting to %r", url) 67 68 log.info("identifiers in file: %s", ", ".join(idents)) 69 for ident in idents: 70 # we don't have a config so we can't call _sqlite_drop_db due to the 71 # decorator 72 for path in ["%s.db" % ident, "%s_test_schema.db" % ident]: 73 if os.path.exists(path): 74 log.info("deleting SQLite database file: %s" % path) 75 os.remove(path) 76