from sqlalchemy.sql.expression import _BindParamClause
from sqlalchemy.ext.compiler import compiles
from sqlalchemy import schema, text, sql
-from sqlalchemy.sql import expression
from sqlalchemy import types as sqltypes
from ..compat import string_types, text_type, with_metaclass
from .. import util
from . import base
+if util.sqla_08:
+ from sqlalchemy.sql.expression import TextClause
+else:
+ from sqlalchemy.sql.expression import _TextClause as TextClause
+
class ImplMeta(type):
c = schema.Column(text_, sqltypes.NULLTYPE)
table.append_column(c)
return c
- elif isinstance(text_, expression.TextClause):
+ elif isinstance(text_, TextClause):
return _textual_index_element(table, text_)
else:
raise ValueError("String or text() construct expected")
tname = "%s.%s" % (referent_schema, referent) if referent_schema \
else referent
+
+ if util.sqla_08:
+ # "match" kw unsupported in 0.7
+ dialect_kw['match'] = match
+
f = sa_schema.ForeignKeyConstraint(local_cols,
["%s.%s" % (tname, n)
for n in remote_cols],
ondelete=ondelete,
deferrable=deferrable,
initially=initially,
- match=match,
**dialect_kw
)
t1.append_constraint(f)
from .assertions import eq_, ne_, is_, assert_raises_message, \
eq_ignore_whitespace, assert_raises
-from sqlalchemy.testing import config
from alembic import util
-if not util.sqla_100:
- config.test_schema = "test_schema"
-from sqlalchemy.testing.config import requirements as requires
+from .config import requirements as requires
import re
+from alembic import util
from sqlalchemy.engine import default
-from sqlalchemy.testing.assertions import eq_, ne_, is_, \
- assert_raises_message, assert_raises
from alembic.compat import text_type
+if not util.sqla_094:
+ def eq_(a, b, msg=None):
+ """Assert a == b, with repr messaging on failure."""
+ assert a == b, msg or "%r != %r" % (a, b)
+
+ def ne_(a, b, msg=None):
+ """Assert a != b, with repr messaging on failure."""
+ assert a != b, msg or "%r == %r" % (a, b)
+
+ def is_(a, b, msg=None):
+ """Assert a is b, with repr messaging on failure."""
+ assert a is b, msg or "%r is not %r" % (a, b)
+
+ def assert_raises(except_cls, callable_, *args, **kw):
+ try:
+ callable_(*args, **kw)
+ success = False
+ except except_cls:
+ success = True
+
+ # assert outside the block so it works for AssertionError too !
+ assert success, "Callable did not raise an exception"
+
+ def assert_raises_message(except_cls, msg, callable_, *args, **kwargs):
+ try:
+ callable_(*args, **kwargs)
+ assert False, "Callable did not raise an exception"
+ except except_cls as e:
+ assert re.search(
+ msg, text_type(e), re.UNICODE), "%r !~ %s" % (msg, e)
+ print(text_type(e).encode('utf-8'))
+
+else:
+ from sqlalchemy.testing.assertions import eq_, ne_, is_, \
+ assert_raises_message, assert_raises
+
def eq_ignore_whitespace(a, b, msg=None):
a = re.sub(r'^\s+?|\n', "", a)
--- /dev/null
+def get_url_driver_name(url):
+ if '+' not in url.drivername:
+ return url.get_dialect().driver
+ else:
+ return url.drivername.split('+')[1]
+
+
+def get_url_backend_name(url):
+ if '+' not in url.drivername:
+ return url.drivername
+ else:
+ return url.drivername.split('+')[0]
+
--- /dev/null
+# testing/config.py
+# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
+ this should be removable when Alembic targets SQLAlchemy 0.9.4.
+"""
+
+import collections
+
+requirements = None
+db = None
+db_url = None
+db_opts = None
+file_config = None
+test_schema = None
+test_schema_2 = None
+_current = None
+
+
+class Config(object):
+ def __init__(self, db, db_opts, options, file_config):
+ self.db = db
+ self.db_opts = db_opts
+ self.options = options
+ self.file_config = file_config
+ self.test_schema = "test_schema"
+ self.test_schema_2 = "test_schema_2"
+
+ _stack = collections.deque()
+ _configs = {}
+
+ @classmethod
+ def register(cls, db, db_opts, options, file_config):
+ """add a config as one of the global configs.
+
+ If there are no configs set up yet, this config also
+ gets set as the "_current".
+ """
+ cfg = Config(db, db_opts, options, file_config)
+
+ cls._configs[cfg.db.name] = cfg
+ cls._configs[(cfg.db.name, cfg.db.dialect)] = cfg
+ cls._configs[cfg.db] = cfg
+ return cfg
+
+ @classmethod
+ def set_as_current(cls, config):
+ global db, _current, db_url, test_schema, test_schema_2, db_opts
+ _current = config
+ db_url = config.db.url
+ db_opts = config.db_opts
+ test_schema = config.test_schema
+ test_schema_2 = config.test_schema_2
+ db = config.db
+
+ @classmethod
+ def push_engine(cls, db):
+ assert _current, "Can't push without a default Config set up"
+ cls.push(
+ Config(
+ db, _current.db_opts, _current.options, _current.file_config)
+ )
+
+ @classmethod
+ def push(cls, config):
+ cls._stack.append(_current)
+ cls.set_as_current(config)
+
+ @classmethod
+ def reset(cls):
+ if cls._stack:
+ cls.set_as_current(cls._stack[0])
+ cls._stack.clear()
+
+ @classmethod
+ def all_configs(cls):
+ for cfg in set(cls._configs.values()):
+ yield cfg
+
+ @classmethod
+ def all_dbs(cls):
+ for cfg in cls.all_configs():
+ yield cfg.db
+
--- /dev/null
+# testing/engines.py
+# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
+ this should be removable when Alembic targets SQLAlchemy 0.9.4.
+"""
+
+from __future__ import absolute_import
+
+from . import config
+
+
+def testing_engine(url=None, options=None):
+ """Produce an engine configured by --options with optional overrides."""
+
+ from sqlalchemy import create_engine
+
+ url = url or config.db.url
+ if options is None:
+ options = config.db_opts
+
+ engine = create_engine(url, **options)
+
+ return engine
+
#!coding: utf-8
-import io
import os
-import re
import shutil
import textwrap
from alembic.compat import u
from alembic.script import Script, ScriptDirectory
from alembic import util
+from . import engines
+from alembic.testing.plugin import plugin_base
-staging_directory = 'scratch'
-files_directory = 'files'
+
+def _get_staging_directory():
+ if plugin_base.FOLLOWER_IDENT:
+ return "scratch_%s" % plugin_base.FOLLOWER_IDENT
+ else:
+ return 'scratch'
def staging_env(create=True, template="generic", sourceless=False):
from alembic import command, script
cfg = _testing_config()
if create:
- path = os.path.join(staging_directory, 'scripts')
+ path = os.path.join(_get_staging_directory(), 'scripts')
if os.path.exists(path):
shutil.rmtree(path)
command.init(cfg, path)
def clear_staging_env():
- shutil.rmtree(staging_directory, True)
+ shutil.rmtree(_get_staging_directory(), True)
def script_file_fixture(txt):
- dir_ = os.path.join(staging_directory, 'scripts')
+ dir_ = os.path.join(_get_staging_directory(), 'scripts')
path = os.path.join(dir_, "script.py.mako")
with open(path, 'w') as f:
f.write(txt)
def env_file_fixture(txt):
- dir_ = os.path.join(staging_directory, 'scripts')
+ dir_ = os.path.join(_get_staging_directory(), 'scripts')
txt = """
from alembic import context
def _sqlite_file_db():
- from sqlalchemy.testing import engines
- dir_ = os.path.join(staging_directory, 'scripts')
+ dir_ = os.path.join(_get_staging_directory(), 'scripts')
url = "sqlite:///%s/foo.db" % dir_
return engines.testing_engine(url=url)
def _sqlite_testing_config(sourceless=False):
- dir_ = os.path.join(staging_directory, 'scripts')
+ dir_ = os.path.join(_get_staging_directory(), 'scripts')
url = "sqlite:///%s/foo.db" % dir_
return _write_config_file("""
def _no_sql_testing_config(dialect="postgresql", directives=""):
"""use a postgresql url with no host so that
connections guaranteed to fail"""
- dir_ = os.path.join(staging_directory, 'scripts')
+ dir_ = os.path.join(_get_staging_directory(), 'scripts')
return _write_config_file("""
[alembic]
script_location = %s
def _testing_config():
from alembic.config import Config
- if not os.access(staging_directory, os.F_OK):
- os.mkdir(staging_directory)
- return Config(os.path.join(staging_directory, 'test_alembic.ini'))
+ if not os.access(_get_staging_directory(), os.F_OK):
+ os.mkdir(_get_staging_directory())
+ return Config(os.path.join(_get_staging_directory(), 'test_alembic.ini'))
def write_script(
--- /dev/null
+# testing/exclusions.py
+# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
+ this should be removable when Alembic targets SQLAlchemy 0.9.4.
+"""
+
+
+import operator
+from .plugin.plugin_base import SkipTest
+from sqlalchemy.util import decorator
+from . import config
+from sqlalchemy import util
+from alembic import compat
+import inspect
+import contextlib
+from .compat import get_url_driver_name, get_url_backend_name
+
+def skip_if(predicate, reason=None):
+ rule = compound()
+ pred = _as_predicate(predicate, reason)
+ rule.skips.add(pred)
+ return rule
+
+
+def fails_if(predicate, reason=None):
+ rule = compound()
+ pred = _as_predicate(predicate, reason)
+ rule.fails.add(pred)
+ return rule
+
+
+class compound(object):
+ def __init__(self):
+ self.fails = set()
+ self.skips = set()
+ self.tags = set()
+
+ def __add__(self, other):
+ return self.add(other)
+
+ def add(self, *others):
+ copy = compound()
+ copy.fails.update(self.fails)
+ copy.skips.update(self.skips)
+ copy.tags.update(self.tags)
+ for other in others:
+ copy.fails.update(other.fails)
+ copy.skips.update(other.skips)
+ copy.tags.update(other.tags)
+ return copy
+
+ def not_(self):
+ copy = compound()
+ copy.fails.update(NotPredicate(fail) for fail in self.fails)
+ copy.skips.update(NotPredicate(skip) for skip in self.skips)
+ copy.tags.update(self.tags)
+ return copy
+
+ @property
+ def enabled(self):
+ return self.enabled_for_config(config._current)
+
+ def enabled_for_config(self, config):
+ for predicate in self.skips.union(self.fails):
+ if predicate(config):
+ return False
+ else:
+ return True
+
+ def matching_config_reasons(self, config):
+ return [
+ predicate._as_string(config) for predicate
+ in self.skips.union(self.fails)
+ if predicate(config)
+ ]
+
+ def include_test(self, include_tags, exclude_tags):
+ return bool(
+ not self.tags.intersection(exclude_tags) and
+ (not include_tags or self.tags.intersection(include_tags))
+ )
+
+ def _extend(self, other):
+ self.skips.update(other.skips)
+ self.fails.update(other.fails)
+ self.tags.update(other.tags)
+
+ def __call__(self, fn):
+ if hasattr(fn, '_sa_exclusion_extend'):
+ fn._sa_exclusion_extend._extend(self)
+ return fn
+
+ @decorator
+ def decorate(fn, *args, **kw):
+ return self._do(config._current, fn, *args, **kw)
+ decorated = decorate(fn)
+ decorated._sa_exclusion_extend = self
+ return decorated
+
+ @contextlib.contextmanager
+ def fail_if(self):
+ all_fails = compound()
+ all_fails.fails.update(self.skips.union(self.fails))
+
+ try:
+ yield
+ except Exception as ex:
+ all_fails._expect_failure(config._current, ex)
+ else:
+ all_fails._expect_success(config._current)
+
+ def _do(self, config, fn, *args, **kw):
+ for skip in self.skips:
+ if skip(config):
+ msg = "'%s' : %s" % (
+ fn.__name__,
+ skip._as_string(config)
+ )
+ raise SkipTest(msg)
+
+ try:
+ return_value = fn(*args, **kw)
+ except Exception as ex:
+ self._expect_failure(config, ex, name=fn.__name__)
+ else:
+ self._expect_success(config, name=fn.__name__)
+ return return_value
+
+ def _expect_failure(self, config, ex, name='block'):
+ for fail in self.fails:
+ if fail(config):
+ print(("%s failed as expected (%s): %s " % (
+ name, fail._as_string(config), str(ex))))
+ break
+ else:
+ raise ex
+
+ def _expect_success(self, config, name='block'):
+ if not self.fails:
+ return
+ for fail in self.fails:
+ if not fail(config):
+ break
+ else:
+ raise AssertionError(
+ "Unexpected success for '%s' (%s)" %
+ (
+ name,
+ " and ".join(
+ fail._as_string(config)
+ for fail in self.fails
+ )
+ )
+ )
+
+
+def requires_tag(tagname):
+ return tags([tagname])
+
+
+def tags(tagnames):
+ comp = compound()
+ comp.tags.update(tagnames)
+ return comp
+
+
+def only_if(predicate, reason=None):
+ predicate = _as_predicate(predicate)
+ return skip_if(NotPredicate(predicate), reason)
+
+
+def succeeds_if(predicate, reason=None):
+ predicate = _as_predicate(predicate)
+ return fails_if(NotPredicate(predicate), reason)
+
+
+class Predicate(object):
+ @classmethod
+ def as_predicate(cls, predicate, description=None):
+ if isinstance(predicate, compound):
+ return cls.as_predicate(predicate.fails.union(predicate.skips))
+
+ elif isinstance(predicate, Predicate):
+ if description and predicate.description is None:
+ predicate.description = description
+ return predicate
+ elif isinstance(predicate, (list, set)):
+ return OrPredicate(
+ [cls.as_predicate(pred) for pred in predicate],
+ description)
+ elif isinstance(predicate, tuple):
+ return SpecPredicate(*predicate)
+ elif isinstance(predicate, compat.string_types):
+ tokens = predicate.split(" ", 2)
+ op = spec = None
+ db = tokens.pop(0)
+ if tokens:
+ op = tokens.pop(0)
+ if tokens:
+ spec = tuple(int(d) for d in tokens.pop(0).split("."))
+ return SpecPredicate(db, op, spec, description=description)
+ elif util.callable(predicate):
+ return LambdaPredicate(predicate, description)
+ else:
+ assert False, "unknown predicate type: %s" % predicate
+
+ def _format_description(self, config, negate=False):
+ bool_ = self(config)
+ if negate:
+ bool_ = not negate
+ return self.description % {
+ "driver": get_url_driver_name(config.db.url),
+ "database": get_url_backend_name(config.db.url),
+ "doesnt_support": "doesn't support" if bool_ else "does support",
+ "does_support": "does support" if bool_ else "doesn't support"
+ }
+
+ def _as_string(self, config=None, negate=False):
+ raise NotImplementedError()
+
+
+class BooleanPredicate(Predicate):
+ def __init__(self, value, description=None):
+ self.value = value
+ self.description = description or "boolean %s" % value
+
+ def __call__(self, config):
+ return self.value
+
+ def _as_string(self, config, negate=False):
+ return self._format_description(config, negate=negate)
+
+
+class SpecPredicate(Predicate):
+ def __init__(self, db, op=None, spec=None, description=None):
+ self.db = db
+ self.op = op
+ self.spec = spec
+ self.description = description
+
+ _ops = {
+ '<': operator.lt,
+ '>': operator.gt,
+ '==': operator.eq,
+ '!=': operator.ne,
+ '<=': operator.le,
+ '>=': operator.ge,
+ 'in': operator.contains,
+ 'between': lambda val, pair: val >= pair[0] and val <= pair[1],
+ }
+
+ def __call__(self, config):
+ engine = config.db
+
+ if "+" in self.db:
+ dialect, driver = self.db.split('+')
+ else:
+ dialect, driver = self.db, None
+
+ if dialect and engine.name != dialect:
+ return False
+ if driver is not None and engine.driver != driver:
+ return False
+
+ if self.op is not None:
+ assert driver is None, "DBAPI version specs not supported yet"
+
+ version = _server_version(engine)
+ oper = hasattr(self.op, '__call__') and self.op \
+ or self._ops[self.op]
+ return oper(version, self.spec)
+ else:
+ return True
+
+ def _as_string(self, config, negate=False):
+ if self.description is not None:
+ return self._format_description(config)
+ elif self.op is None:
+ if negate:
+ return "not %s" % self.db
+ else:
+ return "%s" % self.db
+ else:
+ if negate:
+ return "not %s %s %s" % (
+ self.db,
+ self.op,
+ self.spec
+ )
+ else:
+ return "%s %s %s" % (
+ self.db,
+ self.op,
+ self.spec
+ )
+
+
+class LambdaPredicate(Predicate):
+ def __init__(self, lambda_, description=None, args=None, kw=None):
+ spec = inspect.getargspec(lambda_)
+ if not spec[0]:
+ self.lambda_ = lambda db: lambda_()
+ else:
+ self.lambda_ = lambda_
+ self.args = args or ()
+ self.kw = kw or {}
+ if description:
+ self.description = description
+ elif lambda_.__doc__:
+ self.description = lambda_.__doc__
+ else:
+ self.description = "custom function"
+
+ def __call__(self, config):
+ return self.lambda_(config)
+
+ def _as_string(self, config, negate=False):
+ return self._format_description(config)
+
+
+class NotPredicate(Predicate):
+ def __init__(self, predicate, description=None):
+ self.predicate = predicate
+ self.description = description
+
+ def __call__(self, config):
+ return not self.predicate(config)
+
+ def _as_string(self, config, negate=False):
+ if self.description:
+ return self._format_description(config, not negate)
+ else:
+ return self.predicate._as_string(config, not negate)
+
+
+class OrPredicate(Predicate):
+ def __init__(self, predicates, description=None):
+ self.predicates = predicates
+ self.description = description
+
+ def __call__(self, config):
+ for pred in self.predicates:
+ if pred(config):
+ return True
+ return False
+
+ def _eval_str(self, config, negate=False):
+ if negate:
+ conjunction = " and "
+ else:
+ conjunction = " or "
+ return conjunction.join(p._as_string(config, negate=negate)
+ for p in self.predicates)
+
+ def _negation_str(self, config):
+ if self.description is not None:
+ return "Not " + self._format_description(config)
+ else:
+ return self._eval_str(config, negate=True)
+
+ def _as_string(self, config, negate=False):
+ if negate:
+ return self._negation_str(config)
+ else:
+ if self.description is not None:
+ return self._format_description(config)
+ else:
+ return self._eval_str(config)
+
+
+_as_predicate = Predicate.as_predicate
+
+
+def _is_excluded(db, op, spec):
+ return SpecPredicate(db, op, spec)(config._current)
+
+
+def _server_version(engine):
+ """Return a server_version_info tuple."""
+
+ # force metadata to be retrieved
+ conn = engine.connect()
+ version = getattr(engine.dialect, 'server_version_info', ())
+ conn.close()
+ return version
+
+
+def db_spec(*dbs):
+ return OrPredicate(
+ [Predicate.as_predicate(db) for db in dbs]
+ )
+
+
+def open():
+ return skip_if(BooleanPredicate(False, "mark as execute"))
+
+
+def closed():
+ return skip_if(BooleanPredicate(True, "marked as skip"))
+
+
+def fails():
+ return fails_if(BooleanPredicate(True, "expected to fail"))
+
+
+@decorator
+def future(fn, *arg):
+ return fails_if(LambdaPredicate(fn), "Future feature")
+
+
+def fails_on(db, reason=None):
+ return fails_if(SpecPredicate(db), reason)
+
+
+def fails_on_everything_except(*dbs):
+ return succeeds_if(
+ OrPredicate([
+ SpecPredicate(db) for db in dbs
+ ])
+ )
+
+
+def skip(db, reason=None):
+ return skip_if(SpecPredicate(db), reason)
+
+
+def only_on(dbs, reason=None):
+ return only_if(
+ OrPredicate([SpecPredicate(db) for db in util.to_list(dbs)])
+ )
+
+
+def exclude(db, op, spec, reason=None):
+ return skip_if(SpecPredicate(db, op, spec), reason)
+
+
+def against(config, *queries):
+ assert queries, "no queries sent!"
+ return OrPredicate([
+ Predicate.as_predicate(query)
+ for query in queries
+ ])(config)
from alembic.operations import Operations
from alembic.ddl.impl import _impls
from contextlib import contextmanager
-
-from sqlalchemy.testing.fixtures import TestBase
+from .plugin.plugin_base import SkipTest
from .assertions import _get_dialect, eq_
from . import mock
testing_config.read(['test.cfg'])
+if not util.sqla_094:
+ class TestBase(object):
+ # A sequence of database names to always run, regardless of the
+ # constraints below.
+ __whitelist__ = ()
+
+ # A sequence of requirement names matching testing.requires decorators
+ __requires__ = ()
+
+ # A sequence of dialect names to exclude from the test class.
+ __unsupported_on__ = ()
+
+ # If present, test class is only runnable for the *single* specified
+ # dialect. If you need multiple, use __unsupported_on__ and invert.
+ __only_on__ = None
+
+ # A sequence of no-arg callables. If any are True, the entire testcase is
+ # skipped.
+ __skip_if__ = None
+
+ def assert_(self, val, msg=None):
+ assert val, msg
+
+ # apparently a handful of tests are doing this....OK
+ def setup(self):
+ if hasattr(self, "setUp"):
+ self.setUp()
+
+ def teardown(self):
+ if hasattr(self, "tearDown"):
+ self.tearDown()
+else:
+ from sqlalchemy.testing.fixtures import TestBase
+
+
def capture_db():
buf = []
-from __future__ import absolute_import
-
-from sqlalchemy.testing import mock
-from sqlalchemy.testing.mock import Mock, call
+# testing/mock.py
+# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
-from alembic import util, compat
+"""Import stub for mock library.
-if util.sqla_09:
- from sqlalchemy.testing.mock import patch
-elif compat.py33:
- from unittest.mock import patch
-else:
- from mock import patch
+ NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
+ this should be removable when Alembic targets SQLAlchemy 0.9.4.
+"""
+from __future__ import absolute_import
+from alembic.compat import py33
+if py33:
+ from unittest.mock import MagicMock, Mock, call, patch
+else:
+ try:
+ from mock import MagicMock, Mock, call, patch
+ except ImportError:
+ raise ImportError(
+ "SQLAlchemy's test suite requires the "
+ "'mock' library as of 0.8.2.")
-from sqlalchemy.testing.requirements import Requirements
-from sqlalchemy.testing import exclusions
from alembic import util
+from . import exclusions
+
+if util.sqla_094:
+ from sqlalchemy.testing.requirements import Requirements
+else:
+ class Requirements(object):
+ pass
+
class SuiteRequirements(Requirements):
@property
return exclusions.open()
+ @property
+ def unique_constraint_reflection(self):
+ return exclusions.skip_if(
+ lambda config: not util.sqla_084,
+ "SQLAlchemy 0.8.4 or greater required"
+ )
+
+ @property
+ def foreign_key_match(self):
+ return exclusions.fails_if(
+ lambda config: not util.sqla_08,
+ "MATCH for foreign keys added in SQLAlchemy 0.8.0"
+ )
+
+ @property
+ def fail_before_sqla_080(self):
+ return exclusions.fails_if(
+ lambda config: not util.sqla_08,
+ "SQLAlchemy 0.8.0 or greater required"
+ )
+
+ @property
+ def fail_before_sqla_083(self):
+ return exclusions.fails_if(
+ lambda config: not util.sqla_083,
+ "SQLAlchemy 0.8.3 or greater required"
+ )
+
+ @property
+ def fail_before_sqla_084(self):
+ return exclusions.fails_if(
+ lambda config: not util.sqla_084,
+ "SQLAlchemy 0.8.4 or greater required"
+ )
+
@property
def sqlalchemy_08(self):
--- /dev/null
+#!/usr/bin/env python
+# testing/runner.py
+# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""
+Nose test runner module.
+
+This script is a front-end to "nosetests" which
+installs SQLAlchemy's testing plugin into the local environment.
+
+The script is intended to be used by third-party dialects and extensions
+that run within SQLAlchemy's testing framework. The runner can
+be invoked via::
+
+ python -m alembic.testing.runner
+
+The script is then essentially the same as the "nosetests" script, including
+all of the usual Nose options. The test environment requires that a
+setup.cfg is locally present including various required options.
+
+Note that when using this runner, Nose's "coverage" plugin will not be
+able to provide coverage for SQLAlchemy itself, since SQLAlchemy is
+imported into sys.modules before coverage is started. The special
+script sqla_nose.py is provided as a top-level script which loads the
+plugin in a special (somewhat hacky) way so that coverage against
+SQLAlchemy itself is possible.
+
+"""
+
+from alembic.testing.plugin.noseplugin import NoseSQLAlchemy
+
+import nose
+
+
+def main():
+ nose.main(addplugins=[NoseSQLAlchemy()])
+
+
+def setup_py_test():
+ """Runner to use for the 'test_suite' entry of your setup.py.
+
+ Prevents any name clash shenanigans from the command line
+ argument "test" that the "setup.py test" command sends
+ to nose.
+
+ """
+ nose.main(addplugins=[NoseSQLAlchemy()], argv=['runner'])
--- /dev/null
+# testing/warnings.py
+# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
+ this should be removable when Alembic targets SQLAlchemy 0.9.4.
+"""
+
+from __future__ import absolute_import
+
+import warnings
+from sqlalchemy import exc as sa_exc
+import re
+
+
+def setup_filters():
+ """Set global warning behavior for the test suite."""
+
+ warnings.filterwarnings('ignore',
+ category=sa_exc.SAPendingDeprecationWarning)
+ warnings.filterwarnings('error', category=sa_exc.SADeprecationWarning)
+ warnings.filterwarnings('error', category=sa_exc.SAWarning)
+
+
+def assert_warnings(fn, warning_msgs, regex=False):
+ """Assert that each of the given warnings are emitted by fn."""
+
+ from .assertions import eq_
+
+ with warnings.catch_warnings(record=True) as log:
+ # ensure that nothing is going into __warningregistry__
+ warnings.filterwarnings("always")
+
+ result = fn()
+ for warning in log:
+ popwarn = warning_msgs.pop(0)
+ if regex:
+ assert re.match(popwarn, str(warning.message))
+ else:
+ eq_(popwarn, str(warning.message))
+ return result
_vers = tuple(
[_safe_int(x) for x in re.findall(r'(\d+|[abc]\d)', __version__)])
sqla_07 = _vers > (0, 7, 2)
-sqla_08 = sqla_084 = _vers >= (0, 8, 4)
+sqla_08 = _vers >= (0, 8, 0)
+sqla_083 = _vers >= (0, 8, 3)
+sqla_084 = _vers >= (0, 8, 4)
sqla_09 = _vers >= (0, 9, 0)
sqla_092 = _vers >= (0, 9, 2)
sqla_094 = _vers >= (0, 9, 4)
sqla_094 = _vers >= (0, 9, 4)
sqla_100 = _vers >= (1, 0, 0)
-if not sqla_084:
+if not sqla_07:
raise CommandError(
- "SQLAlchemy 0.8.4 or greater is required. ")
+ "SQLAlchemy 0.7.3 or greater is required. ")
from sqlalchemy.util import format_argspec_plus, update_wrapper
from sqlalchemy.util.compat import inspect_getfullargspec
.. change::
:tags: change
- Minimum SQLAlchemy version is now 0.8.4.
+ Minimum SQLAlchemy version is now 0.7.6, however at least
+ 0.8.4 is strongly recommended.
.. changelog::
:version: 0.6.7
-from sqlalchemy.testing import runner
+from alembic.testing import runner
runner.main()
[nosetests]
with-sqla_testing = true
+where = tests
[sqla_testing]
readme = os.path.join(os.path.dirname(__file__), 'README.rst')
requires = [
- 'SQLAlchemy>=0.8.4',
+ 'SQLAlchemy>=0.7.6',
'Mako',
]
for pth in ['../lib']:
sys.path.insert(0, path.join(path.dirname(path.abspath(__file__)), pth))
-from sqlalchemy.testing.plugin.pytestplugin import *
+from alembic.testing.plugin.pytestplugin import *
from alembic.testing.requirements import SuiteRequirements
-from sqlalchemy.testing import exclusions
+from alembic.testing import exclusions
class DefaultRequirements(SuiteRequirements):
+
@property
def schemas(self):
"""Target database must support external schemas, and have one
from sqlalchemy import MetaData, Column, Table, Integer, String, \
Numeric, UniqueConstraint, Index, ForeignKeyConstraint,\
ForeignKey
-from sqlalchemy.testing import engines
+from alembic.testing import engines
from alembic.testing import eq_
from alembic.testing.env import staging_env
class NoUqReflection(object):
+ __requires__ = ()
+
def setUp(self):
staging_env()
self.bind = eng = engines.testing_engine()
raise NotImplementedError()
eng.dialect.get_unique_constraints = unimpl
+ @config.requirements.fail_before_sqla_083
+ def test_add_ix_on_table_create(self):
+ return super(NoUqReflection, self).test_add_ix_on_table_create()
+
+ @config.requirements.fail_before_sqla_080
+ def test_add_idx_non_col(self):
+ return super(NoUqReflection, self).test_add_idx_non_col()
+
class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
reports_unique_constraints = True
+ __requires__ = ('unique_constraint_reflection', )
__only_on__ = 'sqlite'
def test_index_flag_becomes_named_unique_constraint(self):
from alembic.testing.fixtures import capture_context_buffer, op_fixture
from alembic.testing.env import staging_env, _no_sql_testing_config, \
three_rev_fixture, clear_staging_env
+from alembic.testing import config
class FullEnvironmentTests(TestBase):
@classmethod
def setup_class(cls):
- env = staging_env()
+ staging_env()
cls.cfg = cfg = _no_sql_testing_config("mssql")
cls.a, cls.b, cls.c = \
nullable=False)
context.assert_('ALTER TABLE tests ALTER COLUMN col BIT NOT NULL')
+ @config.requirements.fail_before_sqla_084
def test_drop_index(self):
context = op_fixture('mssql')
op.drop_index('my_idx', 'my_table')
- # TODO: annoying that SQLA escapes unconditionally
context.assert_contains("DROP INDEX my_idx ON my_table")
def test_drop_column_w_default(self):
from alembic.testing import eq_, assert_raises_message
from alembic.testing import mock
from alembic.testing.fixtures import TestBase
+from alembic.testing import config
@event.listens_for(Table, "after_parent_attach")
context.assert_(
'CREATE INDEX geocoded ON locations ("IShouldBeQuoted")')
+ @config.requirements.fail_before_sqla_080
def test_create_index_expressions(self):
context = op_fixture()
op.create_index(
context.assert_(
"CREATE INDEX geocoded ON locations (lower(coordinates))")
+ @config.requirements.fail_before_sqla_080
def test_create_index_postgresql_expressions(self):
context = op_fixture("postgresql")
op.create_index(
"REFERENCES t2 (bat, hoho) INITIALLY INITIAL"
)
+ @config.requirements.foreign_key_match
def test_add_foreign_key_match(self):
context = op_fixture()
op.create_foreign_key('fk_test', 't1', 't2',
)
def test_add_foreign_key_dialect_kw(self):
- context = op_fixture()
+ op_fixture()
with mock.patch(
"alembic.operations.sa_schema.ForeignKeyConstraint") as fkc:
op.create_foreign_key('fk_test', 't1', 't2',
['foo', 'bar'], ['bat', 'hoho'],
foobar_arg='xyz')
- eq_(fkc.mock_calls[0],
- mock.call(['foo', 'bar'], ['t2.bat', 't2.hoho'],
- onupdate=None, ondelete=None, name='fk_test',
- foobar_arg='xyz',
- deferrable=None, initially=None, match=None))
+ if config.requirements.foreign_key_match.enabled:
+ eq_(fkc.mock_calls[0],
+ mock.call(['foo', 'bar'], ['t2.bat', 't2.hoho'],
+ onupdate=None, ondelete=None, name='fk_test',
+ foobar_arg='xyz',
+ deferrable=None, initially=None, match=None))
+ else:
+ eq_(fkc.mock_calls[0],
+ mock.call(['foo', 'bar'], ['t2.bat', 't2.hoho'],
+ onupdate=None, ondelete=None, name='fk_test',
+ foobar_arg='xyz',
+ deferrable=None, initially=None))
def test_add_foreign_key_self_referential(self):
context = op_fixture()
op.alter_column("t", "c", new_column_name="x")
context.assert_("ALTER TABLE t RENAME c TO x")
- context = op_fixture('mssql')
- op.drop_index('ik_test', tablename='t1')
- context.assert_("DROP INDEX ik_test ON t1")
-
context = op_fixture('mysql')
op.drop_constraint("f1", "t1", type="foreignkey")
context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
r"Unknown arguments: badarg\d, badarg\d",
op.alter_column, "t", "c", badarg1="x", badarg2="y"
)
+
+ @config.requirements.fail_before_sqla_084
+ def test_naming_changes_drop_idx(self):
+ context = op_fixture('mssql')
+ op.drop_index('ik_test', tablename='t1')
+ context.assert_("DROP INDEX ik_test ON t1")
from alembic.testing.fixtures import TestBase
from alembic.testing import eq_, ne_, is_
from alembic.testing.env import clear_staging_env, staging_env, \
- staging_directory, _no_sql_testing_config, env_file_fixture, \
+ _get_staging_directory, _no_sql_testing_config, env_file_fixture, \
script_file_fixture, _testing_config
from alembic import command
from alembic.script import ScriptDirectory, Script
def test_args(self):
script = ScriptDirectory(
- staging_directory,
+ _get_staging_directory(),
file_template="%(rev)s_%(slug)s_"
"%(year)s_%(month)s_"
"%(day)s_%(hour)s_"
eq_(
script._rev_path("12345", "this is a message", create_date),
"%s/versions/12345_this_is_a_"
- "message_2012_7_25_15_8_5.py" % staging_directory
+ "message_2012_7_25_15_8_5.py" % _get_staging_directory()
)
[tox]
minversion=1.8.dev1
-envlist = py{27,33}-sqla{09,10}, coverage
+envlist = py{27,33}-sqla{079,084,09,10}, coverage
[testenv]
deps=pytest
mock
+ sqla079: git+http://git.sqlalchemy.org/sqlalchemy.git@rel_0_7_9
+ sqla084: git+http://git.sqlalchemy.org/sqlalchemy.git@rel_0_8_4
sqla09: git+http://git.sqlalchemy.org/sqlalchemy.git@rel_0_9
sqla10: git+http://git.sqlalchemy.org/sqlalchemy.git@master
-recreate=True
+
sitepackages=True
usedevelop=True
commands=
- python -m pytest {posargs}
+ py{27,33}-sqla{084,09,10}: python -m pytest -n 4 {posargs}
+ py{27,33}-sqla{079}: python -m pytest {posargs}
+
+[testenv:py27-sqla10]
+recreate=True
+
+[testenv:py27-sqla09]
+recreate=True
+
+[testenv:py33-sqla10]
+recreate=True
+
+[testenv:py33-sqla09]
+recreate=True
[testenv:coverage]
deps=coverage
# F841,F811,F401
exclude=.venv,.git,.tox,dist,doc,*egg,build
+
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
