summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2016-02-09 13:40:14 -0500
committerMike Bayer <mike_mp@zzzcomputing.com>2016-02-09 13:40:14 -0500
commit46f19a10c1e94696357786c7b7765790c023a0fd (patch)
tree1537542ccb6c4250f13e86688fc138fac91ab1a3
parent2ce9175d28ce094d8214bef1d7d8d75843f35237 (diff)
downloadsqlalchemy-46f19a10c1e94696357786c7b7765790c023a0fd.tar.gz
- backport changes to SQLite attached DB tests so that
we dont get file conflicts, cherry-pick from 5bb2536cc57c55c7d8c5901b5b622d18a9a6c646
-rw-r--r--lib/sqlalchemy/testing/provision.py30
-rw-r--r--test/dialect/test_sqlite.py38
2 files changed, 38 insertions, 30 deletions
diff --git a/lib/sqlalchemy/testing/provision.py b/lib/sqlalchemy/testing/provision.py
index b928d006d..52b0dee1a 100644
--- a/lib/sqlalchemy/testing/provision.py
+++ b/lib/sqlalchemy/testing/provision.py
@@ -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:
@@ -109,6 +111,11 @@ def _configure_follower(cfg, ident):
@register.init
+def _post_configure_engine(url, engine):
+ pass
+
+
+@register.init
def _follower_url_from_main(url, ident):
url = sa_url.make_url(url)
url.database = ident
@@ -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")
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index 17920c127..f02b4cfe2 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -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."""