diff --git a/migrate/changeset/schema.py b/migrate/changeset/schema.py index a18c9ae..5375447 100644 --- a/migrate/changeset/schema.py +++ b/migrate/changeset/schema.py @@ -107,8 +107,11 @@ def alter_column(*p, **k): # deprecation if len(p) >= 2 and isinstance(p[1], sqlalchemy.Column): - warnings.warn("Alter column with comparing columns is deprecated." - " Just pass in parameters instead.", MigrateDeprecationWarning) + warnings.warn( + "Passing a Column object to alter_column is deprecated." + " Just pass in keyword parameters instead.", + MigrateDeprecationWarning + ) engine = k['engine'] delta = ColumnDelta(*p, **k) diff --git a/migrate/tests/changeset/test_changeset.py b/migrate/tests/changeset/test_changeset.py index d1a8fd2..6f2e6e8 100644 --- a/migrate/tests/changeset/test_changeset.py +++ b/migrate/tests/changeset/test_changeset.py @@ -1,13 +1,16 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- +from __future__ import with_statement 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 @@ -448,16 +451,19 @@ class TestColumnChange(fixture.DB): self.assertEquals(num_rows(self.table.c.data, content), 1) # ...as a function, given a new object - col = Column('atad', String(40), server_default=self.table.c.data.server_default) - alter_column(self.table.c.data, col) + 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 - col = Column('data', String(40), server_default=self.table.c.atad.server_default) - self.table.c.atad.alter(col) + 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 @@ -467,7 +473,7 @@ class TestColumnChange(fixture.DB): def test_type(self): """Can change a column's type""" # Entire column definition given - self.table.c.data.alter(Column('data', String(42))) + self.table.c.data.alter(name='data', type=String(42)) self.refresh_table(self.table.name) self.assert_(isinstance(self.table.c.data.type, String)) self.assertEquals(self.table.c.data.type.length, 42) @@ -509,7 +515,7 @@ class TestColumnChange(fixture.DB): # Column object default = 'your_default' - self.table.c.data.alter(Column('data', String(40), server_default=DefaultClause(default))) + self.table.c.data.alter(name='data', 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)) @@ -532,8 +538,8 @@ class TestColumnChange(fixture.DB): """Can change a column's null constraint""" self.assertEquals(self.table.c.data.nullable, True) - # Column object - self.table.c.data.alter(Column('data', String(40), nullable=False)) + # Full column + self.table.c.data.alter(name='data', type=String(40), nullable=False) self.table.nullable = None self.refresh_table(self.table.name) self.assertEquals(self.table.c.data.nullable, False) @@ -543,17 +549,32 @@ class TestColumnChange(fixture.DB): self.refresh_table(self.table.name) self.assertEquals(self.table.c.data.nullable, True) + @fixture.usedb() + def test_alter_metadata_deprecated(self): + with catch_warnings(record=True) as w: + 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)) + @fixture.usedb() def test_alter_metadata(self): """Test if alter_metadata is respected""" - self.table.c.data.alter(Column('data', String(100))) + self.table.c.data.alter(name='data', type=String(100)) self.assert_(isinstance(self.table.c.data.type, String)) self.assertEqual(self.table.c.data.type.length, 100) # nothing should change - self.table.c.data.alter(Column('data', String(200)), alter_metadata=False) + self.table.c.data.alter(name='data', type=String(200), + alter_metadata=False) self.assert_(isinstance(self.table.c.data.type, String)) self.assertEqual(self.table.c.data.type.length, 100) @@ -561,7 +582,7 @@ class TestColumnChange(fixture.DB): def test_alter_returns_delta(self): """Test if alter constructs return delta""" - delta = self.table.c.data.alter(Column('data', String(100))) + delta = self.table.c.data.alter(name='data', type=String(100)) self.assert_('type' in delta) @fixture.usedb() diff --git a/migrate/tests/fixture/warnings.py b/migrate/tests/fixture/warnings.py new file mode 100755 index 0000000..2f54e28 --- /dev/null +++ b/migrate/tests/fixture/warnings.py @@ -0,0 +1,84 @@ +# lifted from Python 2.6, so we can use it in Python 2.5 +import sys + +class WarningMessage(object): + + """Holds the result of a single showwarning() call.""" + + _WARNING_DETAILS = ("message", "category", "filename", "lineno", "file", + "line") + + def __init__(self, message, category, filename, lineno, file=None, + line=None): + local_values = locals() + for attr in self._WARNING_DETAILS: + setattr(self, attr, local_values[attr]) + self._category_name = category.__name__ if category else None + + def __str__(self): + return ("{message : %r, category : %r, filename : %r, lineno : %s, " + "line : %r}" % (self.message, self._category_name, + self.filename, self.lineno, self.line)) + + +class catch_warnings(object): + + """A context manager that copies and restores the warnings filter upon + exiting the context. + + The 'record' argument specifies whether warnings should be captured by a + custom implementation of warnings.showwarning() and be appended to a list + returned by the context manager. Otherwise None is returned by the context + manager. The objects appended to the list are arguments whose attributes + mirror the arguments to showwarning(). + + The 'module' argument is to specify an alternative module to the module + named 'warnings' and imported under that name. This argument is only useful + when testing the warnings module itself. + + """ + + def __init__(self, record=False, module=None): + """Specify whether to record warnings and if an alternative module + should be used other than sys.modules['warnings']. + + For compatibility with Python 3.0, please consider all arguments to be + keyword-only. + + """ + self._record = record + self._module = sys.modules['warnings'] if module is None else module + self._entered = False + + def __repr__(self): + args = [] + if self._record: + args.append("record=True") + if self._module is not sys.modules['warnings']: + args.append("module=%r" % self._module) + name = type(self).__name__ + return "%s(%s)" % (name, ", ".join(args)) + + def __enter__(self): + if self._entered: + raise RuntimeError("Cannot enter %r twice" % self) + self._entered = True + self._filters = self._module.filters + self._module.filters = self._filters[:] + self._showwarning = self._module.showwarning + if self._record: + log = [] + def showwarning(*args, **kwargs): + log.append(WarningMessage(*args, **kwargs)) + self._module.showwarning = showwarning + return log + else: + return None + + def __exit__(self, *exc_info): + if not self._entered: + raise RuntimeError("Cannot exit %r without entering first" % self) + self._module.filters = self._filters + self._module.showwarning = self._showwarning + + diff --git a/migrate/tests/versioning/test_util.py b/migrate/tests/versioning/test_util.py index 76bd1fd..64285ee 100644 --- a/migrate/tests/versioning/test_util.py +++ b/migrate/tests/versioning/test_util.py @@ -1,13 +1,17 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- +from __future__ import with_statement import os from sqlalchemy import * +from migrate.exceptions import MigrateDeprecationWarning from migrate.tests import fixture +from migrate.tests.fixture.warnings import catch_warnings from migrate.versioning.util import * +import warnings class TestUtil(fixture.Pathed): @@ -37,8 +41,18 @@ class TestUtil(fixture.Pathed): self.assertTrue(engine.dialect.encoding) # deprecated echo=True parameter - engine = construct_engine(url, echo='True') - self.assertTrue(engine.echo) + with catch_warnings(record=True) as w: + warnings.simplefilter("always") + engine = construct_engine(url, echo='True') + self.assertTrue(engine.echo) + + self.assertEqual(len(w),1) + self.assertTrue(issubclass(w[-1].category, + MigrateDeprecationWarning)) + self.assertEqual( + 'echo=True parameter is deprecated, pass ' + 'engine_arg_echo=True or engine_dict={"echo": True}', + str(w[-1].message)) # unsupported argument self.assertRaises(ValueError, construct_engine, 1) @@ -69,8 +83,20 @@ class TestUtil(fixture.Pathed): f.write("class FakeFloat(int): pass") f.close() - FakeFloat = load_model('test_load_model.FakeFloat') - self.assert_(isinstance(FakeFloat(), int)) + with catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # deprecated spelling + FakeFloat = load_model('test_load_model.FakeFloat') + self.assert_(isinstance(FakeFloat(), int)) + + self.assertEqual(len(w),1) + self.assertTrue(issubclass(w[-1].category, + MigrateDeprecationWarning)) + self.assertEqual( + 'model should be in form of module.model:User ' + 'and not module.model.User', + str(w[-1].message)) FakeFloat = load_model('test_load_model:FakeFloat') self.assert_(isinstance(FakeFloat(), int))