- capture deprecation warnings and assert they re as they should be
- re-word alter_column deprecation warning to make more sense
This commit is contained in:
parent
17f5670c3c
commit
cade5fb261
@ -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)
|
||||
|
||||
|
@ -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()
|
||||
|
84
migrate/tests/fixture/warnings.py
Executable file
84
migrate/tests/fixture/warnings.py
Executable file
@ -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
|
||||
|
||||
|
@ -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))
|
||||
|
Loading…
x
Reference in New Issue
Block a user