@testing.provide_metadata
def test_infinite_float(self):
+ metadata = self.metadata
t = Table('t', metadata,
Column('data', Float)
)
@testing.provide_metadata
def test_strlen(self):
+ metadata = self.metadata
# On FB the length() function is implemented by an external UDF,
# strlen(). Various SA tests fail because they pass a parameter
@testing.provide_metadata
def test_rowcount_flag(self):
+ metadata = self.metadata
engine = engines.testing_engine(options={'enable_rowcount'
: True})
assert engine.dialect.supports_sane_rowcount
@testing.provide_metadata
def test_identity(self):
+ metadata = self.metadata
table = Table(
'identity_test', metadata,
Column('col1', Integer, Sequence('fred', 2, 3), primary_key=True)
@testing.emits_warning("Did not recognize")
@testing.provide_metadata
def test_skip_types(self):
+ metadata = self.metadata
testing.db.execute("""
create table foo (id integer primary key, data xml)
""")
@testing.provide_metadata
def test_indexes_cols(self):
+ metadata = self.metadata
t1 = Table('t', metadata, Column('x', Integer), Column('y', Integer))
Index('foo', t1.c.x, t1.c.y)
@testing.provide_metadata
def test_indexes_cols_with_commas(self):
+ metadata = self.metadata
t1 = Table('t', metadata,
Column('x, col', Integer, key='x'),
@testing.provide_metadata
def test_indexes_cols_with_spaces(self):
+ metadata = self.metadata
t1 = Table('t', metadata, Column('x col', Integer, key='x'),
Column('y', Integer))
@testing.provide_metadata
def test_rowid(self):
+ metadata = self.metadata
t = Table('t1', metadata,
Column('x', Integer)
)
@testing.provide_metadata
def test_numerics_broken_inspection(self):
+ metadata = self.metadata
"""Numeric scenarios where Oracle type info is 'broken',
returning us precision, scale of the form (0, 0) or (0, -127).
We convert to Decimal and let int()/float() processors take over.
@testing.provide_metadata
def test_output_type_handler(self):
+ metadata = self.metadata
for stmt, exp, kw in [
("SELECT 0.1 FROM DUAL", decimal.Decimal("0.1"), {}),
("SELECT 15 FROM DUAL", 15, {}),
@testing.provide_metadata
def test_limit_offset_for_update(self):
+ metadata = self.metadata
# oracle can't actually do the ROWNUM thing with FOR UPDATE
# very well.
@testing.only_on('postgresql', 'foo')
@testing.provide_metadata
def test_reverse_eng_name(self):
+ metadata = self.metadata
engine = engines.testing_engine(options=dict(implicit_returning=False))
for tname, cname in [
('tb1' * 30, 'abc'),
@testing.provide_metadata
def test_arrays(self):
+ metadata = self.metadata
t1 = Table('t', metadata,
Column('x', postgresql.ARRAY(Float)),
Column('y', postgresql.ARRAY(postgresql.REAL)),
@testing.provide_metadata
def test_numeric_default(self):
+ metadata = self.metadata
# pg8000 appears to fail when the value is 0,
# returns an int instead of decimal.
t =Table('t', metadata,
@testing.provide_metadata
def test_date_reflection(self):
+ metadata = self.metadata
t1 = Table('pgdate', metadata, Column('date1',
DateTime(timezone=True)), Column('date2',
DateTime(timezone=False)))
@testing.provide_metadata
def test_renamed_sequence_reflection(self):
+ metadata = self.metadata
t = Table('t', metadata, Column('id', Integer, primary_key=True))
metadata.create_all()
m2 = MetaData(testing.db)
@testing.provide_metadata
def test_index_reflection(self):
+ metadata = self.metadata
""" Reflecting partial & expression-based indexes should warn
"""
@testing.only_on('postgresql', 'DB specific feature')
@testing.provide_metadata
def test_reflection(self):
+ metadata = self.metadata
t1 = Table(
't1',
metadata,
@testing.provide_metadata
def test_tuple_flag(self):
+ metadata = self.metadata
assert_raises_message(
exc.ArgumentError,
"mutable must be set to False if as_tuple is True.",
@testing.provide_metadata
def test_bit_reflection(self):
+ metadata = self.metadata
t1 = Table('t1', metadata,
Column('bit1', postgresql.BIT()),
Column('bit5', postgresql.BIT(5)),
@testing.provide_metadata
def test_type_reflection(self):
+ metadata = self.metadata
# (ask_for, roundtripped_as_if_different)
@testing.emits_warning('Did not recognize')
@testing.provide_metadata
def test_unknown_reflection(self):
+ metadata = self.metadata
t = Table('t', metadata,
Column('x', sqltypes.BINARY(16)),
Column('y', sqltypes.BINARY())
if cls.run_define_tables == 'once':
cls.define_tables(cls.metadata)
if cls.run_create_tables == 'once':
- cls.metadata.create_all()
+ cls.metadata.create_all(cls.bind)
cls.tables.update(cls.metadata.tables)
def _setup_each_tables(self):
if self.run_define_tables == 'each':
self.tables.clear()
if self.run_create_tables == 'each':
- drop_all_tables(self.metadata)
+ drop_all_tables(self.metadata, self.bind)
self.metadata.clear()
self.define_tables(self.metadata)
if self.run_create_tables == 'each':
- self.metadata.create_all()
+ self.metadata.create_all(self.bind)
self.tables.update(self.metadata.tables)
def _setup_each_inserts(self):
@classmethod
def _teardown_once_metadata_bind(cls):
if cls.run_create_tables:
- cls.metadata.drop_all()
+ drop_all_tables(cls.metadata, cls.bind)
if cls.run_dispose_bind == 'once':
cls.dispose_bind(cls.bind)
for table in self.metadata.sorted_tables:
if table not in headers:
continue
- table.bind.execute(
+ self.bind.execute(
table.insert(),
[dict(zip(headers[table], column_values))
for column_values in rows[table]])
@testing.provide_metadata
def test_no_rowcount_on_selects_inserts(self):
+ metadata = self.metadata
"""assert that rowcount is only called on deletes and updates.
This because cursor.rowcount can be expensive on some dialects
@testing.provide_metadata
def test_override_composite_fk(self):
+ metadata = self.metadata
"""Test double-remove of composite foreign key, when replaced."""
a = Table('a',
@testing.requires.views
@testing.provide_metadata
def test_views(self):
+ metadata = self.metadata
users, addresses = createTables(metadata, None)
try:
metadata.create_all()
@testing.requires.views
@testing.provide_metadata
def test_reflect_all_with_views(self):
+ metadata = self.metadata
users, addresses = createTables(metadata, None)
try:
metadata.create_all()
testing_reaper = ConnectionKiller()
-def drop_all_tables(metadata):
+def drop_all_tables(metadata, bind):
testing_reaper.close_all()
- metadata.drop_all()
+ if hasattr(bind, 'close'):
+ bind.close()
+ metadata.drop_all(bind)
@decorator
def assert_conns_closed(fn, *args, **kw):
def fail(msg):
assert False, msg
-def fixture(table, columns, *rows):
- """Insert data into table after creation."""
- def onload(event, schema_item, connection):
- insert = table.insert()
- column_names = [col.key for col in columns]
- connection.execute(insert, [dict(zip(column_names, column_values))
- for column_values in rows])
- table.append_ddl_listener('after-create', onload)
@decorator
def provide_metadata(fn, *args, **kw):
- """Provides a bound MetaData object for a single test,
- drops it afterwards."""
+ """Provide bound MetaData for a single test, dropping afterwards."""
+
metadata = schema.MetaData(db)
- context = dict(fn.func_globals)
- context['metadata'] = metadata
- # jython bug #1034
- rebound = types.FunctionType(
- fn.func_code, context, fn.func_name, fn.func_defaults,
- fn.func_closure)
+ self = args[0]
+ prev_meta = getattr(self, 'metadata', None)
+ self.metadata = metadata
try:
- return rebound(*args, **kw)
+ return fn(*args, **kw)
finally:
metadata.drop_all()
+ self.metadata = prev_meta
class adict(dict):
"""Dict keys available as attributes. Shadows."""
from test.lib.schema import Table
from test.lib.schema import Column
from sqlalchemy.orm import attributes
-from test.lib.testing import fixture
from test.orm import _base
__all__ = ()
)
@testing.provide_metadata
def test_prop_filters_defaults(self):
+ metadata = self.metadata
t = Table('t', metadata,
Column('id', Integer(), primary_key=True, test_needs_autoincrement=True),
Column('x', Integer(), nullable=False, server_default='0')
@testing.provide_metadata
def test_update_attr_names(self):
+ metadata = self.metadata
data = Table('data', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('counter', Integer, nullable=False, default=0)
def test_func_embedded_whereclause(self):
"""test can use next_value() in whereclause"""
+ metadata = self.metadata
t1 = Table('t', metadata,
Column('x', Integer)
)
def test_func_embedded_valuesbase(self):
"""test can use next_value() in values() of _ValuesBase"""
+ metadata = self.metadata
t1 = Table('t', metadata,
Column('x', Integer)
)
"""test inserted_primary_key contains [None] when
pk_col=next_value(), implicit returning is not used."""
+ metadata = self.metadata
e = engines.testing_engine(options={'implicit_returning':False})
s = Sequence("my_sequence")
metadata.bind = e
"""test inserted_primary_key contains the result when
pk_col=next_value(), when implicit returning is used."""
+ metadata = self.metadata
e = engines.testing_engine(options={'implicit_returning':True})
s = Sequence("my_sequence")
metadata.bind = e
@testing.provide_metadata
def test_table_overrides_metadata_create(self):
+ metadata = self.metadata
s1 = Sequence("s1", metadata=metadata)
s2 = Sequence("s2", metadata=metadata)
s3 = Sequence("s3")
@testing.provide_metadata
def _run_test(self, *arg, **kw):
+ metadata = self.metadata
implicit_returning = kw.pop('implicit_returning', True)
kw['primary_key'] = True
if kw.get('autoincrement', True):
like this. Testing that all backends do the same thing here.
"""
+ metadata = self.metadata
t = Table('x', metadata,
Column('y', String(10), server_default='key_one', primary_key=True),
Column('data', String(10)),
@testing.provide_metadata
def test_string_default_on_insert_with_returning(self):
"""With implicit_returning, we get a string PK default back no problem."""
+ metadata = self.metadata
t = Table('x', metadata,
Column('y', String(10), server_default='key_one', primary_key=True),
Column('data', String(10))
@testing.provide_metadata
def test_int_default_none_on_insert(self):
+ metadata = self.metadata
t = Table('x', metadata,
Column('y', Integer,
server_default='5', primary_key=True),
)
@testing.provide_metadata
def test_autoincrement_reflected_from_server_default(self):
+ metadata = self.metadata
t = Table('x', metadata,
Column('y', Integer,
server_default='5', primary_key=True),
@testing.provide_metadata
def test_int_default_none_on_insert_reflected(self):
+ metadata = self.metadata
t = Table('x', metadata,
Column('y', Integer,
server_default='5', primary_key=True),
@testing.requires.returning
@testing.provide_metadata
def test_int_default_on_insert_with_returning(self):
+ metadata = self.metadata
t = Table('x', metadata,
Column('y', Integer,
server_default='5', primary_key=True),
@testing.provide_metadata
def test_dupe_tables(self):
+ metadata = self.metadata
t1 = Table('table1', metadata,
Column('col1', Integer, primary_key=True),
Column('col2', String(20)))
from test.lib.util import picklers
from sqlalchemy.util.compat import decimal
from test.lib.util import round_decimal
+from test.engine import _base
class AdaptTest(TestBase):
def _all_dialect_modules(self):
mt = loads(dumps(meta))
-class UserDefinedTest(TestBase, AssertsCompiledSQL):
+class UserDefinedTest(_base.TablesTest, AssertsCompiledSQL):
"""tests user-defined types."""
def test_processing(self):
-
- global users
+ users = self.tables.users
users.insert().execute(
user_id=2, goofy='jack', goofy2='jack', goofy4=u'jack',
goofy7=u'jack', goofy8=12, goofy9=12)
def test_type_coerce(self):
"""test ad-hoc usage of custom types with type_coerce()."""
+ metadata = self.metadata
class MyType(types.TypeDecorator):
impl = String
)
@classmethod
- def setup_class(cls):
- global users, metadata
-
+ def define_tables(cls, metadata):
class MyType(types.UserDefinedType):
def get_col_spec(self):
return "VARCHAR(100)"
def copy(self):
return MyUnicodeType(self.impl.length)
- metadata = MetaData(testing.db)
- users = Table('type_users', metadata,
+ Table('users', metadata,
Column('user_id', Integer, primary_key = True),
# totall custom type
Column('goofy', MyType, nullable = False),
Column('goofy9', MyNewIntSubClass, nullable = False),
)
- metadata.create_all()
-
- @classmethod
- def teardown_class(cls):
- metadata.drop_all()
class UnicodeTest(TestBase, AssertsExecutionResults):
"""tests the Unicode type. also tests the TypeDecorator with instances in the types package."""
@testing.fails_on('sqlite', "Doesn't provide Decimal results natively")
@testing.provide_metadata
def test_decimal_fp(self):
+ metadata = self.metadata
t = self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45.5"))
val = testing.db.execute("select val from t").scalar()
assert isinstance(val, decimal.Decimal)
@testing.fails_on('sqlite', "Doesn't provide Decimal results natively")
@testing.provide_metadata
def test_decimal_int(self):
+ metadata = self.metadata
t = self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45"))
val = testing.db.execute("select val from t").scalar()
assert isinstance(val, decimal.Decimal)
@testing.provide_metadata
def test_ints(self):
+ metadata = self.metadata
t = self._fixture(metadata, Integer, 45)
val = testing.db.execute("select val from t").scalar()
assert isinstance(val, (int, long))
@testing.provide_metadata
def test_float(self):
+ metadata = self.metadata
t = self._fixture(metadata, Float, 46.583)
val = testing.db.execute("select val from t").scalar()
assert isinstance(val, float)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
