from testlib import *
-class DependencySortTest(PersistTest):
+class DependencySortTest(TestBase):
def assert_sort(self, tuples, node, collection=None):
print str(node)
def assert_tuple(tuple, node):
from testlib import *
from testlib import sorted
-class OrderedDictTest(PersistTest):
+class OrderedDictTest(TestBase):
def test_odict(self):
o = util.OrderedDict()
o['a'] = 1
self.assert_(o.keys() == ['a', 'b', 'c', 'd', 'e', 'f'])
self.assert_(o.values() == [1, 2, 3, 4, 5, 6])
-class OrderedSetTest(PersistTest):
+class OrderedSetTest(TestBase):
def test_mutators_against_iter(self):
# testing a set modified against an iterator
o = util.OrderedSet([3,2, 4, 5])
self.assertEquals(o.union(iter([3,4, 6])),
util.OrderedSet([2, 3, 4, 5, 6]))
-class ColumnCollectionTest(PersistTest):
+class ColumnCollectionTest(TestBase):
def test_in(self):
cc = sql.ColumnCollection()
cc.add(sql.column('col1'))
self._notok(duck6())
-class ArgInspectionTest(PersistTest):
+class ArgInspectionTest(TestBase):
def test_get_cls_kwargs(self):
class A(object):
def __init__(self, a):
from testlib import *
-class BasicTest(AssertMixin):
+class BasicTest(TestBase, AssertsExecutionResults):
# A simple import of the database/ module should work on all systems.
def test_import(self):
# we got this far, right?
from testlib import *
-class DomainReflectionTest(AssertMixin):
+class DomainReflectionTest(TestBase, AssertsExecutionResults):
"Test Firebird domains"
__only_on__ = 'firebird'
self.assertEquals(table.c.dt.type.__class__, firebird.FBDateTime)
-class CompileTest(SQLCompileTest):
+class CompileTest(TestBase, AssertsCompiledSQL):
__dialect__ = firebird.FBDialect()
def test_alias(self):
self.assert_compile(select([func.max(t.c.col1)]), "SELECT max(sometable.col1) AS max_1 FROM sometable")
-class MiscFBTests(PersistTest):
+class MiscFBTests(TestBase):
__only_on__ = 'firebird'
from testlib import *
-class BasicTest(AssertMixin):
+class BasicTest(TestBase, AssertsExecutionResults):
# A simple import of the database/ module should work on all systems.
def test_import(self):
# we got this far, right?
# - datetime tests
# - the orm/query 'test_has' destabilizes the server- cover here
-class ReflectionTest(AssertMixin):
+class ReflectionTest(TestBase, AssertsExecutionResults):
"""Extra reflection tests."""
__only_on__ = 'maxdb'
except exceptions.DatabaseError:
pass
-class DBAPITest(AssertMixin):
+class DBAPITest(TestBase, AssertsExecutionResults):
"""Asserts quirks in the native Python DB-API driver.
If any of these fail, that's good- the bug is fixed!
from testlib import *
-class CompileTest(SQLCompileTest):
+class CompileTest(TestBase, AssertsCompiledSQL):
__dialect__ = mssql.MSSQLDialect()
def test_insert(self):
t = Table('sometable', m, Column('col1', Integer), Column('col2', Integer))
self.assert_compile(select([func.max(t.c.col1)]), "SELECT max(sometable.col1) AS max_1 FROM sometable")
-class ReflectionTest(PersistTest):
+class ReflectionTest(TestBase):
__only_on__ = 'mssql'
def testidentity(self):
table.drop()
-class QueryTest(PersistTest):
+class QueryTest(TestBase):
__only_on__ = 'mssql'
def test_fetchid_trigger(self):
for k in kw:
setattr(self, k, kw[k])
-class GenerativeQueryTest(PersistTest):
+class GenerativeQueryTest(TestBase):
__only_on__ = 'mssql'
def setUpAll(self):
from testlib import *
-class TypesTest(AssertMixin):
+class TypesTest(TestBase, AssertsExecutionResults):
"Test MySQL column types"
__only_on__ = 'mysql'
self.assertEqual(got, wanted)
-class SQLTest(SQLCompileTest):
+class SQLTest(TestBase, AssertsCompiledSQL):
"""Tests MySQL-dialect specific compilation."""
__dialect__ = mysql.dialect()
from testlib import *
-class OutParamTest(AssertMixin):
+class OutParamTest(TestBase, AssertsExecutionResults):
__only_on__ = 'oracle'
def setUpAll(self):
testing.db.execute("DROP PROCEDURE foo")
-class CompileTest(SQLCompileTest):
+class CompileTest(TestBase, AssertsCompiledSQL):
__dialect__ = oracle.OracleDialect()
def test_subquery(self):
"ON addresses.address_type_id = address_types_1.id WHERE addresses.user_id = :addresses_user_id_1 ORDER BY addresses.rowid, "
"address_types.rowid")
-class TypesTest(SQLCompileTest):
+class TypesTest(TestBase, AssertsCompiledSQL):
__only_on__ = 'oracle'
def test_no_clobs_for_string_params(self):
finally:
testing.db.execute("DROP TABLE Z_TEST")
-class SequenceTest(SQLCompileTest):
+class SequenceTest(TestBase, AssertsCompiledSQL):
def test_basic(self):
seq = Sequence("my_seq_no_schema")
dialect = oracle.OracleDialect()
from sqlalchemy.sql import table, column
-class SequenceTest(SQLCompileTest):
+class SequenceTest(TestBase, AssertsCompiledSQL):
def test_basic(self):
seq = Sequence("my_seq_no_schema")
dialect = postgres.PGDialect()
seq = Sequence("My_Seq", schema="Some_Schema")
assert dialect.identifier_preparer.format_sequence(seq) == '"Some_Schema"."My_Seq"'
-class CompileTest(SQLCompileTest):
+class CompileTest(TestBase, AssertsCompiledSQL):
def test_update_returning(self):
dialect = postgres.dialect()
table1 = table('mytable',
i = insert(table1, values=dict(name='foo'), postgres_returning=[func.length(table1.c.name)])
self.assert_compile(i, "INSERT INTO mytable (name) VALUES (%(name)s) RETURNING length(mytable.name)", dialect=dialect)
-class ReturningTest(AssertMixin):
+class ReturningTest(TestBase, AssertsExecutionResults):
__only_on__ = 'postgres'
@testing.exclude('postgres', '<', (8, 2))
table.drop()
-class InsertTest(AssertMixin):
+class InsertTest(TestBase, AssertsExecutionResults):
__only_on__ = 'postgres'
def setUpAll(self):
(33, 'd4'),
]
-class DomainReflectionTest(AssertMixin):
+class DomainReflectionTest(TestBase, AssertsExecutionResults):
"Test PostgreSQL domains"
__only_on__ = 'postgres'
self.assertEquals(str(table.columns.answer.default.arg), '0', "Reflected default value didn't equal expected value")
self.assertTrue(table.columns.answer.nullable, "Expected reflected column to be nullable.")
-class MiscTest(AssertMixin):
+class MiscTest(TestBase, AssertsExecutionResults):
__only_on__ = 'postgres'
def test_date_reflection(self):
assert executed_sql == ['CREATE INDEX test_idx1 ON testtbl (data) WHERE testtbl.data > 5 AND testtbl.data < 10']
-class TimezoneTest(AssertMixin):
+class TimezoneTest(TestBase, AssertsExecutionResults):
"""Test timezone-aware datetimes.
psycopg will return a datetime with a tzinfo attached to it, if postgres
c = notztable.update(notztable.c.id==1).execute(name='newname')
print notztable.select(tztable.c.id==1).execute().fetchone()
-class ArrayTest(AssertMixin):
+class ArrayTest(TestBase, AssertsExecutionResults):
__only_on__ = 'postgres'
def setUpAll(self):
from testlib import *
-class TestTypes(AssertMixin):
+class TestTypes(TestBase, AssertsExecutionResults):
__only_on__ = 'sqlite'
def test_date(self):
finally:
m.drop_all()
-class DialectTest(AssertMixin):
+class DialectTest(TestBase, AssertsExecutionResults):
__only_on__ = 'sqlite'
def test_extra_reserved_words(self):
pass
raise
-class InsertTest(AssertMixin):
+class InsertTest(TestBase, AssertsExecutionResults):
"""Tests inserts and autoincrement."""
__only_on__ = 'sqlite'
from testlib import *
-class BasicTest(AssertMixin):
+class BasicTest(TestBase, AssertsExecutionResults):
# A simple import of the database/ module should work on all systems.
def test_import(self):
# we got this far, right?
from testlib import *
-class BindTest(PersistTest):
+class BindTest(TestBase):
def test_create_drop_explicit(self):
metadata = MetaData()
table = Table('test_table', metadata,
from testlib import *
-class DDLEventTest(PersistTest):
+class DDLEventTest(TestBase):
class Canary(object):
def __init__(self, schema_item, bind):
self.state = None
self.assertRaises(LookupError, metadata.append_ddl_listener, 'blah', fn)
-class DDLExecutionTest(PersistTest):
+class DDLExecutionTest(TestBase):
def mock_engine(self):
buffer = []
def executor(sql, *a, **kw):
r = eval(py)
assert list(r) == [(1,)], py
-class DDLTest(PersistTest):
+class DDLTest(TestBase):
def mock_engine(self):
executor = lambda *a, **kw: None
engine = create_engine(testing.db.name + '://',
from sqlalchemy import exceptions
from testlib import *
-class ExecuteTest(PersistTest):
+class ExecuteTest(TestBase):
def setUpAll(self):
global users, metadata
metadata = MetaData(testing.db)
from testlib import *
import pickle
-class MetaDataTest(PersistTest, ComparesTables):
+class MetaDataTest(TestBase, ComparesTables):
def test_metadata_connect(self):
metadata = MetaData()
t1 = Table('table1', metadata, Column('col1', Integer, primary_key=True),
from testlib import *
-class ParseConnectTest(PersistTest):
+class ParseConnectTest(TestBase):
def test_rfc1738(self):
for text in (
'dbtype://username:password@hostspec:110//usr/db_file.db',
assert u.host == 'hostspec' or u.host == '127.0.0.1' or (not u.host)
assert str(u) == text
-class CreateEngineTest(PersistTest):
+class CreateEngineTest(TestBase):
"""test that create_engine arguments of different types get propigated properly"""
def test_connect_query(self):
dbapi = MockDBAPI(foober='12', lala='18', fooz='somevalue')
pass
mock_dbapi = MockDBAPI()
-class PoolTest(PersistTest):
+class PoolTest(TestBase):
def setUp(self):
pool.clear_managers()
def close(self):
pass
-class MockReconnectTest(PersistTest):
+class MockReconnectTest(TestBase):
def setUp(self):
global db, dbapi
dbapi = MockDBAPI()
assert len(dbapi.connections) == 1
-class RealReconnectTest(PersistTest):
+class RealReconnectTest(TestBase):
def setUp(self):
global engine
engine = engines.reconnecting_engine()
from testlib import engines
-class ReflectionTest(PersistTest, ComparesTables):
+class ReflectionTest(TestBase, ComparesTables):
@testing.exclude('mysql', '<', (4, 1, 1))
def test_basic_reflection(self):
self.assert_(not m9.tables)
-class CreateDropTest(PersistTest):
+class CreateDropTest(TestBase):
def setUpAll(self):
global metadata, users
metadata = MetaData()
self.assert_(not Set(metadata.tables) - Set(testing.db.table_names()))
metadata.drop_all(bind=testing.db)
-class SchemaManipulationTest(PersistTest):
+class SchemaManipulationTest(TestBase):
def test_append_constraint_unique(self):
meta = MetaData()
assert len(addresses.c.user_id.foreign_keys) == 1
assert addresses.constraints == set([addresses.primary_key, fk])
-class UnicodeReflectionTest(PersistTest):
+class UnicodeReflectionTest(TestBase):
def test_basic(self):
try:
bind.dispose()
-class SchemaTest(PersistTest):
+class SchemaTest(TestBase):
def test_iteration(self):
metadata = MetaData()
metadata.drop_all()
-class HasSequenceTest(PersistTest):
+class HasSequenceTest(TestBase):
def setUpAll(self):
global metadata, users
metadata = MetaData()
from testlib import *
-class TransactionTest(PersistTest):
+class TransactionTest(TestBase):
def setUpAll(self):
global users, metadata
metadata = MetaData()
conn.close()
-class AutoRollbackTest(PersistTest):
+class AutoRollbackTest(TestBase):
def setUpAll(self):
global metadata
metadata = MetaData()
users.drop(conn2)
conn2.close()
-class ExplicitAutoCommitTest(PersistTest):
+class ExplicitAutoCommitTest(TestBase):
"""test the 'autocommit' flag on select() and text() objects.
Requires Postgres so that we may define a custom function which modifies the database.
conn2.close()
-class TLTransactionTest(PersistTest):
+class TLTransactionTest(TestBase):
def setUpAll(self):
global users, metadata, tlengine
tlengine = create_engine(testing.db.url, strategy='threadlocal')
assert c1.connection.connection is not None
-class ForUpdateTest(PersistTest):
+class ForUpdateTest(TestBase):
def setUpAll(self):
global counters, metadata
metadata = MetaData()
from testlib import *
-class testcase(PersistTest):
+class testcase(TestBase):
def setUpAll(self):
clear_mappers()
objectstore.clear()
self.assertEquals(Person.query.count(), 2)
-class testmanytomany(PersistTest):
+class testmanytomany(TestBase):
def setUpAll(self):
clear_mappers()
objectstore.clear()
foo1.bazrel.append(baz1)
assert (foo1.bazrel == [baz1])
-class testselfreferential(PersistTest):
+class testselfreferential(TestBase):
def setUpAll(self):
clear_mappers()
objectstore.clear()
from testlib import *
-class AssignMapperTest(PersistTest):
+class AssignMapperTest(TestBase):
def setUpAll(self):
global metadata, table, table2
metadata = MetaData(testing.db)
def __iter__(self):
return iter(self.values)
-class _CollectionOperations(PersistTest):
+class _CollectionOperations(TestBase):
def setUp(self):
collection_class = self.collection_class
except TypeError:
pass
-class ScalarTest(PersistTest):
+class ScalarTest(TestBase):
def test_scalar_proxy(self):
metadata = MetaData(testing.db)
p2.bar = 'quux'
-class LazyLoadTest(PersistTest):
+class LazyLoadTest(TestBase):
def setUp(self):
metadata = MetaData(testing.db)
s += chr(index + 65)
return s
-class OrderingListTest(PersistTest):
+class OrderingListTest(TestBase):
def setUp(self):
global metadata, slides_table, bullets_table, Slide, Bullet
slides_table, bullets_table = None, None
from sqlalchemy.orm import *
from testlib import *
-class AssociationTest(PersistTest):
+class AssociationTest(TestBase):
@testing.uses_deprecated('association option')
def setUpAll(self):
global items, item_keywords, keywords, metadata, Item, Keyword, KeywordAssociation
sess.flush()
self.assert_(item_keywords.count().scalar() == 0)
-class AssociationTest2(PersistTest):
+class AssociationTest2(TestBase):
def setUpAll(self):
global table_originals, table_people, table_isauthor, metadata, Originals, People, IsAuthor
metadata = MetaData(testing.db)
from testlib import *
from testlib import fixtures
-class EagerTest(AssertMixin):
+class EagerTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global dbmeta, owners, categories, tests, options, Owner, Category, Test, Option, false
dbmeta = MetaData(testing.db)
print result
assert result == [u'3 Some Category']
-class EagerTest2(AssertMixin):
+class EagerTest2(TestBase, AssertsExecutionResults):
def setUpAll(self):
global metadata, middle, left, right
metadata = MetaData(testing.db)
class MyTest(object):pass
class MyTest2(object):pass
-class AttributesTest(PersistTest):
+class AttributesTest(TestBase):
def test_basic(self):
class User(object):pass
assert False
-class BackrefTest(PersistTest):
+class BackrefTest(TestBase):
def test_manytomany(self):
class Student(object):pass
j.port = None
self.assert_(p.jack is None)
-class DeferredBackrefTest(PersistTest):
+class DeferredBackrefTest(TestBase):
def setUp(self):
global Post, Blog, called, lazy_load
called[0] = 0
lazy_load = (p1, p2, p3) = [Post("post 1"), Post("post 2"), Post("post 3")]
-class HistoryTest(PersistTest):
+class HistoryTest(TestBase):
def test_scalar(self):
class Foo(fixtures.Base):
pass
from testlib import *
import testlib.tables as tables
-class O2MCascadeTest(AssertMixin):
+class O2MCascadeTest(TestBase, AssertsExecutionResults):
def tearDown(self):
tables.delete()
self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 0)
-class M2OCascadeTest(AssertMixin):
+class M2OCascadeTest(TestBase, AssertsExecutionResults):
def tearDown(self):
ctx.current.clear()
for t in metadata.table_iterator(reverse=True):
-class M2MCascadeTest(AssertMixin):
+class M2MCascadeTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global metadata, a, b, atob
metadata = MetaData(testing.db)
assert item.id is None
assert attr.id is None
-class DoubleParentOrphanTest(AssertMixin):
+class DoubleParentOrphanTest(TestBase, AssertsExecutionResults):
"""test orphan detection for an entity with two parent relations"""
def setUpAll(self):
except exceptions.FlushError, e:
assert True
-class CollectionAssignmentOrphanTest(AssertMixin):
+class CollectionAssignmentOrphanTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global metadata, table_a, table_b
return Entity(a or str(_id), b or 'value %s' % _id, c)
-class CollectionsTest(PersistTest):
+class CollectionsTest(TestBase):
def _test_adapter(self, typecallable, creator=entity_maker,
to_set=None):
class Foo(object):
from testlib import *
-class CompileTest(AssertMixin):
+class CompileTest(TestBase, AssertsExecutionResults):
"""test various mapper compilation scenarios"""
def tearDown(self):
clear_mappers()
def __repr__(self):
return "%s(%s)" % (self.__class__.__name__, repr(self.data))
-class SelfReferentialTest(AssertMixin):
+class SelfReferentialTest(TestBase, AssertsExecutionResults):
"""tests a self-referential mapper, with an additional list of child objects."""
def setUpAll(self):
global t1, t2, metadata
sess.delete(a)
sess.flush()
-class SelfReferentialNoPKTest(AssertMixin):
+class SelfReferentialNoPKTest(TestBase, AssertsExecutionResults):
"""test self-referential relationship that joins on a column other than the primary key column"""
def setUpAll(self):
global table, meta
assert t.uuid == t2.uuid
assert t.parent.uuid == t1.uuid
-class InheritTestOne(AssertMixin):
+class InheritTestOne(TestBase, AssertsExecutionResults):
def setUpAll(self):
global parent, child1, child2, meta
meta = MetaData(testing.db)
sess.save(o3)
sess.flush()
-class BiDirectionalOneToManyTest(AssertMixin):
+class BiDirectionalOneToManyTest(TestBase, AssertsExecutionResults):
"""tests two mappers with a one-to-many relation to each other."""
def setUpAll(self):
global t1, t2, metadata
[sess.save(x) for x in [a,b,c,d,e,f]]
sess.flush()
-class BiDirectionalOneToManyTest2(AssertMixin):
+class BiDirectionalOneToManyTest2(TestBase, AssertsExecutionResults):
"""tests two mappers with a one-to-many relation to each other, with a second one-to-many on one of the mappers"""
def setUpAll(self):
global t1, t2, t3, metadata
sess.delete(c)
sess.flush()
-class OneToManyManyToOneTest(AssertMixin):
+class OneToManyManyToOneTest(TestBase, AssertsExecutionResults):
"""tests two mappers, one has a one-to-many on the other mapper, the other has a separate many-to-one relationship to the first.
two tests will have a row for each item that is dependent on the other. without the "post_update" flag, such relationships
raise an exception when dependencies are sorted."""
)
])
-class SelfReferentialPostUpdateTest(AssertMixin):
+class SelfReferentialPostUpdateTest(TestBase, AssertsExecutionResults):
"""test using post_update on a single self-referential mapper"""
def setUpAll(self):
global metadata, node_table
),
])
-class SelfReferentialPostUpdateTest2(AssertMixin):
+class SelfReferentialPostUpdateTest2(TestBase, AssertsExecutionResults):
def setUpAll(self):
global metadata, a_table
metadata = MetaData(testing.db)
from testlib import *
from testlib.tables import *
-class EntityTest(AssertMixin):
+class EntityTest(TestBase, AssertsExecutionResults):
"""tests mappers that are constructed based on "entity names", which allows the same class
to have multiple primary mappers """
for k in kwargs:
setattr(self, k, kwargs[k])
-class GenerativeQueryTest(PersistTest):
+class GenerativeQueryTest(TestBase):
def setUpAll(self):
global foo, metadata
metadata = MetaData(testing.db)
class Obj2(object):
pass
-class GenerativeTest2(PersistTest):
+class GenerativeTest2(TestBase):
def setUpAll(self):
global metadata, table1, table2
metadata = MetaData()
res = query.filter(and_(table1.c.id==table2.c.t1id,table2.c.t1id==1)).distinct()
self.assertEqual(res.count(), 1)
-class RelationsTest(AssertMixin):
+class RelationsTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
tables.create()
tables.data()
self.assert_result(list(x), tables.User, *tables.user_result[1:3])
-class CaseSensitiveTest(PersistTest):
+class CaseSensitiveTest(TestBase):
def setUpAll(self):
global metadata, table1, table2
metadata = MetaData(testing.db)
for p in r:
assert p.car_id == p.car.car_id
-class GenerativeTest(AssertMixin):
+class GenerativeTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
# cars---owned by--- people (abstract) --- has a --- status
# | ^ ^ |
from testlib import *
-class SingleInheritanceTest(AssertMixin):
+class SingleInheritanceTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
metadata = MetaData(testing.db)
global employees_table
from sqlalchemy.orm import *
from testlib import *
-class LazyTest(AssertMixin):
+class LazyTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global info_table, data_table, rel_table, metadata
metadata = MetaData(testing.db)
import testlib.tables as tables
-class MapperSuperTest(AssertMixin):
+class MapperSuperTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
tables.create()
tables.data()
{'user_id' : 7, 'addresses' : (Address, [{'address_id' : 1}])},
)
-class MapperExtensionTest(PersistTest):
+class MapperExtensionTest(TestBase):
def setUpAll(self):
tables.create()
self.assertEquals(methods, ['before_insert', 'after_insert', 'load', 'translate_row', 'populate_instance', 'get',
'translate_row', 'create_instance', 'populate_instance', 'before_update', 'after_update', 'before_delete', 'after_delete'])
-class RequirementsTest(AssertMixin):
+class RequirementsTest(TestBase, AssertsExecutionResults):
"""Tests the contract for user classes."""
def setUpAll(self):
gc.collect()
assert len(_mapper_registry) == 0
-class EnsureZeroed(AssertMixin):
+class EnsureZeroed(TestBase, AssertsExecutionResults):
def setUp(self):
_sessions.clear()
_mapper_registry.clear()
from testlib.tables import *
import testlib.tables as tables
-class MergeTest(AssertMixin):
+class MergeTest(TestBase, AssertsExecutionResults):
"""tests session.merge() functionality"""
def setUpAll(self):
tables.create()
self.name=name
self.description = description
-class O2OTest(AssertMixin):
+class O2OTest(TestBase, AssertsExecutionResults):
@testing.uses_deprecated('SessionContext')
def setUpAll(self):
global jack, port, metadata, ctx
from sqlalchemy.orm.collections import collection
from testlib import *
-class RelationTest(PersistTest):
+class RelationTest(TestBase):
"""An extended topological sort test
This is essentially an extension of the "dependency.py" topological sort
session.delete(c) # fails
session.flush()
-class RelationTest2(PersistTest):
+class RelationTest2(TestBase):
"""Tests a relationship on a column included in multiple foreign keys.
This test tests a relationship on a column that is included in multiple
assert sess.query(Employee).get([c1.company_id, 3]).reports_to.name == 'emp1'
assert sess.query(Employee).get([c2.company_id, 3]).reports_to.name == 'emp5'
-class RelationTest3(PersistTest):
+class RelationTest3(TestBase):
def setUpAll(self):
global jobs, pageversions, pages, metadata, Job, Page, PageVersion, PageComment
import datetime
import gc
-class SessionTest(AssertMixin):
+class SessionTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
tables.create()
self.assertEquals(SomeObject(id=1, data="hello", options=[SomeOtherObject(someid=1)]), SomeObject.query.one())
self.assertEquals(SomeOtherObject(someid=1), SomeOtherObject.query.filter(SomeOtherObject.someid==sso.someid).one())
-class ScopedMapperTest(PersistTest):
+class ScopedMapperTest(TestBase):
def setUpAll(self):
global metadata, table, table2
metadata = MetaData(testing.db)
Column('user_name', String(40)),
)
-class SessionContextTest(AssertMixin):
+class SessionContextTest(TestBase, AssertsExecutionResults):
def setUp(self):
clear_mappers()
from sqlalchemy.orm import *
from sqlalchemy.orm.shard import ShardedSession
from sqlalchemy.sql import operators
-from testlib import PersistTest
+from testlib import TestBase
# TODO: ShardTest can be turned into a base for further subclasses
-class ShardTest(PersistTest):
+class ShardTest(TestBase):
def setUpAll(self):
global db1, db2, db3, db4, weather_locations, weather_reports
NUM = 2500
-class LoadTest(AssertMixin):
+class LoadTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global items, meta
meta = MetaData(testing.db)
NUM = 2500
-class SaveTest(AssertMixin):
+class SaveTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global items, metadata
metadata = MetaData(testing.db)
ITERATIONS = 100
-class SessionTest(AssertMixin):
+class SessionTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global t1, t2, metadata,T1, T2
metadata = MetaData(testing.db)
from testlib import *
-class CompileTest(AssertMixin):
+class CompileTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global t1, t2, metadata
metadata = MetaData()
from sqlalchemy.pool import QueuePool
from sqlalchemy.databases import sqlite
-class QueuePoolTest(AssertMixin):
+class QueuePoolTest(TestBase, AssertsExecutionResults):
def setUp(self):
global pool
pool = QueuePool(creator = lambda: sqlite.SQLiteDialect.dbapi().connect(':memory:'), pool_size = 3, max_overflow = -1, use_threadlocal = True)
import sys
import time
import testenv; testenv.configure_for_tests()
-from testlib import engines, testing, profiling
from sqlalchemy import *
-from testlib import set
+from testlib import *
ITERATIONS = 1
_run_type = 'suite'
_running_in = lambda: _run_type
-class ZooMarkTest(testing.AssertMixin):
+class ZooMarkTest(testing.TestBase, AssertsExecutionResults):
"""Runs the ZooMark and squawks if method counts vary from the norm.
Each test has an associated `call_range`, the total number of accepted
@profiling.conditional_call_count(
_running_in, {'isolation': (1806,),
- 'suite': (1569, {'2.4': 1579})})
+ 'suite': (1569, {'2.4': 1579}, 0.20)})
def test_1_create_tables(self):
global metadata
metadata = MetaData(testing.db)
from testlib import *
-class CaseTest(PersistTest):
+class CaseTest(TestBase):
def setUpAll(self):
metadata = MetaData(testing.db)
from testlib import *
from testlib import config, engines
-class ConstraintTest(AssertMixin):
+class ConstraintTest(TestBase, AssertsExecutionResults):
def setUp(self):
global metadata
ss = events.select().execute().fetchall()
-class ConstraintCompilationTest(AssertMixin):
+class ConstraintCompilationTest(TestBase, AssertsExecutionResults):
class accum(object):
def __init__(self):
self.statements = []
from testlib import *
-class DefaultTest(PersistTest):
+class DefaultTest(TestBase):
def setUpAll(self):
global t, f, f2, ts, currenttime, metadata, default_generator
finally:
testing.db.execute("drop table speedy_users", None)
-class PKDefaultTest(PersistTest):
+class PKDefaultTest(TestBase):
def setUpAll(self):
global metadata, t1, t2
assert r.last_inserted_ids() == [2]
-class AutoIncrementTest(PersistTest):
+class AutoIncrementTest(TestBase):
def setUp(self):
global aitable, aimeta
metadata.drop_all()
-class SequenceTest(PersistTest):
+class SequenceTest(TestBase):
__unsupported_on__ = ('sqlite', 'mysql', 'mssql', 'firebird',
'sybase', 'access')
dialects.append(testing.db.dialect)
-class CompileTest(SQLCompileTest):
+class CompileTest(TestBase, AssertsCompiledSQL):
def test_compile(self):
for dialect in dialects:
bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
, checkparams={'y_1': 45, 'x_1': 17, 'y_2': 12, 'x_2': 5})
-class ExecuteTest(PersistTest):
+class ExecuteTest(TestBase):
def test_standalone_execute(self):
x = testing.db.func.current_date().execute().scalar()
from sqlalchemy.sql import util as sql_util
-class TraversalTest(AssertMixin):
+class TraversalTest(TestBase, AssertsExecutionResults):
"""test ClauseVisitor's traversal, particularly its ability to copy and modify
a ClauseElement in place."""
assert struct3 == s3
-class ClauseTest(SQLCompileTest):
+class ClauseTest(TestBase, AssertsCompiledSQL):
"""test copy-in-place behavior of various ClauseElements."""
def setUpAll(self):
self.assert_compile(Vis().traverse(s, clone=True), "SELECT * FROM table1 WHERE table1.col1 = table2.col1 AND table1.col2 = :table1_col2_1")
-class ClauseAdapterTest(SQLCompileTest):
+class ClauseAdapterTest(TestBase, AssertsCompiledSQL):
def setUpAll(self):
global t1, t2
t1 = table("table1",
"LEFT OUTER JOIN table1 AS bar ON anon_1.col1 = bar.col1")
-class SelectTest(SQLCompileTest):
+class SelectTest(TestBase, AssertsCompiledSQL):
"""tests the generative capability of Select"""
def setUpAll(self):
IDENT_LENGTH = 29
-class LabelTypeTest(PersistTest):
+class LabelTypeTest(TestBase):
def test_type(self):
m = MetaData()
t = Table('sometable', m,
assert isinstance(t.c.col1.label('hi').type, Integer)
assert isinstance(select([t.c.col2]).as_scalar().label('lala').type, Float)
-class LongLabelsTest(SQLCompileTest):
+class LongLabelsTest(TestBase, AssertsCompiledSQL):
def setUpAll(self):
global metadata, table1, maxlen
metadata = MetaData(testing.db)
from testlib import *
-class QueryTest(PersistTest):
+class QueryTest(TestBase):
def setUpAll(self):
global users, addresses, metadata
assert len(r) == 1
-class CompoundTest(PersistTest):
+class CompoundTest(TestBase):
"""test compound statements like UNION, INTERSECT, particularly their ability to nest on
different databases."""
def setUpAll(self):
self.assertEquals(found, wanted)
-class JoinTest(PersistTest):
+class JoinTest(TestBase):
"""Tests join execution.
The compiled SQL emitted by the dialect might be ANSI joins or
self.assertRows(expr, [(10, 20, 30)])
-class OperatorTest(PersistTest):
+class OperatorTest(TestBase):
def setUpAll(self):
global metadata, flds
metadata = MetaData(testing.db)
from sqlalchemy.sql import compiler
from testlib import *
-class QuoteTest(PersistTest):
+class QuoteTest(TestBase):
def setUpAll(self):
# TODO: figure out which databases/which identifiers allow special characters to be used,
# such as: spaces, quote characters, punctuation characters, set up tests for those as
assert str(x) == '''SELECT "SomeLabel" \nFROM (SELECT 'FooCol' AS "SomeLabel" \nFROM "ImATable")'''
-class PreparerTest(PersistTest):
+class PreparerTest(TestBase):
"""Test the db-agnostic quoting services of IdentifierPreparer."""
def test_unformat(self):
from testlib import *
-class FoundRowsTest(AssertMixin):
+class FoundRowsTest(TestBase, AssertsExecutionResults):
"""tests rowcount functionality"""
def setUpAll(self):
metadata = MetaData(testing.db)
column('zip')
)
-class SelectTest(SQLCompileTest):
+class SelectTest(TestBase, AssertsCompiledSQL):
def test_attribute_sanity(self):
assert hasattr(table1, 'c')
else:
self.assert_compile(s1, "SELECT %s FROM (SELECT %s FROM mytable)" % (expr,expr))
-class CRUDTest(SQLCompileTest):
+class CRUDTest(TestBase, AssertsCompiledSQL):
def test_insert(self):
# generic insert, will create bind params for all columns
self.assert_compile(insert(table1), "INSERT INTO mytable (myid, name, description) VALUES (:myid, :name, :description)")
u = table1.delete(table1.c.name==s)
self.assert_compile(u, "DELETE FROM mytable WHERE mytable.name = (SELECT myothertable.othername FROM myothertable WHERE myothertable.otherid = mytable.myid)")
-class InlineDefaultTest(SQLCompileTest):
+class InlineDefaultTest(TestBase, AssertsCompiledSQL):
def test_insert(self):
m = MetaData()
foo = Table('foo', m,
self.assert_compile(t.update(inline=True, values={'col3':'foo'}), "UPDATE test SET col1=foo(:foo_1), col2=(SELECT coalesce(max(foo.id)) AS coalesce_1 FROM foo), col3=:col3")
-class SchemaTest(SQLCompileTest):
+class SchemaTest(TestBase, AssertsCompiledSQL):
@testing.fails_on('mssql')
def test_select(self):
# these tests will fail with the MS-SQL compiler since it will alias schema-qualified tables
Column('coly', Integer),
)
-class SelectableTest(AssertMixin):
+class SelectableTest(TestBase, AssertsExecutionResults):
def testdistance(self):
# same column three times
s = select([table.c.col1.label('c2'), table.c.col1, table.c.col1.label('c1')])
assert u.corresponding_column(s.oid_column) is u.oid_column
assert u.corresponding_column(s2.oid_column) is u.oid_column
-class PrimaryKeyTest(AssertMixin):
+class PrimaryKeyTest(TestBase, AssertsExecutionResults):
def test_join_pk_collapse_implicit(self):
"""test that redundant columns in a join get 'collapsed' into a minimal primary key,
which is the root column along a chain of foreign key relationships."""
)
-class ReduceTest(AssertMixin):
+class ReduceTest(TestBase, AssertsExecutionResults):
def test_reduce(self):
meta = MetaData()
t1 = Table('t1', meta,
)
-class DerivedTest(AssertMixin):
+class DerivedTest(TestBase, AssertsExecutionResults):
def test_table(self):
meta = MetaData()
t1 = Table('t1', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))
-class AdaptTest(PersistTest):
+class AdaptTest(TestBase):
def testadapt(self):
e1 = url.URL('postgres').get_dialect()()
e2 = url.URL('mysql').get_dialect()()
-class UserDefinedTest(PersistTest):
+class UserDefinedTest(TestBase):
"""tests user-defined types."""
def testbasic(self):
def tearDownAll(self):
metadata.drop_all()
-class ColumnsTest(AssertMixin):
+class ColumnsTest(TestBase, AssertsExecutionResults):
def testcolumns(self):
expectedResults = { 'int_column': 'int_column INTEGER',
db.dialect.schemagenerator(db.dialect, db, None, None).\
get_column_specification(aCol))
-class UnicodeTest(AssertMixin):
+class UnicodeTest(TestBase, AssertsExecutionResults):
"""tests the Unicode type. also tests the TypeDecorator with instances in the types package."""
def setUpAll(self):
global unicode_table
teststr = u'aaa\x1234'
self.assert_(testing.db.func.length(teststr).scalar() == len(teststr))
-class BinaryTest(AssertMixin):
+class BinaryTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global binary_table, MyPickleType
# put a number less than the typical MySQL default BLOB size
return file(f).read(len)
-class ExpressionTest(AssertMixin):
+class ExpressionTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global test_table, meta
# processing rules on the column.
assert testing.db.execute(select([expr])).scalar() == -15
-class DateTest(AssertMixin):
+class DateTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global users_with_date, insert_data
finally:
t.drop(checkfirst=True)
-class StringTest(AssertMixin):
+class StringTest(TestBase, AssertsExecutionResults):
def test_nolen_string_deprecated(self):
metadata = MetaData(testing.db)
foo =Table('foo', metadata,
except ImportError:
return True
-class NumericTest(AssertMixin):
+class NumericTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global numeric_table, metadata
metadata = MetaData(testing.db)
assert isinstance(row['fcasdec'], util.decimal_type)
-class IntervalTest(AssertMixin):
+class IntervalTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global interval_table, metadata
metadata = MetaData(testing.db)
interval_table.insert().execute(id=1, inverval=None)
assert interval_table.select().execute().fetchone()['interval'] is None
-class BooleanTest(AssertMixin):
+class BooleanTest(TestBase, AssertsExecutionResults):
def setUpAll(self):
global bool_table
metadata = MetaData(testing.db)
from testlib.engines import utf8_engine
from sqlalchemy.sql import column
-class UnicodeSchemaTest(PersistTest):
+class UnicodeSchemaTest(TestBase):
@testing.unsupported('maxdb', 'oracle', 'sybase')
def setUpAll(self):
global unicode_bind, metadata, t1, t2, t3
meta.drop_all()
metadata.create_all()
-class EscapesDefaultsTest(testing.PersistTest):
+class EscapesDefaultsTest(testing.TestBase):
def test_default_exec(self):
metadata = MetaData(testing.db)
t1 = Table('t1', metadata,
from testlib.orm import mapper
import testlib.testing as testing
from testlib.testing import rowset
-from testlib.testing import PersistTest, AssertMixin, ORMTest, SQLCompileTest, ComparesTables
+from testlib.testing import TestBase, AssertsExecutionResults, ORMTest, AssertsCompiledSQL, ComparesTables
import testlib.profiling as profiling
import testlib.engines as engines
from testlib.compat import set, frozenset, sorted, _function_named
'mapper',
'Table', 'Column',
'rowset',
- 'PersistTest', 'AssertMixin', 'ORMTest', 'SQLCompileTest', 'ComparesTables',
+ 'TestBase', 'AssertsExecutionResults', 'ORMTest', 'AssertsCompiledSQL', 'ComparesTables',
'profiling', 'engines',
'set', 'frozenset', 'sorted', '_function_named')
sql, sqltypes, schema, MetaData, clear_mappers, Session, util = None, None, None, None, None, None, None
sa_exceptions = None
-__all__ = ('PersistTest', 'AssertMixin', 'ComparesTables', 'ORMTest', 'SQLCompileTest')
+__all__ = ('TestBase', 'AssertsExecutionResults', 'ComparesTables', 'ORMTest', 'AssertsCompiledSQL')
_ops = { '<': operator.lt,
'>': operator.gt,
query = re.sub(r':([\w_]+)', repl, query)
return query
-class PersistTest(unittest.TestCase):
+class TestBase(unittest.TestCase):
# A sequence of dialect names to exclude from the test class.
__unsupported_on__ = ()
if not hasattr(unittest.TestCase, 'assertFalse'):
assertFalse = unittest.TestCase.failIf
-class SQLCompileTest(PersistTest):
+class AssertsCompiledSQL(object):
def assert_compile(self, clause, result, params=None, checkparams=None, dialect=None):
if dialect is None:
dialect = getattr(self, '__dialect__', None)
assert reflected_table.primary_key.columns[c.name]
-class AssertMixin(PersistTest):
+class AssertsExecutionResults(object):
def assert_result(self, result, class_, *objects):
result = list(result)
print repr(result)
testdata.buffer = None
_otest_metadata = None
-class ORMTest(AssertMixin):
+class ORMTest(TestBase, AssertsExecutionResults):
keep_mappers = False
keep_data = False
metadata = None
"""A TestSuite with once per TestCase setUpAll() and tearDownAll()"""
def __init__(self, tests=()):
- if len(tests) > 0 and isinstance(tests[0], PersistTest):
+ if len(tests) > 0 and isinstance(tests[0], TestBase):
self._initTest = tests[0]
else:
self._initTest = None
from zblog.blog import *
-class ZBlogTest(AssertMixin):
+class ZBlogTest(TestBase, AssertsExecutionResults):
def create_tables(self):
tables.metadata.drop_all(bind=testing.db)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
