]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
- backport changes to SQLite attached DB tests so that
authorMike Bayer <mike_mp@zzzcomputing.com>
Tue, 9 Feb 2016 18:40:14 +0000 (13:40 -0500)
committerMike Bayer <mike_mp@zzzcomputing.com>
Tue, 9 Feb 2016 18:40:14 +0000 (13:40 -0500)
we dont get file conflicts, cherry-pick from 5bb2536cc57c55c7d8c5901b5b622d18a9a6c646

lib/sqlalchemy/testing/provision.py
test/dialect/test_sqlite.py

index b928d006d3d077dba5705a0e49a3f89bbf70befc..52b0dee1a8e870245d8823cb5e821a4d274126ad 100644 (file)
@@ -5,6 +5,7 @@ from sqlalchemy.util import compat
 from . import config, engines
 import time
 import logging
+import os
 log = logging.getLogger(__name__)
 
 FOLLOWER_IDENT = None
@@ -55,6 +56,7 @@ def setup_config(db_url, options, file_config, follower_ident):
     db_opts = {}
     _update_db_opts(db_url, db_opts)
     eng = engines.testing_engine(db_url, db_opts)
+    _post_configure_engine(db_url, eng, follower_ident)
     eng.connect().close()
     cfg = config.Config.register(eng, db_opts, options, file_config)
     if follower_ident:
@@ -108,6 +110,11 @@ def _configure_follower(cfg, ident):
     pass
 
 
+@register.init
+def _post_configure_engine(url, engine):
+    pass
+
+
 @register.init
 def _follower_url_from_main(url, ident):
     url = sa_url.make_url(url)
@@ -129,6 +136,23 @@ def _sqlite_follower_url_from_main(url, ident):
         return sa_url.make_url("sqlite:///%s.db" % ident)
 
 
+@_post_configure_engine.for_db("sqlite")
+def _sqlite_post_configure_engine(url, engine, follower_ident):
+    from sqlalchemy import event
+
+    @event.listens_for(engine, "connect")
+    def connect(dbapi_connection, connection_record):
+        # use file DBs in all cases, memory acts kind of strangely
+        # as an attached
+        if not follower_ident:
+            dbapi_connection.execute(
+                'ATTACH DATABASE "test_schema.db" AS test_schema')
+        else:
+            dbapi_connection.execute(
+                'ATTACH DATABASE "%s_test_schema.db" AS test_schema'
+                % follower_ident)
+
+
 @_create_db.for_db("postgresql")
 def _pg_create_db(cfg, eng, ident):
     with eng.connect().execution_options(
@@ -190,8 +214,10 @@ def _pg_drop_db(cfg, eng, ident):
 
 @_drop_db.for_db("sqlite")
 def _sqlite_drop_db(cfg, eng, ident):
-    pass
-    #os.remove("%s.db" % ident)
+    if ident:
+        os.remove("%s_test_schema.db" % ident)
+    else:
+        os.remove("%s.db" % ident)
 
 
 @_drop_db.for_db("mysql")
index 17920c1272fef9cc3336413234493a2ba1e96396..f02b4cfe219acdbc80299d1fc0edb2633048e11d 100644 (file)
@@ -535,29 +535,12 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults):
         assert e.pool.__class__ is pool.NullPool
 
 
-
-class AttachedMemoryDBTest(fixtures.TestBase):
+class AttachedDBTest(fixtures.TestBase):
     __only_on__ = 'sqlite'
 
-    dbname = None
-
-    def setUp(self):
-        self.conn = conn = testing.db.connect()
-        if self.dbname is None:
-            dbname = ':memory:'
-        else:
-            dbname = self.dbname
-        conn.execute('ATTACH DATABASE "%s" AS  test_schema' % dbname)
-        self.metadata = MetaData()
-
-    def tearDown(self):
-        self.metadata.drop_all(self.conn)
-        self.conn.execute('DETACH DATABASE test_schema')
-        if self.dbname:
-            os.remove(self.dbname)
-
     def _fixture(self):
         meta = self.metadata
+        self.conn = testing.db.connect()
         ct = Table(
             'created', meta,
             Column('id', Integer),
@@ -567,6 +550,14 @@ class AttachedMemoryDBTest(fixtures.TestBase):
         meta.create_all(self.conn)
         return ct
 
+    def setup(self):
+        self.conn = testing.db.connect()
+        self.metadata = MetaData()
+
+    def teardown(self):
+        self.metadata.drop_all(self.conn)
+        self.conn.close()
+
     def test_no_tables(self):
         insp = inspect(self.conn)
         eq_(insp.get_table_names("test_schema"), [])
@@ -576,11 +567,6 @@ class AttachedMemoryDBTest(fixtures.TestBase):
         insp = inspect(self.conn)
         eq_(insp.get_table_names("test_schema"), ["created"])
 
-    def test_table_names_system(self):
-        self._fixture()
-        insp = inspect(self.conn)
-        eq_(insp.get_table_names("test_schema"), ["created"])
-
     def test_reflect_system_table(self):
         meta = MetaData(self.conn)
         alt_master = Table(
@@ -633,10 +619,6 @@ class AttachedMemoryDBTest(fixtures.TestBase):
         eq_(row['name'], 'foo')
 
 
-class AttachedFileDBTest(AttachedMemoryDBTest):
-    dbname = 'attached_db.db'
-
-
 class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
 
     """Tests SQLite-dialect specific compilation."""