diff options
Diffstat (limited to 'lib/python2.7/site-packages/sqlalchemy_migrate-0.7.2-py2.7.egg/migrate/tests/changeset/test_changeset.py')
| -rwxr-xr-x | lib/python2.7/site-packages/sqlalchemy_migrate-0.7.2-py2.7.egg/migrate/tests/changeset/test_changeset.py | 910 |
1 files changed, 0 insertions, 910 deletions
diff --git a/lib/python2.7/site-packages/sqlalchemy_migrate-0.7.2-py2.7.egg/migrate/tests/changeset/test_changeset.py b/lib/python2.7/site-packages/sqlalchemy_migrate-0.7.2-py2.7.egg/migrate/tests/changeset/test_changeset.py deleted file mode 100755 index e1166f43..00000000 --- a/lib/python2.7/site-packages/sqlalchemy_migrate-0.7.2-py2.7.egg/migrate/tests/changeset/test_changeset.py +++ /dev/null @@ -1,910 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -import sqlalchemy -import warnings - -from sqlalchemy import * - -from migrate import changeset, exceptions -from migrate.changeset import * -from migrate.changeset.schema import ColumnDelta -from migrate.tests import fixture -from migrate.tests.fixture.warnings import catch_warnings - -class TestAddDropColumn(fixture.DB): - """Test add/drop column through all possible interfaces - also test for constraints - """ - level = fixture.DB.CONNECT - table_name = 'tmp_adddropcol' - table_int = 0 - - def _setup(self, url): - super(TestAddDropColumn, self)._setup(url) - self.meta = MetaData() - self.table = Table(self.table_name, self.meta, - Column('id', Integer, unique=True), - ) - self.meta.bind = self.engine - if self.engine.has_table(self.table.name): - self.table.drop() - self.table.create() - - def _teardown(self): - if self.engine.has_table(self.table.name): - self.table.drop() - self.meta.clear() - super(TestAddDropColumn,self)._teardown() - - def run_(self, create_column_func, drop_column_func, *col_p, **col_k): - col_name = 'data' - - def assert_numcols(num_of_expected_cols): - # number of cols should be correct in table object and in database - self.refresh_table(self.table_name) - result = len(self.table.c) - - self.assertEquals(result, num_of_expected_cols), - if col_k.get('primary_key', None): - # new primary key: check its length too - result = len(self.table.primary_key) - self.assertEquals(result, num_of_expected_cols) - - # we have 1 columns and there is no data column - assert_numcols(1) - self.assertTrue(getattr(self.table.c, 'data', None) is None) - if len(col_p) == 0: - col_p = [String(40)] - col = Column(col_name, *col_p, **col_k) - create_column_func(col) - assert_numcols(2) - # data column exists - self.assert_(self.table.c.data.type.length, 40) - - col2 = self.table.c.data - drop_column_func(col2) - assert_numcols(1) - - @fixture.usedb() - def test_undefined(self): - """Add/drop columns not yet defined in the table""" - def add_func(col): - return create_column(col, self.table) - def drop_func(col): - return drop_column(col, self.table) - return self.run_(add_func, drop_func) - - @fixture.usedb() - def test_defined(self): - """Add/drop columns already defined in the table""" - def add_func(col): - self.meta.clear() - self.table = Table(self.table_name, self.meta, - Column('id', Integer, primary_key=True), - col, - ) - return create_column(col) - def drop_func(col): - return drop_column(col) - return self.run_(add_func, drop_func) - - @fixture.usedb() - def test_method_bound(self): - """Add/drop columns via column methods; columns bound to a table - ie. no table parameter passed to function - """ - def add_func(col): - self.assert_(col.table is None, col.table) - self.table.append_column(col) - return col.create() - def drop_func(col): - #self.assert_(col.table is None,col.table) - #self.table.append_column(col) - return col.drop() - return self.run_(add_func, drop_func) - - @fixture.usedb() - def test_method_notbound(self): - """Add/drop columns via column methods; columns not bound to a table""" - def add_func(col): - return col.create(self.table) - def drop_func(col): - return col.drop(self.table) - return self.run_(add_func, drop_func) - - @fixture.usedb() - def test_tablemethod_obj(self): - """Add/drop columns via table methods; by column object""" - def add_func(col): - return self.table.create_column(col) - def drop_func(col): - return self.table.drop_column(col) - return self.run_(add_func, drop_func) - - @fixture.usedb() - def test_tablemethod_name(self): - """Add/drop columns via table methods; by column name""" - def add_func(col): - # must be bound to table - self.table.append_column(col) - return self.table.create_column(col.name) - def drop_func(col): - # Not necessarily bound to table - return self.table.drop_column(col.name) - return self.run_(add_func, drop_func) - - @fixture.usedb() - def test_byname(self): - """Add/drop columns via functions; by table object and column name""" - def add_func(col): - self.table.append_column(col) - return create_column(col.name, self.table) - def drop_func(col): - return drop_column(col.name, self.table) - return self.run_(add_func, drop_func) - - @fixture.usedb() - def test_drop_column_not_in_table(self): - """Drop column by name""" - def add_func(col): - return self.table.create_column(col) - def drop_func(col): - if SQLA_07: - self.table._columns.remove(col) - else: - self.table.c.remove(col) - return self.table.drop_column(col.name) - self.run_(add_func, drop_func) - - @fixture.usedb() - def test_fk(self): - """Can create columns with foreign keys""" - # create FK's target - reftable = Table('tmp_ref', self.meta, - Column('id', Integer, primary_key=True), - ) - if self.engine.has_table(reftable.name): - reftable.drop() - reftable.create() - - # create column with fk - col = Column('data', Integer, ForeignKey(reftable.c.id)) - col.create(self.table) - - # check if constraint is added - for cons in self.table.constraints: - if isinstance(cons, sqlalchemy.schema.ForeignKeyConstraint): - break - else: - self.fail('No constraint found') - - # TODO: test on db level if constraints work - - if SQLA_07: - self.assertEqual(reftable.c.id.name, - list(col.foreign_keys)[0].column.name) - else: - self.assertEqual(reftable.c.id.name, - col.foreign_keys[0].column.name) - col.drop(self.table) - - if self.engine.has_table(reftable.name): - reftable.drop() - - @fixture.usedb(not_supported='sqlite') - def test_pk(self): - """Can create columns with primary key""" - col = Column('data', Integer, nullable=False) - self.assertRaises(exceptions.InvalidConstraintError, - col.create, self.table, primary_key_name=True) - col.create(self.table, primary_key_name='data_pkey') - - # check if constraint was added (cannot test on objects) - self.table.insert(values={'data': 4}).execute() - try: - self.table.insert(values={'data': 4}).execute() - except (sqlalchemy.exc.IntegrityError, - sqlalchemy.exc.ProgrammingError): - pass - else: - self.fail() - - col.drop() - - @fixture.usedb(not_supported='mysql') - def test_check(self): - """Can create columns with check constraint""" - col = Column('data', - Integer, - sqlalchemy.schema.CheckConstraint('data > 4')) - col.create(self.table) - - # check if constraint was added (cannot test on objects) - self.table.insert(values={'data': 5}).execute() - try: - self.table.insert(values={'data': 3}).execute() - except (sqlalchemy.exc.IntegrityError, - sqlalchemy.exc.ProgrammingError): - pass - else: - self.fail() - - col.drop() - - @fixture.usedb() - def test_unique_constraint(self): - self.assertRaises(exceptions.InvalidConstraintError, - Column('data', Integer, unique=True).create, self.table) - - col = Column('data', Integer) - col.create(self.table, unique_name='data_unique') - - # check if constraint was added (cannot test on objects) - self.table.insert(values={'data': 5}).execute() - try: - self.table.insert(values={'data': 5}).execute() - except (sqlalchemy.exc.IntegrityError, - sqlalchemy.exc.ProgrammingError): - pass - else: - self.fail() - - col.drop(self.table) - -# TODO: remove already attached columns with indexes, uniques, pks, fks .. - - def _check_index(self,expected): - if 'mysql' in self.engine.name or 'postgres' in self.engine.name: - for index in tuple( - Table(self.table.name, MetaData(), - autoload=True, autoload_with=self.engine).indexes - ): - if index.name=='ix_data': - break - self.assertEqual(expected,index.unique) - - @fixture.usedb() - def test_index(self): - col = Column('data', Integer) - col.create(self.table, index_name='ix_data') - - self._check_index(False) - - col.drop() - - @fixture.usedb() - def test_index_unique(self): - # shows how to create a unique index - col = Column('data', Integer) - col.create(self.table) - Index('ix_data', col, unique=True).create(bind=self.engine) - - # check if index was added - self.table.insert(values={'data': 5}).execute() - try: - self.table.insert(values={'data': 5}).execute() - except (sqlalchemy.exc.IntegrityError, - sqlalchemy.exc.ProgrammingError): - pass - else: - self.fail() - - self._check_index(True) - - col.drop() - - @fixture.usedb() - def test_server_defaults(self): - """Can create columns with server_default values""" - col = Column('data', String(244), server_default='foobar') - col.create(self.table) - - self.table.insert(values={'id': 10}).execute() - row = self._select_row() - self.assertEqual(u'foobar', row['data']) - - col.drop() - - @fixture.usedb() - def test_populate_default(self): - """Test populate_default=True""" - def default(): - return 'foobar' - col = Column('data', String(244), default=default) - col.create(self.table, populate_default=True) - - self.table.insert(values={'id': 10}).execute() - row = self._select_row() - self.assertEqual(u'foobar', row['data']) - - col.drop() - - # TODO: test sequence - # TODO: test quoting - # TODO: test non-autoname constraints - - @fixture.usedb() - def test_drop_doesnt_delete_other_indexes(self): - # add two indexed columns - self.table.drop() - self.meta.clear() - self.table = Table( - self.table_name, self.meta, - Column('id', Integer, primary_key=True), - Column('d1', String(10), index=True), - Column('d2', String(10), index=True), - ) - self.table.create() - - # paranoid check - self.refresh_table() - self.assertEqual( - sorted([i.name for i in self.table.indexes]), - [u'ix_tmp_adddropcol_d1', u'ix_tmp_adddropcol_d2'] - ) - - # delete one - self.table.c.d2.drop() - - # ensure the other index is still there - self.refresh_table() - self.assertEqual( - sorted([i.name for i in self.table.indexes]), - [u'ix_tmp_adddropcol_d1'] - ) - - def _actual_foreign_keys(self): - from sqlalchemy.schema import ForeignKeyConstraint - result = [] - for cons in self.table.constraints: - if isinstance(cons,ForeignKeyConstraint): - col_names = [] - for col_name in cons.columns: - if not isinstance(col_name,basestring): - col_name = col_name.name - col_names.append(col_name) - result.append(col_names) - result.sort() - return result - - @fixture.usedb() - def test_drop_with_foreign_keys(self): - self.table.drop() - self.meta.clear() - - # create FK's target - reftable = Table('tmp_ref', self.meta, - Column('id', Integer, primary_key=True), - ) - if self.engine.has_table(reftable.name): - reftable.drop() - reftable.create() - - # add a table with two foreign key columns - self.table = Table( - self.table_name, self.meta, - Column('id', Integer, primary_key=True), - Column('r1', Integer, ForeignKey('tmp_ref.id')), - Column('r2', Integer, ForeignKey('tmp_ref.id')), - ) - self.table.create() - - # paranoid check - self.assertEqual([['r1'],['r2']], - self._actual_foreign_keys()) - - # delete one - self.table.c.r2.drop() - - # check remaining foreign key is there - self.assertEqual([['r1']], - self._actual_foreign_keys()) - - @fixture.usedb() - def test_drop_with_complex_foreign_keys(self): - from sqlalchemy.schema import ForeignKeyConstraint - from sqlalchemy.schema import UniqueConstraint - - self.table.drop() - self.meta.clear() - - # create FK's target - reftable = Table('tmp_ref', self.meta, - Column('id', Integer, primary_key=True), - Column('jd', Integer), - UniqueConstraint('id','jd') - ) - if self.engine.has_table(reftable.name): - reftable.drop() - reftable.create() - - # add a table with a complex foreign key constraint - self.table = Table( - self.table_name, self.meta, - Column('id', Integer, primary_key=True), - Column('r1', Integer), - Column('r2', Integer), - ForeignKeyConstraint(['r1','r2'], - [reftable.c.id,reftable.c.jd], - name='test_fk') - ) - self.table.create() - - # paranoid check - self.assertEqual([['r1','r2']], - self._actual_foreign_keys()) - - # delete one - self.table.c.r2.drop() - - # check the constraint is gone, since part of it - # is no longer there - if people hit this, - # they may be confused, maybe we should raise an error - # and insist that the constraint is deleted first, separately? - self.assertEqual([], - self._actual_foreign_keys()) - -class TestRename(fixture.DB): - """Tests for table and index rename methods""" - level = fixture.DB.CONNECT - meta = MetaData() - - def _setup(self, url): - super(TestRename, self)._setup(url) - self.meta.bind = self.engine - - @fixture.usedb(not_supported='firebird') - def test_rename_table(self): - """Tables can be renamed""" - c_name = 'col_1' - table_name1 = 'name_one' - table_name2 = 'name_two' - index_name1 = 'x' + table_name1 - index_name2 = 'x' + table_name2 - - self.meta.clear() - self.column = Column(c_name, Integer) - self.table = Table(table_name1, self.meta, self.column) - self.index = Index(index_name1, self.column, unique=False) - - if self.engine.has_table(self.table.name): - self.table.drop() - if self.engine.has_table(table_name2): - tmp = Table(table_name2, self.meta, autoload=True) - tmp.drop() - tmp.deregister() - del tmp - self.table.create() - - def assert_table_name(expected, skip_object_check=False): - """Refresh a table via autoload - SA has changed some since this test was written; we now need to do - meta.clear() upon reloading a table - clear all rather than a - select few. So, this works only if we're working with one table at - a time (else, others will vanish too). - """ - if not skip_object_check: - # Table object check - self.assertEquals(self.table.name,expected) - newname = self.table.name - else: - # we know the object's name isn't consistent: just assign it - newname = expected - # Table DB check - self.meta.clear() - self.table = Table(newname, self.meta, autoload=True) - self.assertEquals(self.table.name, expected) - - def assert_index_name(expected, skip_object_check=False): - if not skip_object_check: - # Index object check - self.assertEquals(self.index.name, expected) - else: - # object is inconsistent - self.index.name = expected - # TODO: Index DB check - - def add_table_to_meta(name): - # trigger the case where table_name2 needs to be - # removed from the metadata in ChangesetTable.deregister() - tmp = Table(name, self.meta, Column(c_name, Integer)) - tmp.create() - tmp.drop() - - try: - # Table renames - assert_table_name(table_name1) - add_table_to_meta(table_name2) - rename_table(self.table, table_name2) - assert_table_name(table_name2) - self.table.rename(table_name1) - assert_table_name(table_name1) - - # test by just the string - rename_table(table_name1, table_name2, engine=self.engine) - assert_table_name(table_name2, True) # object not updated - - # Index renames - if self.url.startswith('sqlite') or self.url.startswith('mysql'): - self.assertRaises(exceptions.NotSupportedError, - self.index.rename, index_name2) - else: - assert_index_name(index_name1) - rename_index(self.index, index_name2, engine=self.engine) - assert_index_name(index_name2) - self.index.rename(index_name1) - assert_index_name(index_name1) - - # test by just the string - rename_index(index_name1, index_name2, engine=self.engine) - assert_index_name(index_name2, True) - - finally: - if self.table.exists(): - self.table.drop() - - -class TestColumnChange(fixture.DB): - level = fixture.DB.CONNECT - table_name = 'tmp_colchange' - - def _setup(self, url): - super(TestColumnChange, self)._setup(url) - self.meta = MetaData(self.engine) - self.table = Table(self.table_name, self.meta, - Column('id', Integer, primary_key=True), - Column('data', String(40), server_default=DefaultClause("tluafed"), - nullable=True), - ) - if self.table.exists(): - self.table.drop() - try: - self.table.create() - except sqlalchemy.exceptions.SQLError, e: - # SQLite: database schema has changed - if not self.url.startswith('sqlite://'): - raise - - def _teardown(self): - if self.table.exists(): - try: - self.table.drop(self.engine) - except sqlalchemy.exceptions.SQLError,e: - # SQLite: database schema has changed - if not self.url.startswith('sqlite://'): - raise - super(TestColumnChange, self)._teardown() - - @fixture.usedb() - def test_rename(self): - """Can rename a column""" - def num_rows(col, content): - return len(list(self.table.select(col == content).execute())) - # Table content should be preserved in changed columns - content = "fgsfds" - self.engine.execute(self.table.insert(), data=content, id=42) - self.assertEquals(num_rows(self.table.c.data, content), 1) - - # ...as a function, given a column object and the new name - alter_column('data', name='data2', table=self.table) - self.refresh_table() - alter_column(self.table.c.data2, name='atad') - self.refresh_table(self.table.name) - self.assert_('data' not in self.table.c.keys()) - self.assert_('atad' in self.table.c.keys()) - self.assertEquals(num_rows(self.table.c.atad, content), 1) - - # ...as a method, given a new name - self.table.c.atad.alter(name='data') - self.refresh_table(self.table.name) - self.assert_('atad' not in self.table.c.keys()) - self.table.c.data # Should not raise exception - self.assertEquals(num_rows(self.table.c.data, content), 1) - - # ...as a function, given a new object - alter_column(self.table.c.data, - name = 'atad', type=String(40), - server_default=self.table.c.data.server_default) - self.refresh_table(self.table.name) - self.assert_('data' not in self.table.c.keys()) - self.table.c.atad # Should not raise exception - self.assertEquals(num_rows(self.table.c.atad, content), 1) - - # ...as a method, given a new object - self.table.c.atad.alter( - name='data',type=String(40), - server_default=self.table.c.atad.server_default - ) - self.refresh_table(self.table.name) - self.assert_('atad' not in self.table.c.keys()) - self.table.c.data # Should not raise exception - self.assertEquals(num_rows(self.table.c.data,content), 1) - - @fixture.usedb() - def test_type(self): - # Test we can change a column's type - - # Just the new type - self.table.c.data.alter(type=String(43)) - self.refresh_table(self.table.name) - self.assert_(isinstance(self.table.c.data.type, String)) - self.assertEquals(self.table.c.data.type.length, 43) - - # Different type - self.assert_(isinstance(self.table.c.id.type, Integer)) - self.assertEquals(self.table.c.id.nullable, False) - - if not self.engine.name == 'firebird': - self.table.c.id.alter(type=String(20)) - self.assertEquals(self.table.c.id.nullable, False) - self.refresh_table(self.table.name) - self.assert_(isinstance(self.table.c.id.type, String)) - - @fixture.usedb() - def test_default(self): - """Can change a column's server_default value (DefaultClauses only) - Only DefaultClauses are changed here: others are managed by the - application / by SA - """ - self.assertEquals(self.table.c.data.server_default.arg, 'tluafed') - - # Just the new default - default = 'my_default' - self.table.c.data.alter(server_default=DefaultClause(default)) - self.refresh_table(self.table.name) - #self.assertEquals(self.table.c.data.server_default.arg,default) - # TextClause returned by autoload - self.assert_(default in str(self.table.c.data.server_default.arg)) - self.engine.execute(self.table.insert(), id=12) - row = self._select_row() - self.assertEqual(row['data'], default) - - # Column object - default = 'your_default' - self.table.c.data.alter(type=String(40), server_default=DefaultClause(default)) - self.refresh_table(self.table.name) - self.assert_(default in str(self.table.c.data.server_default.arg)) - - # Drop/remove default - self.table.c.data.alter(server_default=None) - self.assertEqual(self.table.c.data.server_default, None) - - self.refresh_table(self.table.name) - # server_default isn't necessarily None for Oracle - #self.assert_(self.table.c.data.server_default is None,self.table.c.data.server_default) - self.engine.execute(self.table.insert(), id=11) - row = self.table.select(self.table.c.id == 11).execution_options(autocommit=True).execute().fetchone() - self.assert_(row['data'] is None, row['data']) - - @fixture.usedb(not_supported='firebird') - def test_null(self): - """Can change a column's null constraint""" - self.assertEquals(self.table.c.data.nullable, True) - - # Full column - self.table.c.data.alter(type=String(40), nullable=False) - self.table.nullable = None - self.refresh_table(self.table.name) - self.assertEquals(self.table.c.data.nullable, False) - - # Just the new status - self.table.c.data.alter(nullable=True) - self.refresh_table(self.table.name) - self.assertEquals(self.table.c.data.nullable, True) - - @fixture.usedb() - def test_alter_deprecated(self): - try: - # py 2.4 compatability :-/ - cw = catch_warnings(record=True) - w = cw.__enter__() - - warnings.simplefilter("always") - self.table.c.data.alter(Column('data', String(100))) - - self.assertEqual(len(w),1) - self.assertTrue(issubclass(w[-1].category, - MigrateDeprecationWarning)) - self.assertEqual( - 'Passing a Column object to alter_column is deprecated. ' - 'Just pass in keyword parameters instead.', - str(w[-1].message)) - finally: - cw.__exit__() - - @fixture.usedb() - def test_alter_returns_delta(self): - """Test if alter constructs return delta""" - - delta = self.table.c.data.alter(type=String(100)) - self.assert_('type' in delta) - - @fixture.usedb() - def test_alter_all(self): - """Tests all alter changes at one time""" - # test for each db separately - # since currently some dont support everything - - # test pre settings - self.assertEqual(self.table.c.data.nullable, True) - self.assertEqual(self.table.c.data.server_default.arg, 'tluafed') - self.assertEqual(self.table.c.data.name, 'data') - self.assertTrue(isinstance(self.table.c.data.type, String)) - self.assertTrue(self.table.c.data.type.length, 40) - - kw = dict(nullable=False, - server_default='foobar', - name='data_new', - type=String(50)) - if self.engine.name == 'firebird': - del kw['nullable'] - self.table.c.data.alter(**kw) - - # test altered objects - self.assertEqual(self.table.c.data.server_default.arg, 'foobar') - if not self.engine.name == 'firebird': - self.assertEqual(self.table.c.data.nullable, False) - self.assertEqual(self.table.c.data.name, 'data_new') - self.assertEqual(self.table.c.data.type.length, 50) - - self.refresh_table(self.table.name) - - # test post settings - if not self.engine.name == 'firebird': - self.assertEqual(self.table.c.data_new.nullable, False) - self.assertEqual(self.table.c.data_new.name, 'data_new') - self.assertTrue(isinstance(self.table.c.data_new.type, String)) - self.assertTrue(self.table.c.data_new.type.length, 50) - - # insert data and assert default - self.table.insert(values={'id': 10}).execute() - row = self._select_row() - self.assertEqual(u'foobar', row['data_new']) - - -class TestColumnDelta(fixture.DB): - """Tests ColumnDelta class""" - - level = fixture.DB.CONNECT - table_name = 'tmp_coldelta' - table_int = 0 - - def _setup(self, url): - super(TestColumnDelta, self)._setup(url) - self.meta = MetaData() - self.table = Table(self.table_name, self.meta, - Column('ids', String(10)), - ) - self.meta.bind = self.engine - if self.engine.has_table(self.table.name): - self.table.drop() - self.table.create() - - def _teardown(self): - if self.engine.has_table(self.table.name): - self.table.drop() - self.meta.clear() - super(TestColumnDelta,self)._teardown() - - def mkcol(self, name='id', type=String, *p, **k): - return Column(name, type, *p, **k) - - def verify(self, expected, original, *p, **k): - self.delta = ColumnDelta(original, *p, **k) - result = self.delta.keys() - result.sort() - self.assertEquals(expected, result) - return self.delta - - def test_deltas_two_columns(self): - """Testing ColumnDelta with two columns""" - col_orig = self.mkcol(primary_key=True) - col_new = self.mkcol(name='ids', primary_key=True) - self.verify([], col_orig, col_orig) - self.verify(['name'], col_orig, col_orig, 'ids') - self.verify(['name'], col_orig, col_orig, name='ids') - self.verify(['name'], col_orig, col_new) - self.verify(['name', 'type'], col_orig, col_new, type=String) - - # Type comparisons - self.verify([], self.mkcol(type=String), self.mkcol(type=String)) - self.verify(['type'], self.mkcol(type=String), self.mkcol(type=Integer)) - self.verify(['type'], self.mkcol(type=String), self.mkcol(type=String(42))) - self.verify([], self.mkcol(type=String(42)), self.mkcol(type=String(42))) - self.verify(['type'], self.mkcol(type=String(24)), self.mkcol(type=String(42))) - self.verify(['type'], self.mkcol(type=String(24)), self.mkcol(type=Text(24))) - - # Other comparisons - self.verify(['primary_key'], self.mkcol(nullable=False), self.mkcol(primary_key=True)) - - # PK implies nullable=False - self.verify(['nullable', 'primary_key'], self.mkcol(nullable=True), self.mkcol(primary_key=True)) - self.verify([], self.mkcol(primary_key=True), self.mkcol(primary_key=True)) - self.verify(['nullable'], self.mkcol(nullable=True), self.mkcol(nullable=False)) - self.verify([], self.mkcol(nullable=True), self.mkcol(nullable=True)) - self.verify([], self.mkcol(server_default=None), self.mkcol(server_default=None)) - self.verify([], self.mkcol(server_default='42'), self.mkcol(server_default='42')) - - # test server default - delta = self.verify(['server_default'], self.mkcol(), self.mkcol('id', String, DefaultClause('foobar'))) - self.assertEqual(delta['server_default'].arg, 'foobar') - - self.verify([], self.mkcol(server_default='foobar'), self.mkcol('id', String, DefaultClause('foobar'))) - self.verify(['type'], self.mkcol(server_default='foobar'), self.mkcol('id', Text, DefaultClause('foobar'))) - - col = self.mkcol(server_default='foobar') - self.verify(['type'], col, self.mkcol('id', Text, DefaultClause('foobar')), alter_metadata=True) - self.assert_(isinstance(col.type, Text)) - - col = self.mkcol() - self.verify(['name', 'server_default', 'type'], col, self.mkcol('beep', Text, DefaultClause('foobar')), - alter_metadata=True) - self.assert_(isinstance(col.type, Text)) - self.assertEqual(col.name, 'beep') - self.assertEqual(col.server_default.arg, 'foobar') - - @fixture.usedb() - def test_deltas_zero_columns(self): - """Testing ColumnDelta with zero columns""" - - self.verify(['name'], 'ids', table=self.table, name='hey') - - # test reflection - self.verify(['type'], 'ids', table=self.table.name, type=String(80), engine=self.engine) - self.verify(['type'], 'ids', table=self.table.name, type=String(80), metadata=self.meta) - - self.meta.clear() - delta = self.verify(['type'], 'ids', table=self.table.name, type=String(80), metadata=self.meta, - alter_metadata=True) - self.assert_(self.table.name in self.meta) - self.assertEqual(delta.result_column.type.length, 80) - self.assertEqual(self.meta.tables.get(self.table.name).c.ids.type.length, 80) - - # test defaults - self.meta.clear() - self.verify(['server_default'], 'ids', table=self.table.name, server_default='foobar', - metadata=self.meta, - alter_metadata=True) - self.meta.tables.get(self.table.name).c.ids.server_default.arg == 'foobar' - - # test missing parameters - self.assertRaises(ValueError, ColumnDelta, table=self.table.name) - self.assertRaises(ValueError, ColumnDelta, 'ids', table=self.table.name, alter_metadata=True) - self.assertRaises(ValueError, ColumnDelta, 'ids', table=self.table.name, alter_metadata=False) - - def test_deltas_one_column(self): - """Testing ColumnDelta with one column""" - col_orig = self.mkcol(primary_key=True) - - self.verify([], col_orig) - self.verify(['name'], col_orig, 'ids') - # Parameters are always executed, even if they're 'unchanged' - # (We can't assume given column is up-to-date) - self.verify(['name', 'primary_key', 'type'], col_orig, 'id', Integer, primary_key=True) - self.verify(['name', 'primary_key', 'type'], col_orig, name='id', type=Integer, primary_key=True) - - # Change name, given an up-to-date definition and the current name - delta = self.verify(['name'], col_orig, name='blah') - self.assertEquals(delta.get('name'), 'blah') - self.assertEquals(delta.current_name, 'id') - - col_orig = self.mkcol(primary_key=True) - self.verify(['name', 'type'], col_orig, name='id12', type=Text, alter_metadata=True) - self.assert_(isinstance(col_orig.type, Text)) - self.assertEqual(col_orig.name, 'id12') - - # test server default - col_orig = self.mkcol(primary_key=True) - delta = self.verify(['server_default'], col_orig, DefaultClause('foobar')) - self.assertEqual(delta['server_default'].arg, 'foobar') - - delta = self.verify(['server_default'], col_orig, server_default=DefaultClause('foobar')) - self.assertEqual(delta['server_default'].arg, 'foobar') - - # no change - col_orig = self.mkcol(server_default=DefaultClause('foobar')) - delta = self.verify(['type'], col_orig, DefaultClause('foobar'), type=PickleType) - self.assert_(isinstance(delta.result_column.type, PickleType)) - - # TODO: test server on update - # TODO: test bind metadata |