summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/testing/plugin/provision.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2014-09-14 21:41:13 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2014-09-14 21:41:13 -0400
commitd7498cf4adb8ef9f35d1efa6dc747ac9eb489e60 (patch)
treede69b7ec88fef69c36704d2a7b06b1a0ffe0f17b /lib/sqlalchemy/testing/plugin/provision.py
parent7fd3f05805a7cf1ff1fa29ffcf375c4eea6627ba (diff)
downloadsqlalchemy-d7498cf4adb8ef9f35d1efa6dc747ac9eb489e60.tar.gz
- remove some crufty old testing options
- reestablish the "bootstrap" system of loading the test runners in testing/plugin; using the updated approach we just came up with for alembic. Coverage should be fixed now when running either py.test or nose. fixes #3196 - upgrade tox.ini and start using a .coveragerc file
Diffstat (limited to 'lib/sqlalchemy/testing/plugin/provision.py')
-rw-r--r--lib/sqlalchemy/testing/plugin/provision.py187
1 files changed, 0 insertions, 187 deletions
diff --git a/lib/sqlalchemy/testing/plugin/provision.py b/lib/sqlalchemy/testing/plugin/provision.py
deleted file mode 100644
index c6b9030f5..000000000
--- a/lib/sqlalchemy/testing/plugin/provision.py
+++ /dev/null
@@ -1,187 +0,0 @@
-from sqlalchemy.engine import url as sa_url
-from sqlalchemy import text
-from sqlalchemy.util import compat
-from .. import config, engines
-import os
-
-
-class register(object):
- def __init__(self):
- self.fns = {}
-
- @classmethod
- def init(cls, fn):
- return register().for_db("*")(fn)
-
- def for_db(self, dbname):
- def decorate(fn):
- self.fns[dbname] = fn
- return self
- return decorate
-
- def __call__(self, cfg, *arg):
- if isinstance(cfg, compat.string_types):
- url = sa_url.make_url(cfg)
- elif isinstance(cfg, sa_url.URL):
- url = cfg
- else:
- url = cfg.db.url
- backend = url.get_backend_name()
- if backend in self.fns:
- return self.fns[backend](cfg, *arg)
- else:
- return self.fns['*'](cfg, *arg)
-
-
-def create_follower_db(follower_ident):
-
- for cfg in _configs_for_db_operation():
- _create_db(cfg, cfg.db, follower_ident)
-
-
-def configure_follower(follower_ident):
- for cfg in config.Config.all_configs():
- _configure_follower(cfg, follower_ident)
-
-
-def setup_config(db_url, db_opts, options, file_config, follower_ident):
- if follower_ident:
- db_url = _follower_url_from_main(db_url, follower_ident)
- eng = engines.testing_engine(db_url, db_opts)
- eng.connect().close()
- cfg = config.Config.register(eng, db_opts, options, file_config)
- if follower_ident:
- _configure_follower(cfg, follower_ident)
- return cfg
-
-
-def drop_follower_db(follower_ident):
- for cfg in _configs_for_db_operation():
- _drop_db(cfg, cfg.db, follower_ident)
-
-
-def _configs_for_db_operation():
- hosts = set()
-
- for cfg in config.Config.all_configs():
- cfg.db.dispose()
-
- for cfg in config.Config.all_configs():
- url = cfg.db.url
- backend = url.get_backend_name()
- host_conf = (
- backend,
- url.username, url.host, url.database)
-
- if host_conf not in hosts:
- yield cfg
- hosts.add(host_conf)
-
- for cfg in config.Config.all_configs():
- cfg.db.dispose()
-
-
-@register.init
-def _create_db(cfg, eng, ident):
- raise NotImplementedError("no DB creation routine for cfg: %s" % eng.url)
-
-
-@register.init
-def _drop_db(cfg, eng, ident):
- raise NotImplementedError("no DB drop routine for cfg: %s" % eng.url)
-
-
-@register.init
-def _configure_follower(cfg, ident):
- pass
-
-
-@register.init
-def _follower_url_from_main(url, ident):
- url = sa_url.make_url(url)
- url.database = ident
- return url
-
-
-@_follower_url_from_main.for_db("sqlite")
-def _sqlite_follower_url_from_main(url, ident):
- url = sa_url.make_url(url)
- if not url.database or url.database == ':memory:':
- return url
- else:
- return sa_url.make_url("sqlite:///%s.db" % ident)
-
-
-@_create_db.for_db("postgresql")
-def _pg_create_db(cfg, eng, ident):
- with eng.connect().execution_options(
- isolation_level="AUTOCOMMIT") as conn:
- try:
- _pg_drop_db(cfg, conn, ident)
- except:
- pass
- currentdb = conn.scalar("select current_database()")
- conn.execute("CREATE DATABASE %s TEMPLATE %s" % (ident, currentdb))
-
-
-@_create_db.for_db("mysql")
-def _mysql_create_db(cfg, eng, ident):
- with eng.connect() as conn:
- try:
- _mysql_drop_db(cfg, conn, ident)
- except:
- pass
- conn.execute("CREATE DATABASE %s" % ident)
- conn.execute("CREATE DATABASE %s_test_schema" % ident)
- conn.execute("CREATE DATABASE %s_test_schema_2" % ident)
-
-
-@_configure_follower.for_db("mysql")
-def _mysql_configure_follower(config, ident):
- config.test_schema = "%s_test_schema" % ident
- config.test_schema_2 = "%s_test_schema_2" % ident
-
-
-@_create_db.for_db("sqlite")
-def _sqlite_create_db(cfg, eng, ident):
- pass
-
-
-@_drop_db.for_db("postgresql")
-def _pg_drop_db(cfg, eng, ident):
- with eng.connect().execution_options(
- isolation_level="AUTOCOMMIT") as conn:
- conn.execute(
- text(
- "select pg_terminate_backend(pid) from pg_stat_activity "
- "where usename=current_user and pid != pg_backend_pid() "
- "and datname=:dname"
- ), dname=ident)
- conn.execute("DROP DATABASE %s" % ident)
-
-
-@_drop_db.for_db("sqlite")
-def _sqlite_drop_db(cfg, eng, ident):
- pass
- #os.remove("%s.db" % ident)
-
-
-@_drop_db.for_db("mysql")
-def _mysql_drop_db(cfg, eng, ident):
- with eng.connect() as conn:
- try:
- conn.execute("DROP DATABASE %s_test_schema" % ident)
- except:
- pass
- try:
- conn.execute("DROP DATABASE %s_test_schema_2" % ident)
- except:
- pass
- try:
- conn.execute("DROP DATABASE %s" % ident)
- except:
- pass
-
-
-
-