from .fixtures import TestBase
from .assertions import eq_, ne_, is_, assert_raises_message, \
- eq_ignore_whitespace
+ 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
import re
from sqlalchemy.engine import default
-from sqlalchemy.testing.assertions import eq_, ne_, is_, assert_raises_message
+from sqlalchemy.testing.assertions import eq_, ne_, is_, \
+ assert_raises_message, assert_raises
from alembic.compat import text_type
f.write(txt)
+def _sqlite_file_db():
+ from sqlalchemy.testing import engines
+ dir_ = os.path.join(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')
+ url = "sqlite:///%s/foo.db" % dir_
+
return _write_config_file("""
[alembic]
script_location = %s
-sqlalchemy.url = sqlite:///%s/foo.db
+sqlalchemy.url = %s
sourceless = %s
[loggers]
[formatter_generic]
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
- """ % (dir_, dir_, "true" if sourceless else "false"))
+ """ % (dir_, url, "true" if sourceless else "false"))
def _no_sql_testing_config(dialect="postgresql", directives=""):
# coding: utf-8
import io
-import os
import re
-import shutil
-import textwrap
-from sqlalchemy.engine import default
from sqlalchemy import create_engine, text, MetaData
-from sqlalchemy.exc import SQLAlchemyError
-from sqlalchemy.util import decorator
import alembic
from alembic.compat import configparser
from alembic import util
-from alembic.compat import string_types, text_type, u, py33
+from alembic.compat import string_types, text_type
from alembic.migration import MigrationContext
from alembic.environment import EnvironmentContext
from alembic.operations import Operations
-from alembic.script import ScriptDirectory, Script
from alembic.ddl.impl import _impls
from contextlib import contextmanager
_engs = {}
-
-
@contextmanager
def capture_context_buffer(**kw):
if kw.pop('bytes_io', False):
_vers = tuple(
[_safe_int(x) for x in re.findall(r'(\d+|[abc]\d)', __version__)])
sqla_07 = _vers > (0, 7, 2)
-sqla_08 = _vers >= (0, 8, 0, 'b2')
+sqla_08 = sqla_084 = _vers >= (0, 8, 4)
sqla_09 = _vers >= (0, 9, 0)
sqla_092 = _vers >= (0, 9, 2)
sqla_094 = _vers >= (0, 9, 4)
-if not sqla_07:
+sqla_094 = _vers >= (0, 9, 4)
+sqla_100 = _vers >= (1, 0, 0)
+if not sqla_084:
raise CommandError(
- "SQLAlchemy 0.7.3 or greater is required. ")
+ "SQLAlchemy 0.8.4 or greater is required. ")
from sqlalchemy.util import format_argspec_plus, update_wrapper
from sqlalchemy.util.compat import inspect_getfullargspec
-from alembic.testing.fixtures import *
+
from alembic.testing import TestBase
from alembic.testing import config
-from sqlalchemy import MetaData, Column, Table, Integer, String, Text, \
- Numeric, DATETIME, INTEGER, \
- TypeDecorator, Unicode, Enum,\
- UniqueConstraint, Boolean, \
- PrimaryKeyConstraint, Index, func, ForeignKeyConstraint,\
+from sqlalchemy import MetaData, Column, Table, Integer, String, \
+ Numeric, UniqueConstraint, Index, ForeignKeyConstraint,\
ForeignKey
-from sqlalchemy.schema import AddConstraint
from sqlalchemy.testing import engines
from alembic.testing import eq_
from alembic.testing.env import staging_env
class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
reports_unique_constraints = True
+ __only_on__ = 'sqlite'
def test_index_flag_becomes_named_unique_constraint(self):
m1 = MetaData()
-"""Test op functions against ORACLE."""
-
-from unittest import TestCase
from sqlalchemy import Integer, Column
from alembic import op, command
-from . import op_fixture, capture_context_buffer, \
- _no_sql_testing_config, staging_env, \
+from alembic.testing.fixtures import TestBase
+
+from alembic.testing.fixtures import op_fixture, capture_context_buffer
+from alembic.testing.env import _no_sql_testing_config, staging_env, \
three_rev_fixture, clear_staging_env
-class FullEnvironmentTests(TestCase):
+class FullEnvironmentTests(TestBase):
@classmethod
def setup_class(cls):
- env = staging_env()
+ staging_env()
cls.cfg = cfg = _no_sql_testing_config("oracle")
cls.a, cls.b, cls.c = \
assert "BYE" in buf.getvalue()
-class OpTest(TestCase):
+class OpTest(TestBase):
def test_add_column(self):
context = op_fixture('oracle')
-from tests import clear_staging_env, staging_env, eq_, ne_, is_, \
- staging_directory
-from tests import _no_sql_testing_config, env_file_fixture, \
+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, \
script_file_fixture, _testing_config
from alembic import command
from alembic.script import ScriptDirectory, Script
from alembic.environment import EnvironmentContext
from alembic import util
import os
-import unittest
import datetime
env, abc, def_ = None, None, None
-class GeneralOrderedTests(unittest.TestCase):
+class GeneralOrderedTests(TestBase):
- def test_001_environment(self):
+ def setUp(self):
+ global env
+ env = staging_env()
+
+ def tearDown(self):
+ clear_staging_env()
+
+ def test_steps(self):
+ self._test_001_environment()
+ self._test_002_rev_ids()
+ self._test_003_api_methods_clean()
+ self._test_004_rev()
+ self._test_005_nextrev()
+ self._test_006_from_clean_env()
+ self._test_007_no_refresh()
+ self._test_008_long_name()
+ self._test_009_long_name_configurable()
+
+ def _test_001_environment(self):
assert_set = set(['env.py', 'script.py.mako', 'README'])
eq_(
assert_set.intersection(os.listdir(env.dir)),
assert_set
)
- def test_002_rev_ids(self):
+ def _test_002_rev_ids(self):
global abc, def_
abc = util.rev_id()
def_ = util.rev_id()
ne_(abc, def_)
- def test_003_api_methods_clean(self):
+ def _test_003_api_methods_clean(self):
eq_(env.get_heads(), [])
eq_(env.get_base(), None)
- def test_004_rev(self):
+ def _test_004_rev(self):
script = env.generate_revision(abc, "this is a message", refresh=True)
eq_(script.doc, "this is a message")
eq_(script.revision, abc)
eq_(env.get_heads(), [abc])
eq_(env.get_base(), abc)
- def test_005_nextrev(self):
+ def _test_005_nextrev(self):
script = env.generate_revision(
def_, "this is the next rev", refresh=True)
assert os.access(
eq_(env.get_heads(), [def_])
eq_(env.get_base(), abc)
- def test_006_from_clean_env(self):
+ def _test_006_from_clean_env(self):
# test the environment so far with a
# new ScriptDirectory instance.
eq_(env.get_heads(), [def_])
eq_(env.get_base(), abc)
- def test_007_no_refresh(self):
+ def _test_007_no_refresh(self):
rid = util.rev_id()
script = env.generate_revision(rid, "dont' refresh")
is_(script, None)
env2 = staging_env(create=False)
eq_(env2._as_rev_number("head"), rid)
- def test_008_long_name(self):
+ def _test_008_long_name(self):
rid = util.rev_id()
env.generate_revision(rid,
"this is a really long name with "
'%s_this_is_a_really_long_name_with_lots_of_.py' % rid),
os.F_OK)
- def test_009_long_name_configurable(self):
+ def _test_009_long_name_configurable(self):
env.truncate_slug_length = 60
rid = util.rev_id()
env.generate_revision(rid,
'of_characters_and_also_.py' % rid),
os.F_OK)
- @classmethod
- def setup_class(cls):
- global env
- env = staging_env()
-
- @classmethod
- def teardown_class(cls):
- clear_staging_env()
-
-class ScriptNamingTest(unittest.TestCase):
+class ScriptNamingTest(TestBase):
@classmethod
def setup_class(cls):
)
-class TemplateArgsTest(unittest.TestCase):
+class TemplateArgsTest(TestBase):
def setUp(self):
staging_env()
-from tests import clear_staging_env, staging_env, eq_, \
- assert_raises_message
+from alembic.testing.env import clear_staging_env, staging_env
+from alembic.testing import assert_raises_message, eq_
from alembic import util
+from alembic.testing.fixtures import TestBase
+
env = None
a, b, c, d, e = None, None, None, None, None
cfg = None
-def setup():
- global env
- env = staging_env()
- global a, b, c, d, e
- a = env.generate_revision(util.rev_id(), '->a', refresh=True)
- b = env.generate_revision(util.rev_id(), 'a->b', refresh=True)
- c = env.generate_revision(util.rev_id(), 'b->c', refresh=True)
- d = env.generate_revision(util.rev_id(), 'c->d', refresh=True)
- e = env.generate_revision(util.rev_id(), 'd->e', refresh=True)
-
-
-def teardown():
- clear_staging_env()
-
-
-def test_upgrade_path():
-
- eq_(
- env._upgrade_revs(e.revision, c.revision),
- [
- (d.module.upgrade, c.revision, d.revision, d.doc),
- (e.module.upgrade, d.revision, e.revision, e.doc),
- ]
- )
-
- eq_(
- env._upgrade_revs(c.revision, None),
- [
- (a.module.upgrade, None, a.revision, a.doc),
- (b.module.upgrade, a.revision, b.revision, b.doc),
- (c.module.upgrade, b.revision, c.revision, c.doc),
- ]
- )
-
-
-def test_relative_upgrade_path():
- eq_(
- env._upgrade_revs("+2", a.revision),
- [
- (b.module.upgrade, a.revision, b.revision, b.doc),
- (c.module.upgrade, b.revision, c.revision, c.doc),
- ]
- )
-
- eq_(
- env._upgrade_revs("+1", a.revision),
- [
- (b.module.upgrade, a.revision, b.revision, b.doc),
- ]
- )
-
- eq_(
- env._upgrade_revs("+3", b.revision),
- [
- (c.module.upgrade, b.revision, c.revision, c.doc),
- (d.module.upgrade, c.revision, d.revision, d.doc),
- (e.module.upgrade, d.revision, e.revision, e.doc),
- ]
- )
-
-
-def test_invalid_relative_upgrade_path():
- assert_raises_message(
- util.CommandError,
- "Relative revision -2 didn't produce 2 migrations",
- env._upgrade_revs, "-2", b.revision
- )
-
- assert_raises_message(
- util.CommandError,
- r"Relative revision \+5 didn't produce 5 migrations",
- env._upgrade_revs, "+5", b.revision
- )
-
-
-def test_downgrade_path():
-
- eq_(
- env._downgrade_revs(c.revision, e.revision),
- [
- (e.module.downgrade, e.revision, e.down_revision, e.doc),
- (d.module.downgrade, d.revision, d.down_revision, d.doc),
- ]
- )
-
- eq_(
- env._downgrade_revs(None, c.revision),
- [
- (c.module.downgrade, c.revision, c.down_revision, c.doc),
- (b.module.downgrade, b.revision, b.down_revision, b.doc),
- (a.module.downgrade, a.revision, a.down_revision, a.doc),
- ]
- )
-
-
-def test_relative_downgrade_path():
- eq_(
- env._downgrade_revs("-1", c.revision),
- [
- (c.module.downgrade, c.revision, c.down_revision, c.doc),
- ]
- )
-
- eq_(
- env._downgrade_revs("-3", e.revision),
- [
- (e.module.downgrade, e.revision, e.down_revision, e.doc),
- (d.module.downgrade, d.revision, d.down_revision, d.doc),
- (c.module.downgrade, c.revision, c.down_revision, c.doc),
- ]
- )
-
-
-def test_invalid_relative_downgrade_path():
- assert_raises_message(
- util.CommandError,
- "Relative revision -5 didn't produce 5 migrations",
- env._downgrade_revs, "-5", b.revision
- )
-
- assert_raises_message(
- util.CommandError,
- r"Relative revision \+2 didn't produce 2 migrations",
- env._downgrade_revs, "+2", b.revision
- )
-
-
-def test_invalid_move_rev_to_none():
- assert_raises_message(
- util.CommandError,
- "Revision %s is not an ancestor of base" % b.revision,
- env._downgrade_revs, b.revision[0:3], None
- )
-
-
-def test_invalid_move_higher_to_lower():
- assert_raises_message(
- util.CommandError,
- "Revision %s is not an ancestor of %s" % (c.revision, b.revision),
- env._downgrade_revs, c.revision[0:4], b.revision
- )
+class RevisionPathTest(TestBase):
+
+ @classmethod
+ def setup_class(cls):
+ global env
+ env = staging_env()
+ global a, b, c, d, e
+ a = env.generate_revision(util.rev_id(), '->a', refresh=True)
+ b = env.generate_revision(util.rev_id(), 'a->b', refresh=True)
+ c = env.generate_revision(util.rev_id(), 'b->c', refresh=True)
+ d = env.generate_revision(util.rev_id(), 'c->d', refresh=True)
+ e = env.generate_revision(util.rev_id(), 'd->e', refresh=True)
+
+ @classmethod
+ def teardown_class(cls):
+ clear_staging_env()
+
+ def test_upgrade_path(self):
+
+ eq_(
+ env._upgrade_revs(e.revision, c.revision),
+ [
+ (d.module.upgrade, c.revision, d.revision, d.doc),
+ (e.module.upgrade, d.revision, e.revision, e.doc),
+ ]
+ )
+
+ eq_(
+ env._upgrade_revs(c.revision, None),
+ [
+ (a.module.upgrade, None, a.revision, a.doc),
+ (b.module.upgrade, a.revision, b.revision, b.doc),
+ (c.module.upgrade, b.revision, c.revision, c.doc),
+ ]
+ )
+
+ def test_relative_upgrade_path(self):
+ eq_(
+ env._upgrade_revs("+2", a.revision),
+ [
+ (b.module.upgrade, a.revision, b.revision, b.doc),
+ (c.module.upgrade, b.revision, c.revision, c.doc),
+ ]
+ )
+
+ eq_(
+ env._upgrade_revs("+1", a.revision),
+ [
+ (b.module.upgrade, a.revision, b.revision, b.doc),
+ ]
+ )
+
+ eq_(
+ env._upgrade_revs("+3", b.revision),
+ [
+ (c.module.upgrade, b.revision, c.revision, c.doc),
+ (d.module.upgrade, c.revision, d.revision, d.doc),
+ (e.module.upgrade, d.revision, e.revision, e.doc),
+ ]
+ )
+
+ def test_invalid_relative_upgrade_path(self):
+ assert_raises_message(
+ util.CommandError,
+ "Relative revision -2 didn't produce 2 migrations",
+ env._upgrade_revs, "-2", b.revision
+ )
+
+ assert_raises_message(
+ util.CommandError,
+ r"Relative revision \+5 didn't produce 5 migrations",
+ env._upgrade_revs, "+5", b.revision
+ )
+
+ def test_downgrade_path(self):
+
+ eq_(
+ env._downgrade_revs(c.revision, e.revision),
+ [
+ (e.module.downgrade, e.revision, e.down_revision, e.doc),
+ (d.module.downgrade, d.revision, d.down_revision, d.doc),
+ ]
+ )
+
+ eq_(
+ env._downgrade_revs(None, c.revision),
+ [
+ (c.module.downgrade, c.revision, c.down_revision, c.doc),
+ (b.module.downgrade, b.revision, b.down_revision, b.doc),
+ (a.module.downgrade, a.revision, a.down_revision, a.doc),
+ ]
+ )
+
+ def test_relative_downgrade_path(self):
+ eq_(
+ env._downgrade_revs("-1", c.revision),
+ [
+ (c.module.downgrade, c.revision, c.down_revision, c.doc),
+ ]
+ )
+
+ eq_(
+ env._downgrade_revs("-3", e.revision),
+ [
+ (e.module.downgrade, e.revision, e.down_revision, e.doc),
+ (d.module.downgrade, d.revision, d.down_revision, d.doc),
+ (c.module.downgrade, c.revision, c.down_revision, c.doc),
+ ]
+ )
+
+ def test_invalid_relative_downgrade_path(self):
+ assert_raises_message(
+ util.CommandError,
+ "Relative revision -5 didn't produce 5 migrations",
+ env._downgrade_revs, "-5", b.revision
+ )
+
+ assert_raises_message(
+ util.CommandError,
+ r"Relative revision \+2 didn't produce 2 migrations",
+ env._downgrade_revs, "+2", b.revision
+ )
+
+ def test_invalid_move_rev_to_none(self):
+ assert_raises_message(
+ util.CommandError,
+ "Revision %s is not an ancestor of base" % b.revision,
+ env._downgrade_revs, b.revision[0:3], None
+ )
+
+ def test_invalid_move_higher_to_lower(self):
+ assert_raises_message(
+ util.CommandError,
+ "Revision %s is not an ancestor of %s" % (c.revision, b.revision),
+ env._downgrade_revs, c.revision[0:4], b.revision
+ )
from __future__ import unicode_literals
-import unittest
+from alembic.testing.fixtures import TestBase
+from alembic import util
-from . import clear_staging_env, staging_env, \
- _no_sql_testing_config, capture_context_buffer, \
- three_rev_fixture, write_script
-from alembic import command, util
+from alembic.testing.env import staging_env, _no_sql_testing_config, \
+ three_rev_fixture, clear_staging_env, write_script
+from alembic.testing.fixtures import capture_context_buffer
+from alembic import command
from alembic.script import ScriptDirectory
import re
a, b, c = None, None, None
-class ThreeRevTest(unittest.TestCase):
+class ThreeRevTest(TestBase):
def setUp(self):
global cfg, env
"SET version_num='%s';" % c in buf.getvalue()
-class EncodingTest(unittest.TestCase):
+class EncodingTest(TestBase):
def setUp(self):
global cfg, env, a
-from tests import op_fixture, assert_raises_message
+from alembic.testing.fixtures import op_fixture
+from alembic.testing import assert_raises_message
from alembic import op
from sqlalchemy import Integer, Column, Boolean
from sqlalchemy.sql import column
-
-
-def test_add_column():
- context = op_fixture('sqlite')
- op.add_column('t1', Column('c1', Integer))
- context.assert_(
- 'ALTER TABLE t1 ADD COLUMN c1 INTEGER'
- )
-
-
-def test_add_column_implicit_constraint():
- context = op_fixture('sqlite')
- op.add_column('t1', Column('c1', Boolean))
- context.assert_(
- 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN'
- )
-
-
-def test_add_explicit_constraint():
- context = op_fixture('sqlite')
- assert_raises_message(
- NotImplementedError,
- "No support for ALTER of constraints in SQLite dialect",
- op.create_check_constraint,
- "foo",
- "sometable",
- column('name') > 5
- )
-
-
-def test_drop_explicit_constraint():
- context = op_fixture('sqlite')
- assert_raises_message(
- NotImplementedError,
- "No support for ALTER of constraints in SQLite dialect",
- op.drop_constraint,
- "foo",
- "sometable",
- )
+from alembic.testing.fixtures import TestBase
+
+
+class SQLiteTest(TestBase):
+
+ def test_add_column(self):
+ context = op_fixture('sqlite')
+ op.add_column('t1', Column('c1', Integer))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 INTEGER'
+ )
+
+ def test_add_column_implicit_constraint(self):
+ context = op_fixture('sqlite')
+ op.add_column('t1', Column('c1', Boolean))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN'
+ )
+
+ def test_add_explicit_constraint(self):
+ op_fixture('sqlite')
+ assert_raises_message(
+ NotImplementedError,
+ "No support for ALTER of constraints in SQLite dialect",
+ op.create_check_constraint,
+ "foo",
+ "sometable",
+ column('name') > 5
+ )
+
+ def test_drop_explicit_constraint(self):
+ op_fixture('sqlite')
+ assert_raises_message(
+ NotImplementedError,
+ "No support for ALTER of constraints in SQLite dialect",
+ op.drop_constraint,
+ "foo",
+ "sometable",
+ )
-import unittest
+from alembic.testing.fixtures import TestBase
-from sqlalchemy import Table, MetaData, Column, String, create_engine
+from alembic.testing import config, eq_, assert_raises
+
+from sqlalchemy import Table, MetaData, Column, String
from sqlalchemy.engine.reflection import Inspector
from alembic.util import CommandError
Column('version_num', String(32), nullable=False))
-class TestMigrationContext(unittest.TestCase):
- _bind = []
+class TestMigrationContext(TestBase):
- @property
- def bind(self):
- if not self._bind:
- engine = create_engine('sqlite:///', echo=True)
- self._bind.append(engine)
- return self._bind[0]
+ @classmethod
+ def setup_class(cls):
+ cls.bind = config.db
def setUp(self):
self.connection = self.bind.connect()
def tearDown(self):
version_table.drop(self.connection, checkfirst=True)
self.transaction.rollback()
+ self.connection.close()
def make_one(self, **kwargs):
from alembic.migration import MigrationContext
rows = result.fetchall()
if len(rows) == 0:
return None
- self.assertEqual(len(rows), 1)
+ eq_(len(rows), 1)
return rows[0]['version_num']
def test_config_default_version_table_name(self):
context = self.make_one(dialect_name='sqlite')
- self.assertEqual(context._version.name, 'alembic_version')
+ eq_(context._version.name, 'alembic_version')
def test_config_explicit_version_table_name(self):
context = self.make_one(dialect_name='sqlite',
opts={'version_table': 'explicit'})
- self.assertEqual(context._version.name, 'explicit')
+ eq_(context._version.name, 'explicit')
def test_config_explicit_version_table_schema(self):
context = self.make_one(dialect_name='sqlite',
opts={'version_table_schema': 'explicit'})
- self.assertEqual(context._version.schema, 'explicit')
+ eq_(context._version.schema, 'explicit')
def test_get_current_revision_creates_version_table(self):
context = self.make_one(connection=self.connection,
opts={'version_table': 'version_table'})
- self.assertEqual(context.get_current_revision(), None)
+ eq_(context.get_current_revision(), None)
insp = Inspector(self.connection)
- self.assertTrue('version_table' in insp.get_table_names())
+ assert ('version_table' in insp.get_table_names())
def test_get_current_revision(self):
context = self.make_one(connection=self.connection,
opts={'version_table': 'version_table'})
version_table.create(self.connection)
- self.assertEqual(context.get_current_revision(), None)
+ eq_(context.get_current_revision(), None)
self.connection.execute(
version_table.insert().values(version_num='revid'))
- self.assertEqual(context.get_current_revision(), 'revid')
+ eq_(context.get_current_revision(), 'revid')
def test_get_current_revision_error_if_starting_rev_given_online(self):
context = self.make_one(connection=self.connection,
opts={'starting_rev': 'boo'})
- self.assertRaises(CommandError, context.get_current_revision)
+ assert_raises(
+ CommandError,
+ context.get_current_revision
+ )
def test_get_current_revision_offline(self):
context = self.make_one(dialect_name='sqlite',
opts={'starting_rev': 'startrev',
'as_sql': True})
- self.assertEqual(context.get_current_revision(), 'startrev')
+ eq_(context.get_current_revision(), 'startrev')
def test__update_current_rev(self):
version_table.create(self.connection)
opts={'version_table': 'version_table'})
context._update_current_rev(None, 'a')
- self.assertEqual(self.get_revision(), 'a')
+ eq_(self.get_revision(), 'a')
context._update_current_rev('a', 'b')
- self.assertEqual(self.get_revision(), 'b')
+ eq_(self.get_revision(), 'b')
context._update_current_rev('b', None)
- self.assertEqual(self.get_revision(), None)
+ eq_(self.get_revision(), None)
import os
-import unittest
from alembic import command, util
from alembic.script import ScriptDirectory
-from . import clear_staging_env, staging_env, \
- _sqlite_testing_config, sqlite_db, eq_, write_script, \
- assert_raises_message
+from alembic.testing.env import clear_staging_env, staging_env, \
+ _sqlite_testing_config, write_script, _sqlite_file_db
+from alembic.testing import eq_, assert_raises_message
+from alembic.testing.fixtures import TestBase
a = b = c = None
-class VersioningTest(unittest.TestCase):
+class VersioningTest(TestBase):
+ __only_on__ = 'sqlite'
+
sourceless = False
- def test_001_revisions(self):
+ def setUp(self):
+ self.bind = _sqlite_file_db()
+ self.env = staging_env(sourceless=self.sourceless)
+ self.cfg = _sqlite_testing_config(sourceless=self.sourceless)
+
+ def tearDown(self):
+ clear_staging_env()
+
+ def test_steps(self):
+ self._test_001_revisions()
+ self._test_002_upgrade()
+ self._test_003_downgrade()
+ self._test_004_downgrade()
+ self._test_005_upgrade()
+ self._test_006_upgrade_again()
+
+ def _test_001_revisions(self):
global a, b, c
a = util.rev_id()
b = util.rev_id()
""" % (c, b), sourceless=self.sourceless)
- def test_002_upgrade(self):
+ def _test_002_upgrade(self):
command.upgrade(self.cfg, c)
- db = sqlite_db()
+ db = self.bind
assert db.dialect.has_table(db.connect(), 'foo')
assert db.dialect.has_table(db.connect(), 'bar')
assert db.dialect.has_table(db.connect(), 'bat')
- def test_003_downgrade(self):
+ def _test_003_downgrade(self):
command.downgrade(self.cfg, a)
- db = sqlite_db()
+ db = self.bind
assert db.dialect.has_table(db.connect(), 'foo')
assert not db.dialect.has_table(db.connect(), 'bar')
assert not db.dialect.has_table(db.connect(), 'bat')
- def test_004_downgrade(self):
+ def _test_004_downgrade(self):
command.downgrade(self.cfg, 'base')
- db = sqlite_db()
+ db = self.bind
assert not db.dialect.has_table(db.connect(), 'foo')
assert not db.dialect.has_table(db.connect(), 'bar')
assert not db.dialect.has_table(db.connect(), 'bat')
- def test_005_upgrade(self):
+ def _test_005_upgrade(self):
command.upgrade(self.cfg, b)
- db = sqlite_db()
+ db = self.bind
assert db.dialect.has_table(db.connect(), 'foo')
assert db.dialect.has_table(db.connect(), 'bar')
assert not db.dialect.has_table(db.connect(), 'bat')
- def test_006_upgrade_again(self):
+ def _test_006_upgrade_again(self):
command.upgrade(self.cfg, b)
- # TODO: test some invalid movements
-
- @classmethod
- def setup_class(cls):
- cls.env = staging_env(sourceless=cls.sourceless)
- cls.cfg = _sqlite_testing_config(sourceless=cls.sourceless)
-
- @classmethod
- def teardown_class(cls):
- clear_staging_env()
-
-class VersionNameTemplateTest(unittest.TestCase):
+class VersionNameTemplateTest(TestBase):
def setUp(self):
self.env = staging_env()
sourceless = True
-class SourcelessNeedsFlagTest(unittest.TestCase):
+class SourcelessNeedsFlagTest(TestBase):
def setUp(self):
self.env = staging_env(sourceless=False)
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
