diff options
Diffstat (limited to 'test/orm')
44 files changed, 736 insertions, 744 deletions
diff --git a/test/orm/alltests.py b/test/orm/alltests.py index dd2bd8446..73406c00d 100644 --- a/test/orm/alltests.py +++ b/test/orm/alltests.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import unittest import inheritance.alltests as inheritance @@ -19,7 +19,7 @@ def suite(): 'orm.assorted_eager', 'orm.naturalpks', - 'orm.sessioncontext', + 'orm.sessioncontext', 'orm.unitofwork', 'orm.session', 'orm.cascade', @@ -28,7 +28,7 @@ def suite(): 'orm.merge', 'orm.pickled', 'orm.memusage', - + 'orm.cycles', 'orm.entity', @@ -49,4 +49,4 @@ def suite(): if __name__ == '__main__': - testbase.main(suite()) + testenv.main(suite()) diff --git a/test/orm/association.py b/test/orm/association.py index c64e7ddcf..4d58f43ce 100644 --- a/test/orm/association.py +++ b/test/orm/association.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * @@ -8,7 +8,7 @@ class AssociationTest(PersistTest): @testing.uses_deprecated('association option') def setUpAll(self): global items, item_keywords, keywords, metadata, Item, Keyword, KeywordAssociation - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) items = Table('items', metadata, Column('item_id', Integer, primary_key=True), Column('name', String(40)), @@ -144,7 +144,7 @@ class AssociationTest(PersistTest): class AssociationTest2(PersistTest): def setUpAll(self): global table_originals, table_people, table_isauthor, metadata, Originals, People, IsAuthor - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) table_originals = Table('Originals', metadata, Column('ID', Integer, primary_key=True), Column('Title', String(200), nullable=False), @@ -223,4 +223,4 @@ table_isauthor.c.Kind], if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/assorted_eager.py b/test/orm/assorted_eager.py index 26f890c59..c4d817ea9 100644 --- a/test/orm/assorted_eager.py +++ b/test/orm/assorted_eager.py @@ -1,6 +1,6 @@ """eager loading unittests derived from mailing list-reported problems and trac tickets.""" -import testbase +import testenv; testenv.configure_for_tests() import random, datetime from sqlalchemy import * from sqlalchemy.orm import * @@ -11,11 +11,11 @@ from testlib import fixtures class EagerTest(AssertMixin): def setUpAll(self): global dbmeta, owners, categories, tests, options, Owner, Category, Test, Option, false - dbmeta = MetaData(testbase.db) + dbmeta = MetaData(testing.db) # determine a literal value for "false" based on the dialect # FIXME: this PassiveDefault setup is bogus. - bp = Boolean().dialect_impl(testbase.db.dialect).bind_processor(testbase.db.dialect) + bp = Boolean().dialect_impl(testing.db.dialect).bind_processor(testing.db.dialect) if bp: false = str(bp(False)) elif testing.against('maxdb'): @@ -218,7 +218,7 @@ class EagerTest(AssertMixin): class EagerTest2(AssertMixin): def setUpAll(self): global metadata, middle, left, right - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) middle = Table('middle', metadata, Column('id', Integer, primary_key = True), Column('data', String(50)), @@ -260,7 +260,7 @@ class EagerTest2(AssertMixin): 'right': relation(Right, lazy=False, backref=backref('middle', lazy=False)), } ) - session = create_session(bind=testbase.db) + session = create_session(bind=testing.db) p = Middle('test1') p.left.append(Left('tag1')) p.right.append(Right('tag2')) @@ -789,10 +789,10 @@ class EagerTest8(ORMTest): ) def setUp(self): - testbase.db.execute(project_t.insert(), {'id':1}) - testbase.db.execute(task_status_t.insert(), {'id':1}) - testbase.db.execute(task_type_t.insert(), {'id':1}) - testbase.db.execute(task_t.insert(), {'title':u'task 1', 'task_type_id':1, 'status_id':1, 'prj_id':1}) + testing.db.execute(project_t.insert(), {'id':1}) + testing.db.execute(task_status_t.insert(), {'id':1}) + testing.db.execute(task_type_t.insert(), {'id':1}) + testing.db.execute(task_t.insert(), {'title':u'task 1', 'task_type_id':1, 'status_id':1, 'prj_id':1}) @testing.fails_on('maxdb') def test_nested_joins(self): @@ -920,9 +920,9 @@ class EagerTest9(ORMTest): for e in acc.entries: assert e.account is acc - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/attributes.py b/test/orm/attributes.py index 7ec838458..4b6e5e19b 100644 --- a/test/orm/attributes.py +++ b/test/orm/attributes.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import pickle import sqlalchemy.orm.attributes as attributes from sqlalchemy.orm.collections import collection @@ -17,17 +17,17 @@ class AttributesTest(PersistTest): def test_basic(self): class User(object):pass - + attributes.register_class(User) attributes.register_attribute(User, 'user_id', uselist = False, useobject=False) attributes.register_attribute(User, 'user_name', uselist = False, useobject=False) attributes.register_attribute(User, 'email_address', uselist = False, useobject=False) - + u = User() u.user_id = 7 u.user_name = 'john' u.email_address = 'lala@123.com' - + self.assert_(u.user_id == 7 and u.user_name == 'john' and u.email_address == 'lala@123.com') u._state.commit_all() self.assert_(u.user_id == 7 and u.user_name == 'john' and u.email_address == 'lala@123.com') @@ -60,7 +60,7 @@ class AttributesTest(PersistTest): pk_o2 = pickle.dumps(o2) # so... pickle is creating a new 'mt2' string after a roundtrip here, - # so we'll brute-force set it to be id-equal to the original string + # so we'll brute-force set it to be id-equal to the original string if False: o_mt2_str = [ k for k in o.__dict__ if k == 'mt2'][0] o2_mt2_str = [ k for k in o2.__dict__ if k == 'mt2'][0] @@ -97,22 +97,22 @@ class AttributesTest(PersistTest): def test_deferred(self): class Foo(object):pass - + data = {'a':'this is a', 'b':12} def loader(instance, keys): for k in keys: instance.__dict__[k] = data[k] return attributes.ATTR_WAS_SET - + attributes.register_class(Foo, deferred_scalar_loader=loader) attributes.register_attribute(Foo, 'a', uselist=False, useobject=False) attributes.register_attribute(Foo, 'b', uselist=False, useobject=False) - + f = Foo() f._state.expire_attributes(None) self.assertEquals(f.a, "this is a") self.assertEquals(f.b, 12) - + f.a = "this is some new a" f._state.expire_attributes(None) self.assertEquals(f.a, "this is a") @@ -130,7 +130,7 @@ class AttributesTest(PersistTest): del f.a self.assertEquals(f.a, None) self.assertEquals(f.b, 12) - + f._state.commit_all() self.assertEquals(f.a, None) self.assertEquals(f.b, 12) @@ -141,11 +141,11 @@ class AttributesTest(PersistTest): for k in keys: instance.__dict__[k] = data[k] return attributes.ATTR_WAS_SET - + attributes.register_class(MyTest, deferred_scalar_loader=loader) attributes.register_attribute(MyTest, 'a', uselist=False, useobject=False) attributes.register_attribute(MyTest, 'b', uselist=False, useobject=False) - + m = MyTest() m._state.expire_attributes(None) assert 'a' not in m.__dict__ @@ -153,11 +153,11 @@ class AttributesTest(PersistTest): assert 'a' not in m2.__dict__ self.assertEquals(m2.a, "this is a") self.assertEquals(m2.b, 12) - + def test_list(self): class User(object):pass class Address(object):pass - + attributes.register_class(User) attributes.register_class(Address) attributes.register_attribute(User, 'user_id', uselist = False, useobject=False) @@ -165,7 +165,7 @@ class AttributesTest(PersistTest): attributes.register_attribute(User, 'addresses', uselist = True, useobject=True) attributes.register_attribute(Address, 'address_id', uselist = False, useobject=False) attributes.register_attribute(Address, 'email_address', uselist = False, useobject=False) - + u = User() u.user_id = 7 u.user_name = 'john' @@ -185,7 +185,7 @@ class AttributesTest(PersistTest): a.email_address = 'foo@bar.com' u.addresses.append(a) self.assert_(u.user_id == 7 and u.user_name == 'heythere' and u.addresses[0].email_address == 'lala@123.com' and u.addresses[1].email_address == 'foo@bar.com') - + def test_lazytrackparent(self): """test that the "hasparent" flag works properly when lazy loaders and backrefs are used""" @@ -193,8 +193,8 @@ class AttributesTest(PersistTest): class Blog(object):pass attributes.register_class(Post) attributes.register_class(Blog) - - # set up instrumented attributes with backrefs + + # set up instrumented attributes with backrefs attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True) attributes.register_attribute(Blog, 'posts', uselist=True, extension=attributes.GenericBackrefExtension('blog'), trackparent=True, useobject=True) @@ -212,23 +212,23 @@ class AttributesTest(PersistTest): # assert connections assert p1.blog is b assert p1 in b.posts - + # manual connections b2 = Blog() p2 = Post() b2.posts.append(p2) assert attributes.has_parent(Blog, p2, 'posts') assert attributes.has_parent(Post, b2, 'blog') - + def test_inheritance(self): """tests that attributes are polymorphic""" class Foo(object):pass class Bar(Foo):pass - - + + attributes.register_class(Foo) attributes.register_class(Bar) - + def func1(): print "func1" return "this is the foo attr" @@ -241,7 +241,7 @@ class AttributesTest(PersistTest): attributes.register_attribute(Foo, 'element', uselist=False, callable_=lambda o:func1, useobject=True) attributes.register_attribute(Foo, 'element2', uselist=False, callable_=lambda o:func3, useobject=True) attributes.register_attribute(Bar, 'element', uselist=False, callable_=lambda o:func2, useobject=True) - + x = Foo() y = Bar() assert x.element == 'this is the foo attr' @@ -258,22 +258,22 @@ class AttributesTest(PersistTest): def __init__(self): states.add(self._state) Foo.__init__(self) - - + + attributes.register_class(Foo) attributes.register_class(Bar) - + b = Bar() self.assertEquals(len(states), 1) self.assertEquals(list(states)[0].obj(), b) - + def test_inheritance2(self): """test that the attribute manager can properly traverse the managed attributes of an object, if the object is of a descendant class with managed attributes in the parent class""" class Foo(object):pass class Bar(Foo):pass - + attributes.register_class(Foo) attributes.register_class(Bar) attributes.register_attribute(Foo, 'element', uselist=False, useobject=True) @@ -293,7 +293,7 @@ class AttributesTest(PersistTest): pass class Bar(fixtures.Base): pass - + attributes.register_class(Foo) attributes.register_class(Bar) @@ -311,33 +311,33 @@ class AttributesTest(PersistTest): x._state.commit_all() x.col2.append(bar4) self.assertEquals(attributes.get_history(x._state, 'col2'), ([bar4], [bar1, bar2, bar3], [])) - - def test_parenttrack(self): + + def test_parenttrack(self): class Foo(object):pass class Bar(object):pass - + attributes.register_class(Foo) attributes.register_class(Bar) - + attributes.register_attribute(Foo, 'element', uselist=False, trackparent=True, useobject=True) attributes.register_attribute(Bar, 'element', uselist=False, trackparent=True, useobject=True) - + f1 = Foo() f2 = Foo() b1 = Bar() b2 = Bar() - + f1.element = b1 b2.element = f2 - + assert attributes.has_parent(Foo, b1, 'element') assert not attributes.has_parent(Foo, b2, 'element') assert not attributes.has_parent(Foo, f2, 'element') assert attributes.has_parent(Bar, f2, 'element') - + b2.element = None assert not attributes.has_parent(Bar, f2, 'element') - + # test that double assignment doesn't accidentally reset the 'parent' flag. b3 = Bar() f4 = Foo() @@ -349,25 +349,25 @@ class AttributesTest(PersistTest): def test_mutablescalars(self): """test detection of changes on mutable scalar items""" class Foo(object):pass - + attributes.register_class(Foo) attributes.register_attribute(Foo, 'element', uselist=False, copy_function=lambda x:[y for y in x], mutable_scalars=True, useobject=False) x = Foo() - x.element = ['one', 'two', 'three'] + x.element = ['one', 'two', 'three'] x._state.commit_all() x.element[1] = 'five' assert x._state.is_modified() - + attributes.unregister_class(Foo) - + attributes.register_class(Foo) attributes.register_attribute(Foo, 'element', uselist=False, useobject=False) x = Foo() - x.element = ['one', 'two', 'three'] + x.element = ['one', 'two', 'three'] x._state.commit_all() x.element[1] = 'five' assert not x._state.is_modified() - + def test_descriptorattributes(self): """changeset: 1633 broke ability to use ORM to map classes with unusual descriptor attributes (for example, classes that inherit from ones @@ -380,16 +380,16 @@ class AttributesTest(PersistTest): class Foo(object): A = des() - + attributes.unregister_class(Foo) - + def test_collectionclasses(self): - + class Foo(object):pass attributes.register_class(Foo) attributes.register_attribute(Foo, "collection", uselist=True, typecallable=set, useobject=True) assert isinstance(Foo().collection, set) - + attributes.unregister_attribute(Foo, "collection") try: @@ -397,7 +397,7 @@ class AttributesTest(PersistTest): assert False except exceptions.ArgumentError, e: assert str(e) == "Type InstrumentedDict must elect an appender method to be a collection class" - + class MyDict(dict): @collection.appender def append(self, item): @@ -409,14 +409,14 @@ class AttributesTest(PersistTest): assert isinstance(Foo().collection, MyDict) attributes.unregister_attribute(Foo, "collection") - + class MyColl(object):pass try: attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyColl, useobject=True) assert False except exceptions.ArgumentError, e: assert str(e) == "Type MyColl must elect an appender method to be a collection class" - + class MyColl(object): @collection.iterator def __iter__(self): @@ -436,7 +436,7 @@ class AttributesTest(PersistTest): class BackrefTest(PersistTest): - + def test_manytomany(self): class Student(object):pass class Course(object):pass @@ -459,12 +459,12 @@ class BackrefTest(PersistTest): self.assert_(s2.courses == [c]) self.assert_(s1.courses == [c]) s1.courses.remove(c) - self.assert_(c.students == [s2,s3]) - + self.assert_(c.students == [s2,s3]) + def test_onetomany(self): class Post(object):pass class Blog(object):pass - + attributes.register_class(Post) attributes.register_class(Blog) attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True) @@ -511,7 +511,7 @@ class BackrefTest(PersistTest): class DeferredBackrefTest(PersistTest): def setUp(self): global Post, Blog, called, lazy_load - + class Post(object): def __init__(self, name): self.name = name @@ -600,20 +600,20 @@ class HistoryTest(PersistTest): def test_scalar(self): class Foo(fixtures.Base): pass - + attributes.register_class(Foo) attributes.register_attribute(Foo, 'someattr', uselist=False, useobject=False) # case 1. new object f = Foo() self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) - + f.someattr = "hi" self.assertEquals(attributes.get_history(f._state, 'someattr'), (['hi'], [], [])) f._state.commit(['someattr']) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['hi'], [])) - + f.someattr = 'there' self.assertEquals(attributes.get_history(f._state, 'someattr'), (['there'], [], ['hi'])) @@ -623,15 +623,15 @@ class HistoryTest(PersistTest): del f.someattr self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], ['there'])) - + # case 2. object with direct dictionary settings (similar to a load operation) f = Foo() f.__dict__['someattr'] = 'new' self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['new'], [])) - + f.someattr = 'old' self.assertEquals(attributes.get_history(f._state, 'someattr'), (['old'], [], ['new'])) - + f._state.commit(['someattr']) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['old'], [])) @@ -641,7 +641,7 @@ class HistoryTest(PersistTest): self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) f.someattr = None self.assertEquals(attributes.get_history(f._state, 'someattr'), ([None], [], [])) - + f = Foo() f.__dict__['someattr'] = 'new' self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['new'], [])) @@ -725,7 +725,7 @@ class HistoryTest(PersistTest): self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['old'], [])) # setting None on uninitialized is currently not a change for an object attribute - # (this is different than scalar attribute). a lazyload has occured so if its + # (this is different than scalar attribute). a lazyload has occured so if its # None, its really None f = Foo() self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [None], [])) @@ -744,15 +744,15 @@ class HistoryTest(PersistTest): class Bar(fixtures.Base): def __nonzero__(self): assert False - + attributes.register_class(Foo) attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True) - + hi = Bar(name='hi') there = Bar(name='there') old = Bar(name='old') new = Bar(name='new') - + # case 1. new object f = Foo() self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) @@ -769,7 +769,7 @@ class HistoryTest(PersistTest): f._state.commit(['someattr']) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [there], [])) - + f.someattr = [hi] self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [there])) @@ -796,7 +796,7 @@ class HistoryTest(PersistTest): pass from sqlalchemy.orm.collections import attribute_mapped_collection - + attributes.register_class(Foo) attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True, typecallable=attribute_mapped_collection('name')) @@ -813,7 +813,7 @@ class HistoryTest(PersistTest): f.someattr['there'] = there self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([hi, there]), set([]), set([]))) - + f._state.commit(['someattr']) self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([]), set([hi, there]), set([]))) @@ -851,16 +851,16 @@ class HistoryTest(PersistTest): f.someattr.remove(there) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi], [there])) - + f.someattr.append(old) f.someattr.append(new) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old, new], [hi], [there])) f._state.commit(['someattr']) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi, old, new], [])) - + f.someattr.pop(0) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [old, new], [hi])) - + # case 2. object with direct settings (similar to a load operation) f = Foo() f.__dict__['id'] = 1 @@ -880,11 +880,11 @@ class HistoryTest(PersistTest): collection.append_without_event(new) f._state.commit_all() self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], [])) - + f.id = 1 f.someattr.remove(new) self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [new])) - + # case 3. mixing appends with sets f = Foo() f.someattr.append(hi) @@ -893,7 +893,7 @@ class HistoryTest(PersistTest): self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi, there], [], [])) f.someattr = [there] self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [], [])) - + def test_collections_via_backref(self): class Foo(fixtures.Base): pass @@ -904,12 +904,12 @@ class HistoryTest(PersistTest): attributes.register_class(Bar) attributes.register_attribute(Foo, 'bars', uselist=True, extension=attributes.GenericBackrefExtension('foo'), trackparent=True, useobject=True) attributes.register_attribute(Bar, 'foo', uselist=False, extension=attributes.GenericBackrefExtension('bars'), trackparent=True, useobject=True) - + f1 = Foo() b1 = Bar() self.assertEquals(attributes.get_history(f1._state, 'bars'), ([], [], [])) self.assertEquals(attributes.get_history(b1._state, 'foo'), ([], [None], [])) - + #b1.foo = f1 f1.bars.append(b1) self.assertEquals(attributes.get_history(f1._state, 'bars'), ([b1], [], [])) @@ -920,7 +920,7 @@ class HistoryTest(PersistTest): self.assertEquals(attributes.get_history(f1._state, 'bars'), ([b1, b2], [], [])) self.assertEquals(attributes.get_history(b1._state, 'foo'), ([f1], [], [])) self.assertEquals(attributes.get_history(b2._state, 'foo'), ([f1], [], [])) - + def test_lazy_backref_collections(self): class Foo(fixtures.Base): pass @@ -932,7 +932,7 @@ class HistoryTest(PersistTest): def load(): return lazy_load return load - + attributes.register_class(Foo) attributes.register_class(Bar) attributes.register_attribute(Foo, 'bars', uselist=True, extension=attributes.GenericBackrefExtension('foo'), trackparent=True, callable_=lazyload, useobject=True) @@ -945,7 +945,7 @@ class HistoryTest(PersistTest): bar4 = Bar() bar4.foo = f self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar2, bar3], [])) - + lazy_load = None f = Foo() bar4 = Bar() @@ -955,7 +955,7 @@ class HistoryTest(PersistTest): lazy_load = [bar1, bar2, bar3] f._state.expire_attributes(['bars']) self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar2, bar3], [])) - + def test_collections_via_lazyload(self): class Foo(fixtures.Base): pass @@ -967,18 +967,18 @@ class HistoryTest(PersistTest): def load(): return lazy_load return load - + attributes.register_class(Foo) attributes.register_class(Bar) attributes.register_attribute(Foo, 'bars', uselist=True, callable_=lazyload, trackparent=True, useobject=True) - + bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)] lazy_load = [bar1, bar2, bar3] f = Foo() f.bars = [] self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [], [bar1, bar2, bar3])) - + f = Foo() f.bars.append(bar4) self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar2, bar3], []) ) @@ -992,7 +992,7 @@ class HistoryTest(PersistTest): f = Foo() del f.bars[1] self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar3], [bar2])) - + lazy_load = None f = Foo() f.bars.append(bar2) @@ -1012,9 +1012,9 @@ class HistoryTest(PersistTest): attributes.register_attribute(Foo, 'bar', uselist=False, callable_=lazyload, useobject=False) lazy_load = "hi" - # with scalar non-object, the lazy callable is only executed on gets, not history + # with scalar non-object, the lazy callable is only executed on gets, not history # operations - + f = Foo() self.assertEquals(f.bar, "hi") self.assertEquals(attributes.get_history(f._state, 'bar'), ([], ["hi"], [])) @@ -1035,7 +1035,7 @@ class HistoryTest(PersistTest): self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [], ["hi"])) assert f.bar is None self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], ["hi"])) - + def test_scalar_object_via_lazyload(self): class Foo(fixtures.Base): pass @@ -1047,19 +1047,19 @@ class HistoryTest(PersistTest): def load(): return lazy_load return load - + attributes.register_class(Foo) attributes.register_class(Bar) attributes.register_attribute(Foo, 'bar', uselist=False, callable_=lazyload, trackparent=True, useobject=True) bar1, bar2 = [Bar(id=1), Bar(id=2)] lazy_load = bar1 - # with scalar object, the lazy callable is only executed on gets and history + # with scalar object, the lazy callable is only executed on gets and history # operations f = Foo() self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [bar1], [])) - + f = Foo() f.bar = None self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1])) @@ -1069,13 +1069,13 @@ class HistoryTest(PersistTest): self.assertEquals(attributes.get_history(f._state, 'bar'), ([bar2], [], [bar1])) f.bar = bar1 self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [bar1], [])) - + f = Foo() self.assertEquals(f.bar, bar1) del f.bar self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1])) assert f.bar is None self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1])) - + if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/cascade.py b/test/orm/cascade.py index c86b0ab5e..aa497d81b 100644 --- a/test/orm/cascade.py +++ b/test/orm/cascade.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions @@ -189,7 +189,7 @@ class M2OCascadeTest(AssertMixin): def setUpAll(self): global ctx, data, metadata, User, Pref, Extra ctx = SessionContext(create_session) - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) extra = Table("extra", metadata, Column("extra_id", Integer, Sequence("extra_id_seq", optional=True), primary_key=True), Column("prefs_id", Integer, ForeignKey("prefs.prefs_id")) @@ -278,7 +278,7 @@ class M2OCascadeTest(AssertMixin): class M2MCascadeTest(AssertMixin): def setUpAll(self): global metadata, a, b, atob - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) a = Table('a', metadata, Column('id', Integer, primary_key=True), Column('data', String(30)) @@ -475,7 +475,7 @@ class DoubleParentOrphanTest(AssertMixin): def setUpAll(self): global metadata, address_table, businesses, homes - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) address_table = Table('addresses', metadata, Column('address_id', Integer, primary_key=True), Column('street', String(30)), @@ -541,7 +541,7 @@ class CollectionAssignmentOrphanTest(AssertMixin): def setUpAll(self): global metadata, table_a, table_b - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) table_a = Table('a', metadata, Column('id', Integer, primary_key=True), Column('foo', String(30))) @@ -593,4 +593,4 @@ class CollectionAssignmentOrphanTest(AssertMixin): assert table_b.count().scalar() == 3 if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/collection.py b/test/orm/collection.py index 4db92f31d..7addb8687 100644 --- a/test/orm/collection.py +++ b/test/orm/collection.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * import sqlalchemy.exceptions as exceptions from sqlalchemy.orm import create_session, mapper, relation, \ @@ -1429,4 +1429,4 @@ class DictHelpersTest(ORMTest): self._test_composite_mapped(collection_class) if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/compile.py b/test/orm/compile.py index 28f33c2a8..5bcca57df 100644 --- a/test/orm/compile.py +++ b/test/orm/compile.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -9,12 +9,12 @@ class CompileTest(AssertMixin): """test various mapper compilation scenarios""" def tearDown(self): clear_mappers() - + def testone(self): global metadata, order, employee, product, tax, orderproduct - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) - order = Table('orders', metadata, + order = Table('orders', metadata, Column('id', Integer, primary_key=True), Column('employee_id', Integer, ForeignKey('employees.id'), nullable=False), Column('type', Unicode(16))) @@ -47,9 +47,9 @@ class CompileTest(AssertMixin): order_join = order.select().alias('pjoin') - order_mapper = mapper(Order, order, - select_table=order_join, - polymorphic_on=order_join.c.type, + order_mapper = mapper(Order, order, + select_table=order_join, + polymorphic_on=order_join.c.type, polymorphic_identity='order', properties={ 'orderproducts': relation(OrderProduct, lazy=True, backref='order')} @@ -65,7 +65,7 @@ class CompileTest(AssertMixin): 'orders': relation(Order, lazy=True, backref='employee')}) mapper(OrderProduct, orderproduct) - + # this requires that the compilation of order_mapper's "surrogate mapper" occur after # the initial setup of MapperProperty objects on the mapper. class_mapper(Product).compile() @@ -73,9 +73,9 @@ class CompileTest(AssertMixin): def testtwo(self): """test that conflicting backrefs raises an exception""" global metadata, order, employee, product, tax, orderproduct - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) - order = Table('orders', metadata, + order = Table('orders', metadata, Column('id', Integer, primary_key=True), Column('type', Unicode(16))) @@ -100,9 +100,9 @@ class CompileTest(AssertMixin): order_join = order.select().alias('pjoin') - order_mapper = mapper(Order, order, - select_table=order_join, - polymorphic_on=order_join.c.type, + order_mapper = mapper(Order, order, + select_table=order_join, + polymorphic_on=order_join.c.type, polymorphic_identity='order', properties={ 'orderproducts': relation(OrderProduct, lazy=True, backref='product')} @@ -122,12 +122,12 @@ class CompileTest(AssertMixin): assert str(e).index("Error creating backref ") > -1 def testthree(self): - metadata = MetaData(testbase.db) - node_table = Table("node", metadata, + metadata = MetaData(testing.db) + node_table = Table("node", metadata, Column('node_id', Integer, primary_key=True), Column('name_index', Integer, nullable=True), ) - node_name_table = Table("node_name", metadata, + node_name_table = Table("node_name", metadata, Column('node_name_id', Integer, primary_key=True), Column('node_id', Integer, ForeignKey('node.node_id')), Column('host_id', Integer, ForeignKey('host.host_id')), @@ -144,7 +144,7 @@ class CompileTest(AssertMixin): class Node(object):pass class NodeName(object):pass class Host(object):pass - + node_mapper = mapper(Node, node_table) host_mapper = mapper(Host, host_table) node_name_mapper = mapper(NodeName, node_name_table, @@ -160,7 +160,7 @@ class CompileTest(AssertMixin): def testfour(self): meta = MetaData() - + a = Table('a', meta, Column('id', Integer, primary_key=True)) b = Table('b', meta, Column('id', Integer, primary_key=True), Column('a_id', Integer, ForeignKey('a.id'))) @@ -173,7 +173,7 @@ class CompileTest(AssertMixin): mapper(B, b, properties={ 'a':relation(A, backref='b') }) - + try: compile_mappers() assert False @@ -181,4 +181,4 @@ class CompileTest(AssertMixin): assert str(e).index("Error creating backref") > -1 if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/cycles.py b/test/orm/cycles.py index 00c8c29e1..be61a84ff 100644 --- a/test/orm/cycles.py +++ b/test/orm/cycles.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -24,7 +24,7 @@ class SelfReferentialTest(AssertMixin): """tests a self-referential mapper, with an additional list of child objects.""" def setUpAll(self): global t1, t2, metadata - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) t1 = Table('t1', metadata, Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True), Column('parent_c1', Integer, ForeignKey('t1.c1')), @@ -110,7 +110,7 @@ class SelfReferentialNoPKTest(AssertMixin): """test self-referential relationship that joins on a column other than the primary key column""" def setUpAll(self): global table, meta - meta = MetaData(testbase.db) + meta = MetaData(testing.db) table = Table('item', meta, Column('id', Integer, primary_key=True), Column('uuid', String(32), unique=True, nullable=False), @@ -155,7 +155,7 @@ class SelfReferentialNoPKTest(AssertMixin): class InheritTestOne(AssertMixin): def setUpAll(self): global parent, child1, child2, meta - meta = MetaData(testbase.db) + meta = MetaData(testing.db) parent = Table("parent", meta, Column("id", Integer, primary_key=True), Column("parent_data", String(50)), @@ -357,7 +357,7 @@ class BiDirectionalOneToManyTest(AssertMixin): """tests two mappers with a one-to-many relation to each other.""" def setUpAll(self): global t1, t2, metadata - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) t1 = Table('t1', metadata, Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True), Column('c2', Integer, ForeignKey('t2.c1')) @@ -398,7 +398,7 @@ class BiDirectionalOneToManyTest2(AssertMixin): """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 - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) t1 = Table('t1', metadata, Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True), Column('c2', Integer, ForeignKey('t2.c1')), @@ -460,7 +460,7 @@ class OneToManyManyToOneTest(AssertMixin): raise an exception when dependencies are sorted.""" def setUpAll(self): global metadata - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) global person global ball ball = Table('ball', metadata, @@ -534,7 +534,7 @@ class OneToManyManyToOneTest(AssertMixin): sess.save(b) sess.save(p) - self.assert_sql(testbase.db, lambda: sess.flush(), [ + self.assert_sql(testing.db, lambda: sess.flush(), [ ( "INSERT INTO person (favorite_ball_id, data) VALUES (:favorite_ball_id, :data)", {'favorite_ball_id': None, 'data':'some data'} @@ -588,7 +588,7 @@ class OneToManyManyToOneTest(AssertMixin): ) ]) sess.delete(p) - self.assert_sql(testbase.db, lambda: sess.flush(), [ + self.assert_sql(testing.db, lambda: sess.flush(), [ # heres the post update (which is a pre-update with deletes) ( "UPDATE person SET favorite_ball_id=:favorite_ball_id WHERE person.id = :person_id", @@ -638,7 +638,7 @@ class OneToManyManyToOneTest(AssertMixin): sess = create_session() [sess.save(x) for x in [b,p,b2,b3,b4]] - self.assert_sql(testbase.db, lambda: sess.flush(), [ + self.assert_sql(testing.db, lambda: sess.flush(), [ ( "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)", {'person_id':None, 'data':'some data'} @@ -717,7 +717,7 @@ class OneToManyManyToOneTest(AssertMixin): ]) sess.delete(p) - self.assert_sql(testbase.db, lambda: sess.flush(), [ + self.assert_sql(testing.db, lambda: sess.flush(), [ ( "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id", lambda ctx:{'person_id': None, 'ball_id': b.id} @@ -748,7 +748,7 @@ class SelfReferentialPostUpdateTest(AssertMixin): """test using post_update on a single self-referential mapper""" def setUpAll(self): global metadata, node_table - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) node_table = Table('node', metadata, Column('id', Integer, Sequence('nodeid_id_seq', optional=True), primary_key=True), Column('path', String(50), nullable=False), @@ -829,7 +829,7 @@ class SelfReferentialPostUpdateTest(AssertMixin): remove_child(root, cats) # pre-trigger lazy loader on 'cats' to make the test easier cats.children - self.assert_sql(testbase.db, lambda: session.flush(), [ + self.assert_sql(testing.db, lambda: session.flush(), [ ( "UPDATE node SET prev_sibling_id=:prev_sibling_id WHERE node.id = :node_id", lambda ctx:{'prev_sibling_id':about.id, 'node_id':stories.id} @@ -851,7 +851,7 @@ class SelfReferentialPostUpdateTest(AssertMixin): class SelfReferentialPostUpdateTest2(AssertMixin): def setUpAll(self): global metadata, a_table - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) a_table = Table("a", metadata, Column("id", Integer(), primary_key=True), Column("fui", String(128)), @@ -890,4 +890,4 @@ class SelfReferentialPostUpdateTest2(AssertMixin): assert f2.foo is f1 if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/dynamic.py b/test/orm/dynamic.py index fe0bc92d7..f188a478f 100644 --- a/test/orm/dynamic.py +++ b/test/orm/dynamic.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import operator from sqlalchemy import * from sqlalchemy.orm import * @@ -10,7 +10,7 @@ from query import QueryTest class DynamicTest(FixtureTest): keep_mappers = False refresh_data = True - + def test_basic(self): mapper(User, users, properties={ 'addresses':dynamic_loader(mapper(Address, addresses)) @@ -31,22 +31,22 @@ class DynamicTest(FixtureTest): sess = create_session() u = sess.query(User).first() assert u.addresses.count() == 1, u.addresses.count() - + def test_backref(self): mapper(Address, addresses, properties={ 'user':relation(User, backref=backref('addresses', lazy='dynamic')) }) mapper(User, users) - + sess = create_session() ad = sess.query(Address).get(1) def go(): ad.user = None - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.flush() u = sess.query(User).get(7) assert ad not in u.addresses - + def test_no_count(self): mapper(User, users, properties={ 'addresses':dynamic_loader(mapper(Address, addresses)) @@ -58,24 +58,24 @@ class DynamicTest(FixtureTest): # result), else additional count() queries are issued when evaluating in a list context def go(): assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(User.id==7).all() - self.assert_sql_count(testbase.db, go, 2) - + self.assert_sql_count(testing.db, go, 2) + def test_m2m(self): mapper(Order, orders, properties={ 'items':relation(Item, secondary=order_items, lazy="dynamic", backref=backref('orders', lazy="dynamic")) }) mapper(Item, items) - + sess = create_session() o1 = Order(id=15, description="order 10") i1 = Item(id=10, description="item 8") o1.items.append(i1) sess.save(o1) sess.flush() - + assert o1 in i1.orders.all() assert i1 in o1.items.all() - + class FlushTest(FixtureTest): def test_basic(self): class Fixture(Base): @@ -124,15 +124,15 @@ class FlushTest(FixtureTest): sess.delete(u.addresses[4]) sess.delete(u.addresses[3]) assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses) - + sess.delete(u) - + # u.addresses relation will have to force the load # of all addresses so that they can be updated sess.flush() sess.close() - - assert testbase.db.scalar(addresses.count(addresses.c.user_id != None)) ==0 + + assert testing.db.scalar(addresses.count(addresses.c.user_id != None)) ==0 @testing.fails_on('maxdb') def test_remove_orphans(self): @@ -214,5 +214,4 @@ for autoflush in (False, True): create_backref_test(autoflush, saveuser) if __name__ == '__main__': - testbase.main() - + testenv.main() diff --git a/test/orm/eager_relations.py b/test/orm/eager_relations.py index 73d5b11dc..e42ef5cb8 100644 --- a/test/orm/eager_relations.py +++ b/test/orm/eager_relations.py @@ -1,6 +1,6 @@ """basic tests of eager loaded attributes""" -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -139,7 +139,7 @@ class EagerTest(FixtureTest): assert a.user_id==7 # assert that the eager loader added 'user_id' to the row # and deferred loading of that col was disabled - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) # do the mapping in reverse # (we would have just used an "addresses" backref but the test fixtures then require the whole @@ -161,7 +161,7 @@ class EagerTest(FixtureTest): assert u.addresses[0].user_id==7 # assert that the eager loader didn't have to affect 'user_id' here # and that its still deferred - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) clear_mappers() @@ -177,7 +177,7 @@ class EagerTest(FixtureTest): def go(): u = sess.query(User).limit(1).get(8) assert User(id=8, addresses=[Address(id=2, dingalings=[Dingaling(id=1)]), Address(id=3), Address(id=4)]) == u - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_many_to_many(self): @@ -189,11 +189,11 @@ class EagerTest(FixtureTest): q = create_session().query(Item) def go(): assert fixtures.item_keyword_result == q.all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def go(): assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_eager_option(self): @@ -207,7 +207,7 @@ class EagerTest(FixtureTest): def go(): assert fixtures.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_cyclical(self): """test that a circular eager relationship breaks the cycle with a lazy loader""" @@ -259,7 +259,7 @@ class EagerTest(FixtureTest): User(id=10) ] == q.all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_double_same_mappers(self): """tests eager loading with two relations simulatneously, from the same table, using aliases. """ @@ -299,7 +299,7 @@ class EagerTest(FixtureTest): User(id=10) ] == q.all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_no_false_hits(self): """test that eager loaders don't interpret main table columns as part of their eager load.""" @@ -362,7 +362,7 @@ class EagerTest(FixtureTest): def go(): l = q.filter(s.c.u2_id==User.c.id).distinct().all() assert fixtures.user_address_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) @testing.fails_on('maxdb') def test_limit_2(self): @@ -444,7 +444,7 @@ class EagerTest(FixtureTest): def go(): l = q.filter(users.c.id == 7).all() assert [User(id=7, address=Address(id=1))] == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) @testing.fails_on('maxdb') def test_many_to_one(self): @@ -459,7 +459,7 @@ class EagerTest(FixtureTest): assert a.user is not None u1 = sess.query(User).get(7) assert a.user is u1 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_one_and_many(self): @@ -480,7 +480,7 @@ class EagerTest(FixtureTest): def go(): assert fixtures.user_order_result[0:3] == l.all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_double_with_aggregate(self): @@ -510,7 +510,7 @@ class EagerTest(FixtureTest): ), User(id=10), ] == q.all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_wide(self): mapper(Order, orders, properties={'items':relation(Item, secondary=order_items, lazy=False, order_by=items.c.id)}) @@ -619,7 +619,7 @@ class AddEntityTest(FixtureTest): def go(): ret = sess.query(User).add_entity(Order).join('orders', aliased=True).order_by(User.id).order_by(Order.id).all() self.assertEquals(ret, self._assert_result()) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_options(self): mapper(User, users, properties={ @@ -637,13 +637,13 @@ class AddEntityTest(FixtureTest): def go(): ret = sess.query(User).options(eagerload('addresses')).add_entity(Order).join('orders', aliased=True).order_by(User.id).order_by(Order.id).all() self.assertEquals(ret, self._assert_result()) - self.assert_sql_count(testbase.db, go, 6) + self.assert_sql_count(testing.db, go, 6) sess.clear() def go(): ret = sess.query(User).options(eagerload('addresses')).add_entity(Order).options(eagerload('items', Order)).join('orders', aliased=True).order_by(User.id).order_by(Order.id).all() self.assertEquals(ret, self._assert_result()) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) class SelfReferentialEagerTest(ORMTest): def define_tables(self, metadata): @@ -684,7 +684,7 @@ class SelfReferentialEagerTest(ORMTest): ]), Node(data='n13') ]) == d - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_lazy_fallback_doesnt_affect_eager(self): @@ -726,7 +726,7 @@ class SelfReferentialEagerTest(ORMTest): Node(data='n122'), Node(data='n123') ] == list(n12.children) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_with_deferred(self): class Node(Base): @@ -747,19 +747,19 @@ class SelfReferentialEagerTest(ORMTest): def go(): assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).first() - self.assert_sql_count(testbase.db, go, 4) + self.assert_sql_count(testing.db, go, 4) sess.clear() def go(): assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).options(undefer('data')).first() - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) sess.clear() def go(): assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).options(undefer('data'), undefer('children.data')).first() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) @@ -793,7 +793,7 @@ class SelfReferentialEagerTest(ORMTest): ]), Node(data='n13') ]) == d - self.assert_sql_count(testbase.db, go, 2) + self.assert_sql_count(testing.db, go, 2) def go(): d = sess.query(Node).filter_by(data='n1').options(eagerload('children.children')).first() @@ -802,7 +802,7 @@ class SelfReferentialEagerTest(ORMTest): # testing only sqlite for now since the query text is slightly different on other # dialects if testing.against('sqlite'): - self.assert_sql(testbase.db, go, [ + self.assert_sql(testing.db, go, [ ( "SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id, nodes.data AS nodes_data FROM nodes WHERE nodes.data = :nodes_data_1 ORDER BY nodes.oid LIMIT 1 OFFSET 0", {'nodes_data_1': 'n1'} @@ -840,7 +840,7 @@ class SelfReferentialEagerTest(ORMTest): ]), Node(data='n13') ]) == d - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) class SelfReferentialM2MEagerTest(ORMTest): def define_tables(self, metadata): @@ -919,4 +919,4 @@ class CyclicalInheritingEagerTest(ORMTest): create_session().query(SubT).all() if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/entity.py b/test/orm/entity.py index 7011fa3e0..c56d22f9d 100644 --- a/test/orm/entity.py +++ b/test/orm/entity.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from sqlalchemy.ext.sessioncontext import SessionContext @@ -12,7 +12,7 @@ class EntityTest(AssertMixin): @testing.uses_deprecated('SessionContext') def setUpAll(self): global user1, user2, address1, address2, metadata, ctx - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) ctx = SessionContext(create_session) user1 = Table('user1', metadata, @@ -216,4 +216,4 @@ class EntityTest(AssertMixin): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/expire.py b/test/orm/expire.py index d54e9fc2d..98963c012 100644 --- a/test/orm/expire.py +++ b/test/orm/expire.py @@ -1,6 +1,6 @@ """test attribute/instance expiration, deferral of attributes, etc.""" -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -10,25 +10,25 @@ from testlib.fixtures import * class ExpireTest(FixtureTest): keep_mappers = False refresh_data = True - + def test_expire(self): mapper(User, users, properties={ 'addresses':relation(Address, backref='user'), }) mapper(Address, addresses) - + sess = create_session() u = sess.query(User).get(7) assert len(u.addresses) == 1 u.name = 'foo' del u.addresses[0] sess.expire(u) - + assert 'name' not in u.__dict__ - + def go(): assert u.name == 'jack' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert 'name' in u.__dict__ u.name = 'foo' @@ -48,46 +48,46 @@ class ExpireTest(FixtureTest): def go(): assert u.name == 'jack' - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_expire_doesntload_on_set(self): mapper(User, users) - + sess = create_session() u = sess.query(User).get(7) - + sess.expire(u, attribute_names=['name']) def go(): u.name = 'somenewname' - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) sess.flush() sess.clear() assert sess.query(User).get(7).name == 'somenewname' - + def test_no_session(self): mapper(User, users) sess = create_session() u = sess.query(User).get(7) - + sess.expire(u, attribute_names=['name']) sess.expunge(u) try: u.name except exceptions.InvalidRequestError, e: assert str(e) == "Instance <class 'testlib.fixtures.User'> is not bound to a Session, and no contextual session is established; attribute refresh operation cannot proceed" - + def test_expire_preserves_changes(self): """test that the expire load operation doesn't revert post-expire changes""" - + mapper(Order, orders) sess = create_session() o = sess.query(Order).get(3) sess.expire(o) - + o.description = "order 3 modified" def go(): assert o.isopen == 1 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert o.description == 'order 3 modified' del o.description @@ -98,14 +98,14 @@ class ExpireTest(FixtureTest): assert "description" not in o.__dict__ assert o.description is None - + o.isopen=15 sess.expire(o, ['isopen', 'description']) o.description = 'some new description' sess.query(Order).all() assert o.isopen == 1 assert o.description == 'some new description' - + if False: # NOTYET: need to implement unconditional population # of expired attriutes in mapper._instances() @@ -114,12 +114,12 @@ class ExpireTest(FixtureTest): del o.isopen def go(): assert o.isopen is None - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_expire_committed(self): """test that the committed state of the attribute receives the most recent DB data""" mapper(Order, orders) - + sess = create_session() o = sess.query(Order).get(3) sess.expire(o) @@ -129,8 +129,8 @@ class ExpireTest(FixtureTest): assert o._state.dict['description'] == 'order 3 modified' def go(): sess.flush() - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_expire_cascade(self): mapper(User, users, properties={ 'addresses':relation(Address, cascade="all, refresh-expire") @@ -162,8 +162,8 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address == 'jack@bean.com' assert u.name == 'jack' - # two loads - self.assert_sql_count(testbase.db, go, 2) + # two loads + self.assert_sql_count(testing.db, go, 2) assert 'name' in u.__dict__ assert 'addresses' in u.__dict__ @@ -183,9 +183,9 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address == 'jack@bean.com' assert u.name == 'jack' - # two loads, since relation() + scalar are + # two loads, since relation() + scalar are # separate right now - self.assert_sql_count(testbase.db, go, 2) + self.assert_sql_count(testing.db, go, 2) assert 'name' in u.__dict__ assert 'addresses' in u.__dict__ @@ -198,35 +198,35 @@ class ExpireTest(FixtureTest): sess = create_session() o = sess.query(Order).get(3) - + sess.expire(o, attribute_names=['description']) assert 'id' in o.__dict__ assert 'description' not in o.__dict__ assert o._state.dict['isopen'] == 1 - + orders.update(orders.c.id==3).execute(description='order 3 modified') - + def go(): assert o.description == 'order 3 modified' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert o._state.dict['description'] == 'order 3 modified' - + o.isopen = 5 sess.expire(o, attribute_names=['description']) assert 'id' in o.__dict__ assert 'description' not in o.__dict__ assert o.__dict__['isopen'] == 5 assert o._state.committed_state['isopen'] == 1 - + def go(): assert o.description == 'order 3 modified' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert o.__dict__['isopen'] == 5 assert o._state.dict['description'] == 'order 3 modified' assert o._state.committed_state['isopen'] == 1 sess.flush() - + sess.expire(o, attribute_names=['id', 'isopen', 'description']) assert 'id' not in o.__dict__ assert 'isopen' not in o.__dict__ @@ -235,7 +235,7 @@ class ExpireTest(FixtureTest): assert o.description == 'order 3 modified' assert o.id == 3 assert o.isopen == 5 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_partial_expire_lazy(self): mapper(User, users, properties={ @@ -245,41 +245,41 @@ class ExpireTest(FixtureTest): sess = create_session() u = sess.query(User).get(8) - + sess.expire(u, ['name', 'addresses']) assert 'name' not in u.__dict__ assert 'addresses' not in u.__dict__ - + # hit the lazy loader. just does the lazy load, # doesnt do the overall refresh def go(): assert u.addresses[0].email_address=='ed@wood.com' - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + assert 'name' not in u.__dict__ - - # check that mods to expired lazy-load attributes + + # check that mods to expired lazy-load attributes # only do the lazy load sess.expire(u, ['name', 'addresses']) def go(): u.addresses = [Address(id=10, email_address='foo@bar.com')] - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + sess.flush() - - # flush has occurred, and addresses was modified, + + # flush has occurred, and addresses was modified, # so the addresses collection got committed and is # longer expired def go(): assert u.addresses[0].email_address=='foo@bar.com' assert len(u.addresses) == 1 - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + # but the name attribute was never loaded and so # still loads def go(): assert u.name == 'ed' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_partial_expire_eager(self): mapper(User, users, properties={ @@ -296,14 +296,14 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address=='ed@wood.com' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) - # check that mods to expired eager-load attributes + # check that mods to expired eager-load attributes # do the refresh sess.expire(u, ['name', 'addresses']) def go(): u.addresses = [Address(id=10, email_address='foo@bar.com')] - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.flush() # this should ideally trigger the whole load @@ -311,34 +311,34 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address=='foo@bar.com' assert len(u.addresses) == 1 - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def go(): assert u.name == 'ed' # scalar attributes have their own load - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) # ideally, this was already loaded, but we arent # doing it that way right now - #self.assert_sql_count(testbase.db, go, 0) + #self.assert_sql_count(testing.db, go, 0) def test_partial_expire_deferred(self): mapper(Order, orders, properties={ 'description':deferred(orders.c.description) }) - + sess = create_session() o = sess.query(Order).get(3) sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ assert 'description' not in o.__dict__ - + # test that expired attribute access refreshes # the deferred def go(): assert o.isopen == 1 assert o.description == 'order 3' - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ assert 'description' not in o.__dict__ @@ -347,10 +347,10 @@ class ExpireTest(FixtureTest): def go(): assert o.description == 'order 3' assert o.isopen == 1 - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + clear_mappers() - + mapper(Order, orders) sess.clear() @@ -362,20 +362,20 @@ class ExpireTest(FixtureTest): # sanity check def go(): assert o.description == 'order 3' - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + assert 'description' in o.__dict__ assert 'isopen' in o.__dict__ sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ assert 'description' not in o.__dict__ - + # test that expired attribute access refreshes # the deferred def go(): assert o.isopen == 1 assert o.description == 'order 3' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ @@ -385,8 +385,8 @@ class ExpireTest(FixtureTest): def go(): assert o.description == 'order 3' assert o.isopen == 1 - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + class RefreshTest(FixtureTest): keep_mappers = False @@ -412,7 +412,7 @@ class RefreshTest(FixtureTest): # username is back to the DB assert u.name == 'jack' - + assert id(a) not in [id(x) for x in u.addresses] u.name = 'foo' @@ -438,28 +438,28 @@ class RefreshTest(FixtureTest): assert 'name' not in u.__dict__ s.refresh(u) assert u.name == 'jack' - + def test_refresh_with_lazy(self): - """test that when a lazy loader is set as a trigger on an object's attribute - (at the attribute level, not the class level), a refresh() operation doesnt + """test that when a lazy loader is set as a trigger on an object's attribute + (at the attribute level, not the class level), a refresh() operation doesnt fire the lazy loader or create any problems""" - + s = create_session() mapper(User, users, properties={'addresses':relation(mapper(Address, addresses))}) q = s.query(User).options(lazyload('addresses')) u = q.filter(users.c.id==8).first() def go(): s.refresh(u) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_refresh_with_eager(self): """test that a refresh/expire operation loads rows properly and sends correct "isnew" state to eager loaders""" - + mapper(User, users, properties={ 'addresses':relation(mapper(Address, addresses), lazy=False) }) - + s = create_session() u = s.get(User, 8) assert len(u.addresses) == 3 @@ -480,12 +480,12 @@ class RefreshTest(FixtureTest): mapper(Address, addresses) mapper(User, users, properties = dict(addresses=relation(Address,cascade="all, delete-orphan",lazy=False)) ) - + u=User() u.name='Justin' a = Address(id=10, email_address='lala') u.addresses.append(a) - + s.save(u) s.flush() s.clear() @@ -497,4 +497,4 @@ class RefreshTest(FixtureTest): s.refresh(u) if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/generative.py b/test/orm/generative.py index d72887a58..294117b0d 100644 --- a/test/orm/generative.py +++ b/test/orm/generative.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from sqlalchemy import exceptions @@ -15,7 +15,7 @@ class Foo(object): class GenerativeQueryTest(PersistTest): def setUpAll(self): global foo, metadata - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) foo = Table('foo', metadata, Column('id', Integer, Sequence('foo_id_seq'), primary_key=True), Column('bar', Integer), @@ -24,7 +24,7 @@ class GenerativeQueryTest(PersistTest): mapper(Foo, foo) metadata.create_all() - sess = create_session(bind=testbase.db) + sess = create_session(bind=testing.db) for i in range(100): sess.save(Foo(bar=i, range=i%10)) sess.flush() @@ -34,14 +34,14 @@ class GenerativeQueryTest(PersistTest): clear_mappers() def test_selectby(self): - res = create_session(bind=testbase.db).query(Foo).filter_by(range=5) + res = create_session(bind=testing.db).query(Foo).filter_by(range=5) assert res.order_by([Foo.c.bar])[0].bar == 5 assert res.order_by([desc(Foo.c.bar)])[0].bar == 95 @testing.unsupported('mssql') @testing.fails_on('maxdb') def test_slice(self): - sess = create_session(bind=testbase.db) + sess = create_session(bind=testing.db) query = sess.query(Foo) orig = query.all() assert query[1] == orig[1] @@ -54,7 +54,7 @@ class GenerativeQueryTest(PersistTest): assert query[10:20][5] == orig[10:20][5] def test_aggregate(self): - sess = create_session(bind=testbase.db) + sess = create_session(bind=testing.db) query = sess.query(Foo) assert query.count() == 100 assert query.filter(foo.c.bar<30).min(foo.c.bar) == 0 @@ -64,38 +64,38 @@ class GenerativeQueryTest(PersistTest): def test_aggregate_1(self): if (testing.against('mysql') and - testbase.db.dialect.dbapi.version_info[:4] == (1, 2, 1, 'gamma')): + testing.db.dialect.dbapi.version_info[:4] == (1, 2, 1, 'gamma')): return - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert query.filter(foo.c.bar<30).sum(foo.c.bar) == 435 @testing.fails_on('postgres', 'mysql', 'firebird', 'mssql') def test_aggregate_2(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert query.filter(foo.c.bar<30).avg(foo.c.bar) == 14.5 @testing.fails_on_everything_except('sqlite', 'postgres', 'mysql', 'firebird', 'mssql') def test_aggregate_2_int(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert int(query.filter(foo.c.bar<30).avg(foo.c.bar)) == 14 @testing.fails_on('postgres', 'mysql', 'firebird', 'mssql') def test_aggregate_3(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert query.filter(foo.c.bar<30).apply_avg(foo.c.bar).first() == 14.5 assert query.filter(foo.c.bar<30).apply_avg(foo.c.bar).one() == 14.5 def test_filter(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert query.count() == 100 assert query.filter(Foo.c.bar < 30).count() == 30 res2 = query.filter(Foo.c.bar < 30).filter(Foo.c.bar > 10) assert res2.count() == 19 def test_options(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) class ext1(MapperExtension): def populate_instance(self, mapper, selectcontext, row, instance, **flags): instance.TEST = "hello world" @@ -103,16 +103,16 @@ class GenerativeQueryTest(PersistTest): assert query.options(extension(ext1()))[0].TEST == "hello world" def test_order_by(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert query.order_by([Foo.c.bar])[0].bar == 0 assert query.order_by([desc(Foo.c.bar)])[0].bar == 99 def test_offset(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert list(query.order_by([Foo.c.bar]).offset(10))[0].bar == 10 def test_offset(self): - query = create_session(bind=testbase.db).query(Foo) + query = create_session(bind=testing.db).query(Foo) assert len(list(query.limit(10))) == 10 class Obj1(object): @@ -133,17 +133,17 @@ class GenerativeTest2(PersistTest): ) mapper(Obj1, table1) mapper(Obj2, table2) - metadata.create_all(bind=testbase.db) - testbase.db.execute(table1.insert(), {'id':1},{'id':2},{'id':3},{'id':4}) - testbase.db.execute(table2.insert(), {'num':1,'t1id':1},{'num':2,'t1id':1},{'num':3,'t1id':1},\ + metadata.create_all(bind=testing.db) + testing.db.execute(table1.insert(), {'id':1},{'id':2},{'id':3},{'id':4}) + testing.db.execute(table2.insert(), {'num':1,'t1id':1},{'num':2,'t1id':1},{'num':3,'t1id':1},\ {'num':4,'t1id':2},{'num':5,'t1id':2},{'num':6,'t1id':3}) def tearDownAll(self): - metadata.drop_all(bind=testbase.db) + metadata.drop_all(bind=testing.db) clear_mappers() def test_distinctcount(self): - query = create_session(bind=testbase.db).query(Obj1) + query = create_session(bind=testing.db).query(Obj1) assert query.count() == 4 res = query.filter(and_(table1.c.id==table2.c.t1id,table2.c.t1id==1)) assert res.count() == 3 @@ -165,7 +165,7 @@ class RelationsTest(AssertMixin): 'items':relation(mapper(tables.Item, tables.orderitems)) })) }) - session = create_session(bind=testbase.db) + session = create_session(bind=testing.db) query = session.query(tables.User) x = query.join(['orders', 'items']).filter(tables.Item.c.item_id==2) print x.compile() @@ -177,7 +177,7 @@ class RelationsTest(AssertMixin): 'items':relation(mapper(tables.Item, tables.orderitems)) })) }) - session = create_session(bind=testbase.db) + session = create_session(bind=testing.db) query = session.query(tables.User) x = query.outerjoin(['orders', 'items']).filter(or_(tables.Order.c.order_id==None,tables.Item.c.item_id==2)) print x.compile() @@ -189,7 +189,7 @@ class RelationsTest(AssertMixin): 'items':relation(mapper(tables.Item, tables.orderitems)) })) }) - session = create_session(bind=testbase.db) + session = create_session(bind=testing.db) query = session.query(tables.User) x = query.outerjoin(['orders', 'items']).filter(or_(tables.Order.c.order_id==None,tables.Item.c.item_id==2)).count() assert x==2 @@ -199,7 +199,7 @@ class RelationsTest(AssertMixin): 'items':relation(mapper(tables.Item, tables.orderitems)) })) }) - session = create_session(bind=testbase.db) + session = create_session(bind=testing.db) query = session.query(tables.User) x = query.select_from(tables.users.outerjoin(tables.orders).outerjoin(tables.orderitems)).\ filter(or_(tables.Order.c.order_id==None,tables.Item.c.item_id==2)) @@ -210,7 +210,7 @@ class RelationsTest(AssertMixin): class CaseSensitiveTest(PersistTest): def setUpAll(self): global metadata, table1, table2 - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) table1 = Table('Table1', metadata, Column('ID', Integer, primary_key=True), ) @@ -230,7 +230,7 @@ class CaseSensitiveTest(PersistTest): clear_mappers() def test_distinctcount(self): - q = create_session(bind=testbase.db).query(Obj1) + q = create_session(bind=testing.db).query(Obj1) assert q.count() == 4 res = q.filter(and_(table1.c.ID==table2.c.T1ID,table2.c.T1ID==1)) assert res.count() == 3 @@ -247,7 +247,7 @@ class SelfRefTest(ORMTest): def test_noautojoin(self): class T(object):pass mapper(T, t1, properties={'children':relation(T)}) - sess = create_session(bind=testbase.db) + sess = create_session(bind=testing.db) try: sess.query(T).join('children').select_by(id=7) assert False @@ -263,4 +263,4 @@ class SelfRefTest(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/abc_inheritance.py b/test/orm/inheritance/abc_inheritance.py index dbe74dc80..5f7a10756 100644 --- a/test/orm/inheritance/abc_inheritance.py +++ b/test/orm/inheritance/abc_inheritance.py @@ -1,9 +1,10 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from sqlalchemy.orm.sync import ONETOMANY, MANYTOONE from testlib import * + def produce_test(parent, child, direction): """produce a testcase for A->B->C inheritance with a self-referential relationship between two of the classes, using either one-to-many or @@ -165,4 +166,4 @@ for parent in ["a", "b", "c"]: if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/abc_polymorphic.py b/test/orm/inheritance/abc_polymorphic.py index da9097637..79cc91d2c 100644 --- a/test/orm/inheritance/abc_polymorphic.py +++ b/test/orm/inheritance/abc_polymorphic.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions, util from sqlalchemy.orm import * @@ -8,7 +8,7 @@ from testlib import fixtures class ABCTest(ORMTest): def define_tables(self, metadata): global a, b, c - a = Table('a', metadata, + a = Table('a', metadata, Column('id', Integer, primary_key=True), Column('adata', String(30)), Column('type', String(30)), @@ -16,26 +16,26 @@ class ABCTest(ORMTest): b = Table('b', metadata, Column('id', Integer, ForeignKey('a.id'), primary_key=True), Column('bdata', String(30))) - c = Table('c', metadata, + c = Table('c', metadata, Column('id', Integer, ForeignKey('b.id'), primary_key=True), Column('cdata', String(30))) - + def make_test(fetchtype): def test_roundtrip(self): class A(fixtures.Base):pass class B(A):pass class C(B):pass - + if fetchtype == 'union': abc = a.outerjoin(b).outerjoin(c) bc = a.join(b).outerjoin(c) else: abc = bc = None - + mapper(A, a, select_table=abc, polymorphic_on=a.c.type, polymorphic_identity='a', polymorphic_fetch=fetchtype) mapper(B, b, select_table=bc, inherits=A, polymorphic_identity='b', polymorphic_fetch=fetchtype) mapper(C, c, inherits=B, polymorphic_identity='c') - + a1 = A(adata='a1') b1 = B(bdata='b1', adata='b1') b2 = B(bdata='b2', adata='b2') @@ -43,13 +43,13 @@ class ABCTest(ORMTest): c1 = C(cdata='c1', bdata='c1', adata='c1') c2 = C(cdata='c2', bdata='c2', adata='c2') c3 = C(cdata='c2', bdata='c2', adata='c2') - + sess = create_session() for x in (a1, b1, b2, b3, c1, c2, c3): sess.save(x) sess.flush() sess.clear() - + #for obj in sess.query(A).all(): # print obj assert [ @@ -79,12 +79,11 @@ class ABCTest(ORMTest): test_roundtrip.__name__ = 'test_%s' % fetchtype return test_roundtrip - + test_union = make_test('union') test_select = make_test('select') test_deferred = make_test('deferred') - - + + if __name__ == '__main__': - testbase.main() -
\ No newline at end of file + testenv.main() diff --git a/test/orm/inheritance/alltests.py b/test/orm/inheritance/alltests.py index d95de6246..e51297f8a 100644 --- a/test/orm/inheritance/alltests.py +++ b/test/orm/inheritance/alltests.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import unittest def suite(): @@ -16,7 +16,7 @@ def suite(): 'orm.inheritance.productspec', 'orm.inheritance.magazine', 'orm.inheritance.selects', - + ) alltests = unittest.TestSuite() for name in modules_to_test: @@ -28,4 +28,4 @@ def suite(): if __name__ == '__main__': - testbase.main(suite()) + testenv.main(suite()) diff --git a/test/orm/inheritance/basic.py b/test/orm/inheritance/basic.py index f2b7c6e4f..39b9fb916 100644 --- a/test/orm/inheritance/basic.py +++ b/test/orm/inheritance/basic.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions, util from sqlalchemy.orm import * @@ -191,7 +191,7 @@ class GetTest(ORMTest): assert sess.query(Bar).get(bl.id) == bl assert sess.query(Blub).get(bl.id) == bl - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) else: # this is testing the 'wrong' behavior of using get() # polymorphically with mappers that are not configured to be @@ -213,7 +213,7 @@ class GetTest(ORMTest): assert sess.query(Blub).get(bl.id) == bl - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) test_get.__name__ = name return test_get @@ -647,4 +647,4 @@ class SyncCompileTest(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/concrete.py b/test/orm/inheritance/concrete.py index c68ff46c0..e1ab870c3 100644 --- a/test/orm/inheritance/concrete.py +++ b/test/orm/inheritance/concrete.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -113,9 +113,9 @@ class ConcreteTest(ORMTest): def go(): c2 = session.query(Company).get(c.id) assert set([repr(x) for x in c2.employees]) == set(["Engineer Kurt knows how to hack", "Manager Tom knows how to manage things"]) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/magazine.py b/test/orm/inheritance/magazine.py index 8c2771c81..b5c5096f5 100644 --- a/test/orm/inheritance/magazine.py +++ b/test/orm/inheritance/magazine.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -51,7 +51,7 @@ class LocationName(BaseObject): class PageSize(BaseObject): def __repr__(self): return "%s(%sx%s, %s)" % (self.__class__.__name__, self.width, self.height, self.name) - + class Magazine(BaseObject): def __repr__(self): return "%s(%s, %s)" % (self.__class__.__name__, repr(self.location), repr(self.size)) @@ -194,7 +194,7 @@ def generate_round_trip_test(use_unions=False, use_joins=False): page2 = MagazinePage(magazine=magazine,page_no=2) page3 = ClassifiedPage(magazine=magazine,page_no=3) session.save(pub) - + session.flush() print [x for x in session] session.clear() @@ -206,13 +206,13 @@ def generate_round_trip_test(use_unions=False, use_joins=False): print p.issues[0].locations[0].magazine.pages print [page, page2, page3] assert repr(p.issues[0].locations[0].magazine.pages) == repr([page, page2, page3]), repr(p.issues[0].locations[0].magazine.pages) - + test_roundtrip.__name__ = "test_%s" % (not use_union and (use_joins and "joins" or "select") or "unions") setattr(MagazineTest, test_roundtrip.__name__, test_roundtrip) - + for (use_union, use_join) in [(True, False), (False, True), (False, False)]: generate_round_trip_test(use_union, use_join) - + if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/manytomany.py b/test/orm/inheritance/manytomany.py index d28ce8ada..f1bbc2ae7 100644 --- a/test/orm/inheritance/manytomany.py +++ b/test/orm/inheritance/manytomany.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -243,4 +243,4 @@ class InheritTest3(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/poly_linked_list.py b/test/orm/inheritance/poly_linked_list.py index 455200136..b2dd6c658 100644 --- a/test/orm/inheritance/poly_linked_list.py +++ b/test/orm/inheritance/poly_linked_list.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -195,4 +195,4 @@ class PolymorphicCircularTest(ORMTest): assert original == forwards == backwards if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/polymorph.py b/test/orm/inheritance/polymorph.py index cf91dbe77..d2d83ea3c 100644 --- a/test/orm/inheritance/polymorph.py +++ b/test/orm/inheritance/polymorph.py @@ -1,6 +1,6 @@ """tests basic polymorphic mapper loading/saving, minimal relations""" -import testbase +import testenv; testenv.configure_for_tests() import sets from sqlalchemy import * from sqlalchemy.orm import * @@ -294,15 +294,15 @@ def generate_round_trip_test(include_base=False, lazy_relation=True, redefine_co if not lazy_relation: if polymorphic_fetch=='union': - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) else: - self.assert_sql_count(testbase.db, go, 5) + self.assert_sql_count(testing.db, go, 5) else: if polymorphic_fetch=='union': - self.assert_sql_count(testbase.db, go, 2) + self.assert_sql_count(testing.db, go, 2) else: - self.assert_sql_count(testbase.db, go, 6) + self.assert_sql_count(testing.db, go, 6) # test selecting from the query, using the base mapped table (people) as the selection criterion. # in the case of the polymorphic Person query, the "people" selectable should be adapted to be "person_join" @@ -364,4 +364,4 @@ for include_base in [True, False]: generate_round_trip_test(include_base, lazy_relation, redefine_colprop, use_literal_join, polymorphic_fetch, False) if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/polymorph2.py b/test/orm/inheritance/polymorph2.py index f87ea9357..96a8ddbf6 100644 --- a/test/orm/inheritance/polymorph2.py +++ b/test/orm/inheritance/polymorph2.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions, util from sqlalchemy.orm import * @@ -401,7 +401,7 @@ class RelationTest4(ORMTest): def go(): testcar = session.query(Car).options(eagerload('employee')).get(car1.car_id) assert str(testcar.employee) == "Engineer E4, status X" - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) session.clear() s = session.query(Car) @@ -620,7 +620,7 @@ class GenerativeTest(AssertMixin): # +--------------------------------------- has a ------+ global metadata, status, people, engineers, managers, cars - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) # table definitions status = Table('status', metadata, Column('status_id', Integer, primary_key=True), @@ -1010,4 +1010,4 @@ class InheritingEagerTest(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/productspec.py b/test/orm/inheritance/productspec.py index b2e375cdb..54810c31f 100644 --- a/test/orm/inheritance/productspec.py +++ b/test/orm/inheritance/productspec.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from datetime import datetime from sqlalchemy import * from sqlalchemy.orm import * @@ -314,4 +314,4 @@ class InheritTest(ORMTest): assert orig == new == '<Assembly a1> specification=[<SpecLine 1.0 <Detail d1>>] documents=[<Document doc1>, <RasterDocument doc2>]' if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/query.py b/test/orm/inheritance/query.py index ee837baed..bace4e6cf 100644 --- a/test/orm/inheritance/query.py +++ b/test/orm/inheritance/query.py @@ -3,7 +3,7 @@ and inheriting mappers.""" # TODO: under construction ! -import testbase +import testenv; testenv.configure_for_tests() import sets from sqlalchemy import * from sqlalchemy.orm import * @@ -12,7 +12,7 @@ from testlib import fixtures class Company(fixtures.Base): pass - + class Person(fixtures.Base): pass class Engineer(Person): @@ -28,44 +28,44 @@ class Paperwork(fixtures.Base): class PolymorphicQueryTest(ORMTest): keep_data = True keep_mappers = True - + def define_tables(self, metadata): global companies, people, engineers, managers, boss, paperwork - - companies = Table('companies', metadata, + + companies = Table('companies', metadata, Column('company_id', Integer, Sequence('company_id_seq', optional=True), primary_key=True), Column('name', String(50))) - people = Table('people', metadata, + people = Table('people', metadata, Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True), Column('company_id', Integer, ForeignKey('companies.company_id')), Column('name', String(50)), Column('type', String(30))) - engineers = Table('engineers', metadata, + engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30)), Column('engineer_name', String(50)), Column('primary_language', String(50)), ) - managers = Table('managers', metadata, + managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30)), Column('manager_name', String(50)) ) - boss = Table('boss', metadata, + boss = Table('boss', metadata, Column('boss_id', Integer, ForeignKey('managers.person_id'), primary_key=True), Column('golf_swing', String(30)), ) - paperwork = Table('paperwork', metadata, + paperwork = Table('paperwork', metadata, Column('paperwork_id', Integer, primary_key=True), - Column('description', String(50)), + Column('description', String(50)), Column('person_id', Integer, ForeignKey('people.person_id'))) - - # create the most awkward polymorphic selects possible; + + # create the most awkward polymorphic selects possible; # the union does not include the "people" table by itself nor does it have # "people.person_id" directly in it, and it also does not include at all # the "boss" table @@ -74,14 +74,14 @@ class PolymorphicQueryTest(ORMTest): 'engineer':people.join(engineers), 'manager':people.join(managers), }, None, 'pjoin') - - # separate join for second-level inherit + + # separate join for second-level inherit manager_join = people.join(managers).outerjoin(boss) mapper(Company, companies, properties={ 'employees':relation(Person) }) - mapper(Person, people, select_table=person_join, polymorphic_on=people.c.type, polymorphic_identity='person', order_by=person_join.c.person_id, + mapper(Person, people, select_table=person_join, polymorphic_on=people.c.type, polymorphic_identity='person', order_by=person_join.c.person_id, properties={ 'paperwork':relation(Paperwork) }) @@ -89,7 +89,7 @@ class PolymorphicQueryTest(ORMTest): mapper(Manager, managers, select_table=manager_join, inherits=Person, polymorphic_identity='manager') mapper(Boss, boss, inherits=Manager, polymorphic_identity='boss') mapper(Paperwork, paperwork) - + def insert_data(self): c1 = Company(name="MegaCorp, Inc.") c2 = Company(name="Elbonia, Inc.") @@ -109,7 +109,7 @@ class PolymorphicQueryTest(ORMTest): Paperwork(description="review #3") ]) c1.employees = [e1, e2, b1, m1] - + e3 = Engineer(name="vlad", engineer_name="vlad", primary_language="cobol", status="elbonian engineer") c2.employees = [e3] sess = create_session() @@ -117,19 +117,17 @@ class PolymorphicQueryTest(ORMTest): sess.save(c2) sess.flush() sess.clear() - + global all_employees, c1_employees, c2_employees all_employees = [e1, e2, b1, m1, e3] c1_employees = [e1, e2, b1, m1] c2_employees = [e3] - + def test_load_all(self): sess = create_session() - + self.assertEquals(sess.query(Person).all(), all_employees) -if __name__ == "__main__": - testbase.main() - - -
\ No newline at end of file + +if __name__ == "__main__": + testenv.main() diff --git a/test/orm/inheritance/selects.py b/test/orm/inheritance/selects.py index a38e54874..b3a343e38 100644 --- a/test/orm/inheritance/selects.py +++ b/test/orm/inheritance/selects.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -17,10 +17,10 @@ class InheritingSelectablesTest(ORMTest): def test_load(self): # TODO: add persistence test also - testbase.db.execute(foo.insert(), a='not bar', b='baz') - testbase.db.execute(foo.insert(), a='also not bar', b='baz') - testbase.db.execute(foo.insert(), a='i am bar', b='bar') - testbase.db.execute(foo.insert(), a='also bar', b='bar') + testing.db.execute(foo.insert(), a='not bar', b='baz') + testing.db.execute(foo.insert(), a='also not bar', b='baz') + testing.db.execute(foo.insert(), a='i am bar', b='bar') + testing.db.execute(foo.insert(), a='also bar', b='bar') class Foo(Base): pass class Bar(Foo): pass @@ -28,7 +28,7 @@ class InheritingSelectablesTest(ORMTest): mapper(Foo, foo, polymorphic_on=foo.c.b) - mapper(Baz, baz, + mapper(Baz, baz, select_table=foo.join(baz, foo.c.b=='baz').alias('baz'), inherits=Foo, inherit_condition=(foo.c.a==baz.c.a), @@ -37,15 +37,15 @@ class InheritingSelectablesTest(ORMTest): mapper(Bar, bar, select_table=foo.join(bar, foo.c.b=='bar').alias('bar'), - inherits=Foo, + inherits=Foo, inherit_condition=(foo.c.a==bar.c.a), inherit_foreign_keys=[bar.c.a], polymorphic_identity='bar') - s = sessionmaker(bind=testbase.db)() + s = sessionmaker(bind=testing.db)() assert [Baz(), Baz(), Bar(), Bar()] == s.query(Foo).all() assert [Bar(), Bar()] == s.query(Bar).all() if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/inheritance/single.py b/test/orm/inheritance/single.py index c359230e2..22ca25bf0 100644 --- a/test/orm/inheritance/single.py +++ b/test/orm/inheritance/single.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -6,7 +6,7 @@ from testlib import * class SingleInheritanceTest(AssertMixin): def setUpAll(self): - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) global employees_table employees_table = Table('employees', metadata, Column('employee_id', Integer, primary_key=True), @@ -63,4 +63,4 @@ class SingleInheritanceTest(AssertMixin): assert session.query(JuniorEngineer).all() == [e2] if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/lazy_relations.py b/test/orm/lazy_relations.py index fdb1468aa..4bd3e71e1 100644 --- a/test/orm/lazy_relations.py +++ b/test/orm/lazy_relations.py @@ -1,6 +1,6 @@ """basic tests of lazy loaded attributes""" -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -257,7 +257,7 @@ class LazyTest(FixtureTest): def go(): # lazy load of a1.user should get it from the session assert a1.user is u1 - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) clear_mappers() def test_many_to_one(self): @@ -285,28 +285,28 @@ class LazyTest(FixtureTest): def go(): ad.user = None assert ad.user is None - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) u1 = sess.query(User).filter_by(id=7).one() def go(): assert ad not in u1.addresses - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.expire(u1, ['addresses']) def go(): assert ad in u1.addresses - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.expire(u1, ['addresses']) ad2 = Address() def go(): ad2.user = u1 assert ad2.user is u1 - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def go(): assert ad2 in u1.addresses - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) class M2OGetTest(FixtureTest): keep_mappers = False @@ -333,7 +333,7 @@ class M2OGetTest(FixtureTest): assert ad2.user.name == 'jack' # no lazy load assert ad3.user is None - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/lazytest1.py b/test/orm/lazytest1.py index 175639617..40f425abe 100644 --- a/test/orm/lazytest1.py +++ b/test/orm/lazytest1.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -6,7 +6,7 @@ from testlib import * class LazyTest(AssertMixin): def setUpAll(self): global info_table, data_table, rel_table, metadata - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) info_table = Table('infos', metadata, Column('pk', Integer, primary_key=True), Column('info', String(128))) @@ -88,4 +88,4 @@ class LazyTest(AssertMixin): assert len(info.rels[0].datas) == 3 if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/manytomany.py b/test/orm/manytomany.py index 32e3a5c37..ca6410533 100644 --- a/test/orm/manytomany.py +++ b/test/orm/manytomany.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from testlib import * @@ -314,4 +314,4 @@ class M2MTest3(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/mapper.py b/test/orm/mapper.py index 464979f4c..70cd81428 100644 --- a/test/orm/mapper.py +++ b/test/orm/mapper.py @@ -1,6 +1,6 @@ """tests general mapper operations with an emphasis on selecting/loading""" -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions, sql from sqlalchemy.orm import * @@ -225,7 +225,7 @@ class MapperTest(MapperSuperTest): self.assert_result([u], User, user_address_result[0]) assert u.user_name == 'jack' assert assert_col == [('get', 'jack')], str(assert_col) - self.assert_sql_count(testbase.db, go, 2) + self.assert_sql_count(testing.db, go, 2) u.name = 'ed' u3 = User() @@ -404,16 +404,16 @@ class MapperTest(MapperSuperTest): m.compile() assert account_ids_table in m._pks_by_table assert account_stuff_table not in m._pks_by_table - metadata.create_all(testbase.db) + metadata.create_all(testing.db) try: - sess = create_session(bind=testbase.db) + sess = create_session(bind=testing.db) a = A() sess.save(a) sess.flush() - assert testbase.db.execute(account_ids_table.count()).scalar() == 1 - assert testbase.db.execute(account_stuff_table.count()).scalar() == 0 + assert testing.db.execute(account_ids_table.count()).scalar() == 1 + assert testing.db.execute(account_stuff_table.count()).scalar() == 0 finally: - metadata.drop_all(testbase.db) + metadata.drop_all(testing.db) def test_mappingtoouterjoin(self): """test mapping to an outer join, with a composite primary key that allows nulls""" @@ -644,7 +644,7 @@ class OptionsTest(MapperSuperTest): def go(): u = sess.query(User).options(eagerload('adlist')).filter_by(user_name='jack').one() self.assert_result(u.adlist, Address, *(user_address_result[0]['addresses'][1])) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) @testing.uses_deprecated('//select_by') def test_extension_options(self): @@ -684,7 +684,7 @@ class OptionsTest(MapperSuperTest): def go(): self.assert_result(l, User, *user_address_result) - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) @testing.fails_on('maxdb') def test_eageroptionswithlimit(self): @@ -697,7 +697,7 @@ class OptionsTest(MapperSuperTest): def go(): assert u.user_id == 8 assert len(u.addresses) == 3 - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) sess.clear() @@ -706,7 +706,7 @@ class OptionsTest(MapperSuperTest): u = sess.query(User).filter_by(user_id=8).one() assert u.user_id == 8 assert len(u.addresses) == 3 - assert "tbl_row_count" not in self.capture_sql(testbase.db, go) + assert "tbl_row_count" not in self.capture_sql(testing.db, go) @testing.fails_on('maxdb') def test_lazyoptionswithlimit(self): @@ -719,7 +719,7 @@ class OptionsTest(MapperSuperTest): def go(): assert u.user_id == 8 assert len(u.addresses) == 3 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_eagerdegrade(self): """tests that an eager relation automatically degrades to a lazy relation if eager columns are not available""" @@ -732,7 +732,7 @@ class OptionsTest(MapperSuperTest): def go(): l = sess.query(usermapper).all() self.assert_result(l, User, *user_address_result) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() @@ -743,7 +743,7 @@ class OptionsTest(MapperSuperTest): r = users.select().execute() l = sess.query(usermapper).instances(r) self.assert_result(l, User, *user_address_result) - self.assert_sql_count(testbase.db, go, 4) + self.assert_sql_count(testing.db, go, 4) clear_mappers() @@ -770,7 +770,7 @@ class OptionsTest(MapperSuperTest): def go(): l = sess.query(usermapper).all() self.assert_result(l, User, *user_all_result) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() @@ -780,7 +780,7 @@ class OptionsTest(MapperSuperTest): r = users.select().execute() l = sess.query(usermapper).instances(r) self.assert_result(l, User, *user_all_result) - self.assert_sql_count(testbase.db, go, 7) + self.assert_sql_count(testing.db, go, 7) def test_lazyoptions(self): @@ -792,7 +792,7 @@ class OptionsTest(MapperSuperTest): l = sess.query(User).options(lazyload('addresses')).all() def go(): self.assert_result(l, User, *user_address_result) - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) def test_deepoptions(self): mapper(User, users, @@ -810,7 +810,7 @@ class OptionsTest(MapperSuperTest): u = sess.query(User).all() def go(): print u[0].orders[1].items[0].keywords[1] - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) sess.clear() @@ -821,7 +821,7 @@ class OptionsTest(MapperSuperTest): def go(): print u[0].orders[1].items[0].keywords[1] print "-------MARK2----------" - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) sess.clear() @@ -830,7 +830,7 @@ class OptionsTest(MapperSuperTest): u = q2.all() def go(): print u[0].orders[1].items[0].keywords[1] - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) sess.clear() @@ -845,7 +845,7 @@ class OptionsTest(MapperSuperTest): print "-------MARK5----------" q3 = sess.query(User).options(eagerload('orders.items.keywords')) u = q3.all() - self.assert_sql_count(testbase.db, go, 2) + self.assert_sql_count(testing.db, go, 2) class DeferredTest(MapperSuperTest): @@ -866,8 +866,8 @@ class DeferredTest(MapperSuperTest): o2 = l[2] print o2.description - orderby = str(orders.default_order_by()[0].compile(bind=testbase.db)) - self.assert_sql(testbase.db, go, [ + orderby = str(orders.default_order_by()[0].compile(bind=testing.db)) + self.assert_sql(testing.db, go, [ ("SELECT orders.order_id AS orders_order_id, orders.user_id AS orders_user_id, orders.isopen AS orders_isopen FROM orders ORDER BY %s" % orderby, {}), ("SELECT orders.description AS orders_description FROM orders WHERE orders.order_id = :param_1", {'param_1':3}) ]) @@ -884,7 +884,7 @@ class DeferredTest(MapperSuperTest): o.order_id = 7 def go(): o.description = "some description" - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_unsavedgroup(self): """test that deferred loading doesnt kick in when just PK cols are set""" @@ -899,7 +899,7 @@ class DeferredTest(MapperSuperTest): o.order_id = 7 def go(): o.description = "some description" - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_save(self): m = mapper(Order, orders, properties={ @@ -929,8 +929,8 @@ class DeferredTest(MapperSuperTest): assert o2.opened == 1 assert o2.userident == 7 assert o2.description == 'order 3' - orderby = str(orders.default_order_by()[0].compile(testbase.db)) - self.assert_sql(testbase.db, go, [ + orderby = str(orders.default_order_by()[0].compile(testing.db)) + self.assert_sql(testing.db, go, [ ("SELECT orders.order_id AS orders_order_id FROM orders ORDER BY %s" % orderby, {}), ("SELECT orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders WHERE orders.order_id = :param_1", {'param_1':3}) ]) @@ -942,7 +942,7 @@ class DeferredTest(MapperSuperTest): o2.description = 'order 3' def go(): sess.flush() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_preserve_changes(self): """test that the deferred load operation doesn't revert modifications on attributes""" @@ -959,7 +959,7 @@ class DeferredTest(MapperSuperTest): assert o.description == 'somenewdescription' def go(): assert o.opened == 1 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert o.description == 'somenewdescription' assert o in sess.dirty @@ -984,7 +984,7 @@ class DeferredTest(MapperSuperTest): def go(): # therefore the flush() shouldnt actually issue any SQL sess.flush() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_options(self): """tests using options on a mapper to create deferred and undeferred columns""" @@ -996,8 +996,8 @@ class DeferredTest(MapperSuperTest): l = q2.all() print l[2].user_id - orderby = str(orders.default_order_by()[0].compile(testbase.db)) - self.assert_sql(testbase.db, go, [ + orderby = str(orders.default_order_by()[0].compile(testing.db)) + self.assert_sql(testing.db, go, [ ("SELECT orders.order_id AS orders_order_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders ORDER BY %s" % orderby, {}), ("SELECT orders.user_id AS orders_user_id FROM orders WHERE orders.order_id = :param_1", {'param_1':3}) ]) @@ -1006,7 +1006,7 @@ class DeferredTest(MapperSuperTest): def go(): l = q3.all() print l[3].user_id - self.assert_sql(testbase.db, go, [ + self.assert_sql(testing.db, go, [ ("SELECT orders.order_id AS orders_order_id, orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders ORDER BY %s" % orderby, {}), ]) @@ -1026,8 +1026,8 @@ class DeferredTest(MapperSuperTest): assert o2.opened == 1 assert o2.userident == 7 assert o2.description == 'order 3' - orderby = str(orders.default_order_by()[0].compile(testbase.db)) - self.assert_sql(testbase.db, go, [ + orderby = str(orders.default_order_by()[0].compile(testing.db)) + self.assert_sql(testing.db, go, [ ("SELECT orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen, orders.order_id AS orders_order_id FROM orders ORDER BY %s" % orderby, {}), ]) @@ -1041,13 +1041,13 @@ class DeferredTest(MapperSuperTest): o1 = sess.query(Order).first() def go(): assert o1.description == 'order 1' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess = create_session() o1 = sess.query(Order).add_column(orders.c.description).first()[0] def go(): assert o1.description == 'order 1' - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_deepoptions(self): m = mapper(User, users, properties={ @@ -1063,7 +1063,7 @@ class DeferredTest(MapperSuperTest): item = l[0].orders[1].items[1] def go(): print item.item_name - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) self.assert_(item.item_name == 'item 4') sess.clear() q2 = q.options(undefer('orders.items.item_name')) @@ -1071,7 +1071,7 @@ class DeferredTest(MapperSuperTest): item = l[0].orders[1].items[1] def go(): print item.item_name - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) self.assert_(item.item_name == 'item 4') class CompositeTypesTest(ORMTest): @@ -1155,7 +1155,7 @@ class CompositeTypesTest(ORMTest): for e1, e2 in zip(g.edges, g2.edges): assert e1.start == e2.start assert e1.end == e2.end - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) # test comparison of CompositeProperties to their object instances g = sess.query(Graph).get([1, 1]) @@ -1216,7 +1216,7 @@ class NoLoadTest(MapperSuperTest): x = q.filter(users.c.user_id == 7).all() x[0].addresses l[0] = x - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) self.assert_result(l[0], User, {'user_id' : 7, 'addresses' : (Address, [])}, @@ -1232,7 +1232,7 @@ class NoLoadTest(MapperSuperTest): x = q.filter(users.c.user_id == 7).all() x[0].addresses l[0] = x - self.assert_sql_count(testbase.db, go, 2) + self.assert_sql_count(testing.db, go, 2) self.assert_result(l[0], User, {'user_id' : 7, 'addresses' : (Address, [{'address_id' : 1}])}, @@ -1393,7 +1393,7 @@ class RequirementsTest(AssertMixin): def setUpAll(self): global metadata, t1, t2, t3, t4, t5, t6 - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) t1 = Table('ht1', metadata, Column('id', Integer, primary_key=True), Column('value', String(10))) @@ -1559,4 +1559,4 @@ class RequirementsTest(AssertMixin): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/memusage.py b/test/orm/memusage.py index f0fba3123..93093365b 100644 --- a/test/orm/memusage.py +++ b/test/orm/memusage.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import gc from sqlalchemy import MetaData, Integer, String, ForeignKey from sqlalchemy.orm import mapper, relation, clear_mappers, create_session @@ -29,21 +29,21 @@ def profile_memory(func): return profile class MemUsageTest(AssertMixin): - + def test_session(self): - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) - table1 = Table("mytable", metadata, + table1 = Table("mytable", metadata, Column('col1', Integer, primary_key=True), Column('col2', String(30)) ) - table2 = Table("mytable2", metadata, + table2 = Table("mytable2", metadata, Column('col1', Integer, primary_key=True), Column('col2', String(30)), Column('col3', Integer, ForeignKey("mytable.col1")) ) - + metadata.create_all() m1 = mapper(A, table1, properties={ @@ -52,7 +52,7 @@ class MemUsageTest(AssertMixin): m2 = mapper(B, table2) m3 = mapper(A, table1, non_primary=True) - + @profile_memory def go(): sess = create_session() @@ -70,29 +70,29 @@ class MemUsageTest(AssertMixin): alist = sess.query(A).all() self.assertEquals( [ - A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]), - A(col2="a2", bs=[]), + A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]), + A(col2="a2", bs=[]), A(col2="a3", bs=[B(col2="b3")]) - ], + ], alist) - + for a in alist: sess.delete(a) sess.flush() go() - + metadata.drop_all() clear_mappers() - + def test_mapper_reset(self): - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) - table1 = Table("mytable", metadata, + table1 = Table("mytable", metadata, Column('col1', Integer, primary_key=True), Column('col2', String(30)) ) - table2 = Table("mytable2", metadata, + table2 = Table("mytable2", metadata, Column('col1', Integer, primary_key=True), Column('col2', String(30)), Column('col3', Integer, ForeignKey("mytable.col1")) @@ -106,7 +106,7 @@ class MemUsageTest(AssertMixin): m2 = mapper(B, table2) m3 = mapper(A, table1, non_primary=True) - + sess = create_session() a1 = A(col2="a1") a2 = A(col2="a2") @@ -122,17 +122,17 @@ class MemUsageTest(AssertMixin): alist = sess.query(A).all() self.assertEquals( [ - A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]), - A(col2="a2", bs=[]), + A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]), + A(col2="a2", bs=[]), A(col2="a3", bs=[B(col2="b3")]) - ], + ], alist) - + for a in alist: sess.delete(a) sess.flush() clear_mappers() - + metadata.create_all() try: go() @@ -140,14 +140,14 @@ class MemUsageTest(AssertMixin): metadata.drop_all() def test_with_inheritance(self): - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) - table1 = Table("mytable", metadata, + table1 = Table("mytable", metadata, Column('col1', Integer, primary_key=True), Column('col2', String(30)) ) - table2 = Table("mytable2", metadata, + table2 = Table("mytable2", metadata, Column('col1', Integer, ForeignKey('mytable.col1'), primary_key=True), Column('col3', String(30)), ) @@ -158,10 +158,10 @@ class MemUsageTest(AssertMixin): pass class B(A): pass - + mapper(A, table1, polymorphic_on=table1.c.col2, polymorphic_identity='a') mapper(B, table2, inherits=A, polymorphic_identity='b') - + sess = create_session() a1 = A() a2 = A() @@ -176,13 +176,13 @@ class MemUsageTest(AssertMixin): self.assertEquals( [ A(), A(), B(col3='b1'), B(col3='b2') - ], + ], alist) for a in alist: sess.delete(a) sess.flush() - + # dont need to clear_mappers() del B del A @@ -194,18 +194,18 @@ class MemUsageTest(AssertMixin): metadata.drop_all() def test_with_manytomany(self): - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) - table1 = Table("mytable", metadata, + table1 = Table("mytable", metadata, Column('col1', Integer, primary_key=True), Column('col2', String(30)) ) - table2 = Table("mytable2", metadata, + table2 = Table("mytable2", metadata, Column('col1', Integer, primary_key=True), Column('col2', String(30)), ) - + table3 = Table('t1tot2', metadata, Column('t1', Integer, ForeignKey('mytable.col1')), Column('t2', Integer, ForeignKey('mytable2.col1')), @@ -239,7 +239,7 @@ class MemUsageTest(AssertMixin): self.assertEquals( [ A(bs=[B(col2='b1')]), A(bs=[B(col2='b2')]) - ], + ], alist) for a in alist: @@ -256,6 +256,6 @@ class MemUsageTest(AssertMixin): finally: metadata.drop_all() - + if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/merge.py b/test/orm/merge.py index 8eeafb8e6..9fa93ffb2 100644 --- a/test/orm/merge.py +++ b/test/orm/merge.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -18,12 +18,12 @@ class MergeTest(AssertMixin): tables.delete() def setUp(self): pass - + def test_unsaved(self): """test merge of a single transient entity.""" mapper(User, users) sess = create_session() - + u = User() u.user_id = 7 u.user_name = "fred" @@ -51,7 +51,7 @@ class MergeTest(AssertMixin): a2.email_address = 'hoho@la.com' u.addresses.append(a1) u.addresses.append(a2) - + u2 = sess.merge(u) self.assert_result([u], User, {'user_id':7, 'user_name':'fred', 'addresses':(Address, [{'email_address':'foo@bar.com'}, {'email_address':'hoho@la.com'}])}) self.assert_result([u2], User, {'user_id':7, 'user_name':'fred', 'addresses':(Address, [{'email_address':'foo@bar.com'}, {'email_address':'hoho@la.com'}])}) @@ -66,7 +66,7 @@ class MergeTest(AssertMixin): 'addresses':relation(mapper(Address, addresses), backref='user') }) sess = create_session() - + # set up data and save u = User() u.user_id = 7 @@ -84,20 +84,20 @@ class MergeTest(AssertMixin): sess2 = create_session() u2 = sess2.query(User).get(7) self.assert_result([u2], User, {'user_id':7, 'user_name':'fred', 'addresses':(Address, [{'email_address':'foo@bar.com'}, {'email_address':'hoho@la.com'}])}) - + # make local changes to data u.user_name = 'fred2' u.addresses[1].email_address = 'hoho@lalala.com' - + # new session, merge modified data into session sess3 = create_session() u3 = sess3.merge(u) # insure local changes are pending self.assert_result([u3], User, {'user_id':7, 'user_name':'fred2', 'addresses':(Address, [{'email_address':'foo@bar.com'}, {'email_address':'hoho@lalala.com'}])}) - + # save merged data sess3.flush() - + # assert modified/merged data was saved sess.clear() u = sess.query(User).get(7) @@ -112,8 +112,8 @@ class MergeTest(AssertMixin): def go(): sess4.flush() # no changes; therefore flush should do nothing - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + # test with "dontload" merge sess5 = create_session() print "------------------" @@ -124,9 +124,9 @@ class MergeTest(AssertMixin): def go(): sess5.flush() # no changes; therefore flush should do nothing - # but also, dont_load wipes out any difference in committed state, + # but also, dont_load wipes out any difference in committed state, # so no flush at all - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) sess4 = create_session() u = sess4.merge(u, dont_load=True) @@ -135,8 +135,8 @@ class MergeTest(AssertMixin): def go(): sess4.flush() # afafds change flushes - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + sess5 = create_session() u2 = sess5.query(User).get(u.user_id) assert u2.user_name == 'fred2' @@ -147,12 +147,12 @@ class MergeTest(AssertMixin): mapper(Order, orders, properties={ 'items':relation(mapper(Item, orderitems)) }) - + mapper(User, users, properties={ 'addresses':relation(mapper(Address, addresses)), 'orders':relation(Order, backref='customer') }) - + sess = create_session() u = User() u.user_name='fred' @@ -165,10 +165,10 @@ class MergeTest(AssertMixin): o.items.append(i1) o.items.append(i2) u.orders.append(o) - + sess.save(u) sess.flush() - + sess2 = create_session() u2 = sess2.query(User).get(u.user_id) u.orders[0].items[1].item_name = 'item 2 modified' @@ -180,7 +180,7 @@ class MergeTest(AssertMixin): o.customer.user_name = 'also fred' sess2.merge(o) assert o2.customer.user_name == 'also fred' - + def test_saved_cascade_3(self): """test merge of a persistent entity with one_to_one relationship""" @@ -208,12 +208,12 @@ class MergeTest(AssertMixin): def test_noload_with_eager(self): """this test illustrates that with noload=True, we can't just copy the committed_state of the merged instance over; since it references collection objects - which themselves are to be merged. This committed_state would instead need to be piecemeal - 'converted' to represent the correct objects. + which themselves are to be merged. This committed_state would instead need to be piecemeal + 'converted' to represent the correct objects. However, at the moment I'd rather not support this use case; if you are merging with dont_load=True, you're typically dealing with caching and the merged objects shouldnt be "dirty". """ - + mapper(User, users, properties={ 'addresses':relation(mapper(Address, addresses),uselist = True) }) @@ -230,17 +230,17 @@ class MergeTest(AssertMixin): sess2 = create_session() u2 = sess2.query(User).options(eagerload('addresses')).get(7) - + sess3 = create_session() u3 = sess3.merge(u2, dont_load=True) def go(): sess3.flush() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_noload_disallows_dirty(self): - """noload doesnt support 'dirty' objects right now (see test_noload_with_eager()). + """noload doesnt support 'dirty' objects right now (see test_noload_with_eager()). Therefore lets assert it.""" - + mapper(User, users) sess = create_session() u = User() @@ -256,16 +256,16 @@ class MergeTest(AssertMixin): assert False except exceptions.InvalidRequestError, e: assert "merge() with dont_load=True option does not support objects marked as 'dirty'. flush() all changes on mapped instances before merging with dont_load=True." in str(e) - + u2 = sess2.query(User).get(7) - + sess3 = create_session() u3 = sess3.merge(u2, dont_load=True) assert not sess3.dirty def go(): sess3.flush() - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_noload_sets_entityname(self): """test that a noload-merged entity has entity_name set, has_mapper() passes, and lazyloads work""" mapper(User, users, properties={ @@ -282,10 +282,10 @@ class MergeTest(AssertMixin): sess.save(u) sess.flush() sess.clear() - + # reload 'u' such that its addresses list hasn't loaded u = sess.query(User).get(7) - + sess2 = create_session() u2 = sess2.merge(u, dont_load=True) assert not sess2.dirty @@ -295,7 +295,7 @@ class MergeTest(AssertMixin): def go(): assert u2.addresses != [] assert len(u2.addresses) == 1 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_noload_sets_backrefs(self): mapper(User, users, properties={ @@ -319,19 +319,19 @@ class MergeTest(AssertMixin): assert not sess2.dirty def go(): assert u2.addresses[0].user is u2 - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_noload_preserves_parents(self): """test that merge with noload does not trigger a 'delete-orphan' operation. - + merge with noload sets attributes without using events. this means the 'hasparent' flag is not propagated to the newly merged instance. in fact this - works out OK, because the '_state.parents' collection on the newly + works out OK, because the '_state.parents' collection on the newly merged instance is empty; since the mapper doesn't see an active 'False' setting - in this collection when _is_orphan() is called, it does not count as an orphan + in this collection when _is_orphan() is called, it does not count as an orphan (i.e. this is the 'optimistic' logic in mapper._is_orphan().) """ - + mapper(User, users, properties={ 'addresses':relation(mapper(Address, addresses),backref='user', cascade="all, delete-orphan") }) @@ -356,9 +356,9 @@ class MergeTest(AssertMixin): sess2.flush() sess2.clear() assert sess2.query(User).get(u2.user_id).addresses[0].email_address == 'somenewaddress' - + # this use case is not supported; this is with a pending Address on the pre-merged - # object, and we currently dont support 'dirty' objects being merged with dont_load=True. + # object, and we currently dont support 'dirty' objects being merged with dont_load=True. # in this case, the empty '_state.parents' collection would be an issue, # since the optimistic flag is False in _is_orphan() for pending instances. # so if we start supporting 'dirty' with dont_load=True, this test will need to pass @@ -379,7 +379,7 @@ class MergeTest(AssertMixin): assert sess2.query(User).get(u2.user_id).addresses[0].email_address == 'somenewaddress' except exceptions.InvalidRequestError, e: assert "dont_load=True option does not support" in str(e) - - -if __name__ == "__main__": - testbase.main() + + +if __name__ == "__main__": + testenv.main() diff --git a/test/orm/naturalpks.py b/test/orm/naturalpks.py index 2a8bead05..ec7d2fca9 100644 --- a/test/orm/naturalpks.py +++ b/test/orm/naturalpks.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from sqlalchemy import exceptions @@ -11,43 +11,43 @@ from testlib import * class NaturalPKTest(ORMTest): def define_tables(self, metadata): global users, addresses, items, users_to_items - + users = Table('users', metadata, Column('username', String(50), primary_key=True), Column('fullname', String(100))) - + addresses = Table('addresses', metadata, Column('email', String(50), primary_key=True), Column('username', String(50), ForeignKey('users.username', onupdate="cascade"))) - + items = Table('items', metadata, Column('itemname', String(50), primary_key=True), Column('description', String(100))) - + users_to_items = Table('userstoitems', metadata, Column('username', String(50), ForeignKey('users.username', onupdate='cascade'), primary_key=True), Column('itemname', String(50), ForeignKey('items.itemname', onupdate='cascade'), primary_key=True), ) - + def test_entity(self): mapper(User, users) - + sess = create_session() u1 = User(username='jack', fullname='jack') - + sess.save(u1) sess.flush() assert sess.get(User, 'jack') is u1 - + u1.username = 'ed' sess.flush() - + def go(): assert sess.get(User, 'ed') is u1 - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) assert sess.get(User, 'jack') is None - + sess.clear() u1 = sess.query(User).get('ed') self.assertEquals(User(username='ed', fullname='jack'), u1) @@ -63,7 +63,7 @@ class NaturalPKTest(ORMTest): assert sess.get(User, 'jack') is u1 users.update(values={u1.c.username:'jack'}).execute(username='ed') - + try: # expire/refresh works off of primary key. the PK is gone # in this case so theres no way to look it up. criterion- @@ -77,47 +77,47 @@ class NaturalPKTest(ORMTest): sess.clear() assert sess.get(User, 'jack') is None assert sess.get(User, 'ed').fullname == 'jack' - + @testing.unsupported('sqlite','mysql') def test_onetomany_passive(self): self._test_onetomany(True) - + def test_onetomany_nonpassive(self): self._test_onetomany(False) - + def _test_onetomany(self, passive_updates): mapper(User, users, properties={ 'addresses':relation(Address, passive_updates=passive_updates) }) mapper(Address, addresses) - + sess = create_session() u1 = User(username='jack', fullname='jack') u1.addresses.append(Address(email='jack1')) u1.addresses.append(Address(email='jack2')) sess.save(u1) sess.flush() - + assert sess.get(Address, 'jack1') is u1.addresses[0] - + u1.username = 'ed' sess.flush() assert u1.addresses[0].username == 'ed' - + sess.clear() self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all()) - + u1 = sess.get(User, 'ed') u1.username = 'jack' def go(): sess.flush() if not passive_updates: - self.assert_sql_count(testbase.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses + self.assert_sql_count(testing.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses else: - self.assert_sql_count(testbase.db, go, 1) # test passive_updates=True; update user + self.assert_sql_count(testing.db, go, 1) # test passive_updates=True; update user sess.clear() assert User(username='jack', addresses=[Address(username='jack'), Address(username='jack')]) == sess.get(User, 'jack') - + u1 = sess.get(User, 'jack') u1.addresses = [] u1.username = 'fred' @@ -133,39 +133,39 @@ class NaturalPKTest(ORMTest): def test_manytoone_nonpassive(self): self._test_manytoone(False) - + def _test_manytoone(self, passive_updates): mapper(User, users) mapper(Address, addresses, properties={ 'user':relation(User, passive_updates=passive_updates) }) - + sess = create_session() a1 = Address(email='jack1') a2 = Address(email='jack2') - + u1 = User(username='jack', fullname='jack') a1.user = u1 a2.user = u1 sess.save(a1) sess.save(a2) sess.flush() - + u1.username = 'ed' - + print id(a1), id(a2), id(u1) print u1._state.parents def go(): sess.flush() if passive_updates: - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) else: - self.assert_sql_count(testbase.db, go, 3) - + self.assert_sql_count(testing.db, go, 3) + def go(): sess.flush() - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + assert a1.username == a2.username == 'ed' sess.clear() self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all()) @@ -200,9 +200,9 @@ class NaturalPKTest(ORMTest): def go(): sess.flush() if passive_updates: - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) else: - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) self.assertEquals([Address(username='ed'), Address(username='ed')], [ad1, ad2]) sess.clear() self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all()) @@ -215,32 +215,32 @@ class NaturalPKTest(ORMTest): sess.flush() # check that the passive_updates is on on the other side if passive_updates: - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) else: - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) sess.clear() self.assertEquals([Address(username='fred'), Address(username='fred')], sess.query(Address).all()) - - + + @testing.unsupported('sqlite', 'mysql') def test_manytomany_passive(self): self._test_manytomany(True) - + def test_manytomany_nonpassive(self): self._test_manytomany(False) - + def _test_manytomany(self, passive_updates): mapper(User, users, properties={ 'items':relation(Item, secondary=users_to_items, backref='users', passive_updates=passive_updates) }) mapper(Item, items) - + sess = create_session() u1 = User(username='jack') u2 = User(username='fred') i1 = Item(itemname='item1') i2 = Item(itemname='item2') - + u1.items.append(i1) u1.items.append(i2) i2.users.append(u2) @@ -254,39 +254,39 @@ class NaturalPKTest(ORMTest): self.assertEquals(['jack'], [u.username for u in r[0].users]) self.assertEquals(Item(itemname='item2'), r[1]) self.assertEquals(['jack', 'fred'], [u.username for u in r[1].users]) - + u2.username='ed' def go(): sess.flush() go() def go(): sess.flush() - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + sess.clear() r = sess.query(Item).all() self.assertEquals(Item(itemname='item1'), r[0]) self.assertEquals(['jack'], [u.username for u in r[0].users]) self.assertEquals(Item(itemname='item2'), r[1]) self.assertEquals(['ed', 'jack'], sorted([u.username for u in r[1].users])) - + class SelfRefTest(ORMTest): def define_tables(self, metadata): global nodes, Node - + nodes = Table('nodes', metadata, Column('name', String(50), primary_key=True), Column('parent', String(50), ForeignKey('nodes.name', onupdate='cascade')) ) - + class Node(Base): pass - + def test_onetomany(self): mapper(Node, nodes, properties={ 'children':relation(Node, backref=backref('parentnode', remote_side=nodes.c.name, passive_updates=False), passive_updates=False) }) - + sess = create_session() n1 = Node(name='n1') n1.children.append(Node(name='n11')) @@ -294,13 +294,13 @@ class SelfRefTest(ORMTest): n1.children.append(Node(name='n13')) sess.save(n1) sess.flush() - + n1.name = 'new n1' sess.flush() self.assertEquals(n1.children[1].parent, 'new n1') self.assertEquals(['new n1', 'new n1', 'new n1'], [n.parent for n in sess.query(Node).filter(Node.name.in_(['n11', 'n12', 'n13']))]) - - + + class NonPKCascadeTest(ORMTest): def define_tables(self, metadata): global users, addresses @@ -335,9 +335,9 @@ class NonPKCascadeTest(ORMTest): sess.save(u1) sess.flush() a1 = u1.addresses[0] - + self.assertEquals(select([addresses.c.username]).execute().fetchall(), [('jack',), ('jack',)]) - + assert sess.get(Address, a1.id) is u1.addresses[0] u1.username = 'ed' @@ -353,13 +353,13 @@ class NonPKCascadeTest(ORMTest): def go(): sess.flush() if not passive_updates: - self.assert_sql_count(testbase.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses + self.assert_sql_count(testing.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses else: - self.assert_sql_count(testbase.db, go, 1) # test passive_updates=True; update user + self.assert_sql_count(testing.db, go, 1) # test passive_updates=True; update user sess.clear() assert User(username='jack', addresses=[Address(username='jack'), Address(username='jack')]) == sess.get(User, u1.id) sess.clear() - + u1 = sess.get(User, u1.id) u1.addresses = [] u1.username = 'fred' @@ -373,8 +373,6 @@ class NonPKCascadeTest(ORMTest): u1 = sess.get(User, u1.id) self.assertEquals(User(username='fred', fullname='jack'), u1) - + if __name__ == '__main__': - testbase.main() - - + testenv.main() diff --git a/test/orm/onetoone.py b/test/orm/onetoone.py index 8963a0c83..9c1ea9f16 100644 --- a/test/orm/onetoone.py +++ b/test/orm/onetoone.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from sqlalchemy.ext.sessioncontext import SessionContext @@ -28,7 +28,7 @@ class O2OTest(AssertMixin): @testing.uses_deprecated('SessionContext') def setUpAll(self): global jack, port, metadata, ctx - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) ctx = SessionContext(create_session) jack = Table('jack', metadata, Column('id', Integer, primary_key=True), @@ -90,4 +90,4 @@ class O2OTest(AssertMixin): ctx.current.flush() if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/pickled.py b/test/orm/pickled.py index a9a973cc0..84f5e5daf 100644 --- a/test/orm/pickled.py +++ b/test/orm/pickled.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -8,29 +8,29 @@ import pickle class EmailUser(User): pass - + class PickleTest(FixtureTest): keep_mappers = False keep_data = False - + def test_transient(self): mapper(User, users, properties={ 'addresses':relation(Address, backref="user") }) mapper(Address, addresses) - + sess = create_session() u1 = User(name='ed') u1.addresses.append(Address(email_address='ed@bar.com')) - + u2 = pickle.loads(pickle.dumps(u1)) sess.save(u2) sess.flush() - + sess.clear() - + self.assertEquals(u1, sess.query(User).get(u2.id)) - + def test_class_deferred_cols(self): mapper(User, users, properties={ 'name':deferred(users.c.name), @@ -48,7 +48,7 @@ class PickleTest(FixtureTest): u1 = sess.query(User).get(u1.id) assert 'name' not in u1.__dict__ assert 'addresses' not in u1.__dict__ - + u2 = pickle.loads(pickle.dumps(u1)) sess2 = create_session() sess2.update(u2) @@ -60,24 +60,24 @@ class PickleTest(FixtureTest): u2 = sess2.merge(u2, dont_load=True) self.assertEquals(u2.name, 'ed') self.assertEquals(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')])) - + def test_instance_deferred_cols(self): mapper(User, users, properties={ 'addresses':relation(Address, backref="user") }) mapper(Address, addresses) - + sess = create_session() u1 = User(name='ed') u1.addresses.append(Address(email_address='ed@bar.com')) sess.save(u1) sess.flush() sess.clear() - + u1 = sess.query(User).options(defer('name'), defer('addresses.email_address')).get(u1.id) assert 'name' not in u1.__dict__ assert 'addresses' not in u1.__dict__ - + u2 = pickle.loads(pickle.dumps(u1)) sess2 = create_session() sess2.update(u2) @@ -111,26 +111,24 @@ class PolymorphicDeferredTest(ORMTest): Column('id', Integer, ForeignKey('users.id'), primary_key=True), Column('email_address', String(30)) ) - + def test_polymorphic_deferred(self): mapper(User, users, polymorphic_identity='user', polymorphic_on=users.c.type, polymorphic_fetch='deferred') mapper(EmailUser, email_users, inherits=User, polymorphic_identity='emailuser') - + eu = EmailUser(name="user1", email_address='foo@bar.com') sess = create_session() sess.save(eu) sess.flush() sess.clear() - + eu = sess.query(User).first() eu2 = pickle.loads(pickle.dumps(eu)) sess2 = create_session() sess2.update(eu2) assert 'email_address' not in eu2.__dict__ self.assertEquals(eu2.email_address, 'foo@bar.com') - - - + if __name__ == '__main__': - testbase.main()
\ No newline at end of file + testenv.main() diff --git a/test/orm/query.py b/test/orm/query.py index 2d06be363..cfef709cf 100644 --- a/test/orm/query.py +++ b/test/orm/query.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import operator from sqlalchemy import * from sqlalchemy import exceptions @@ -48,7 +48,7 @@ class UnicodeSchemaTest(QueryTest): def test_get(self): mapper(User, uni_users) - assert User(id=7) == create_session(bind=testbase.db).query(User).get(7) + assert User(id=7) == create_session(bind=testing.db).query(User).get(7) class GetTest(QueryTest): def test_get(self): @@ -436,7 +436,7 @@ class DistinctTest(QueryTest): Address(id=5) ]), ] == q.all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) class YieldTest(QueryTest): @@ -732,14 +732,14 @@ class InstancesTest(QueryTest): def go(): l = q.options(contains_alias('ulist'), contains_eager('addresses')).instances(query.execute()) assert fixtures.user_address_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() def go(): l = q.options(contains_alias('ulist'), contains_eager('addresses')).from_statement(query).all() assert fixtures.user_address_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_contains_eager(self): @@ -750,14 +750,14 @@ class InstancesTest(QueryTest): def go(): l = q.options(contains_eager('addresses')).instances(selectquery.execute()) assert fixtures.user_address_result[0:3] == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() def go(): l = q.options(contains_eager('addresses')).from_statement(selectquery).all() assert fixtures.user_address_result[0:3] == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_contains_eager_alias(self): adalias = addresses.alias('adalias') @@ -769,14 +769,14 @@ class InstancesTest(QueryTest): # test using a string alias name l = q.options(contains_eager('addresses', alias="adalias")).instances(selectquery.execute()) assert fixtures.user_address_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() def go(): # test using the Alias object itself l = q.options(contains_eager('addresses', alias=adalias)).instances(selectquery.execute()) assert fixtures.user_address_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() @@ -790,7 +790,7 @@ class InstancesTest(QueryTest): # test using a custom 'decorate' function l = q.options(contains_eager('addresses', decorator=decorate)).instances(selectquery.execute()) assert fixtures.user_address_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() oalias = orders.alias('o1') @@ -801,7 +801,7 @@ class InstancesTest(QueryTest): def go(): l = q.options(contains_eager('orders', alias='o1'), contains_eager('orders.items', alias='i1')).instances(query.execute()) assert fixtures.user_order_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() @@ -809,7 +809,7 @@ class InstancesTest(QueryTest): def go(): l = q.options(contains_eager('orders', alias=oalias), contains_eager('orders.items', alias=ialias)).instances(query.execute()) assert fixtures.user_order_result == l - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() @@ -1055,7 +1055,7 @@ class SelectFromTest(QueryTest): ]), Order(description=u'order 5',items=[Item(description=u'item 5',keywords=[])])]) ]) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() sel2 = orders.select(orders.c.id.in_([1,2,3])) @@ -1085,19 +1085,19 @@ class SelectFromTest(QueryTest): User(id=8, addresses=[Address(id=2), Address(id=3), Address(id=4)]) ] ) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() def go(): self.assertEquals(sess.query(User).options(eagerload('addresses')).select_from(sel).filter(User.c.id==8).all(), [User(id=8, addresses=[Address(id=2), Address(id=3), Address(id=4)])] ) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() def go(): self.assertEquals(sess.query(User).options(eagerload('addresses')).select_from(sel)[1], User(id=8, addresses=[Address(id=2), Address(id=3), Address(id=4)])) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) class CustomJoinTest(QueryTest): keep_mappers = False @@ -1227,7 +1227,7 @@ class ExternalColumnsTest(QueryTest): sess.clear() def go(): assert address_result == sess.query(Address).options(eagerload('user')).all() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) tuple_address_result = [(address, address.user) for address in address_result] @@ -1238,4 +1238,4 @@ class ExternalColumnsTest(QueryTest): if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/relationships.py b/test/orm/relationships.py index 5c657963b..e23f107ee 100644 --- a/test/orm/relationships.py +++ b/test/orm/relationships.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import datetime from sqlalchemy import * from sqlalchemy import exceptions, types @@ -46,8 +46,8 @@ class RelationTest(PersistTest): ) def setUp(self): global session - session = create_session(bind=testbase.db) - conn = testbase.db.connect() + session = create_session(bind=testing.db) + conn = testing.db.connect() conn.create(tbl_a) conn.create(tbl_b) conn.create(tbl_c) @@ -85,14 +85,14 @@ class RelationTest(PersistTest): session.save_or_update(b) def tearDown(self): - conn = testbase.db.connect() + conn = testing.db.connect() conn.drop(tbl_d) conn.drop(tbl_c) conn.drop(tbl_b) conn.drop(tbl_a) def tearDownAll(self): - metadata.drop_all(testbase.db) + metadata.drop_all(testing.db) def testDeleteRootTable(self): session.flush() @@ -114,7 +114,7 @@ class RelationTest2(PersistTest): def setUpAll(self): global metadata, company_tbl, employee_tbl - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) company_tbl = Table('company', metadata, Column('company_id', Integer, primary_key=True), @@ -232,7 +232,7 @@ class RelationTest3(PersistTest): def setUpAll(self): global jobs, pageversions, pages, metadata, Job, Page, PageVersion, PageComment import datetime - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) jobs = Table("jobs", metadata, Column("jobno", Unicode(15), primary_key=True), Column("created", DateTime, nullable=False, default=datetime.datetime.now), @@ -347,7 +347,7 @@ class RelationTest3(PersistTest): s.save(j1) s.save(j2) - + s.flush() s.clear() @@ -1142,4 +1142,4 @@ class ViewOnlyTest2(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/selectable.py b/test/orm/selectable.py index 3478bc922..30771ec2b 100644 --- a/test/orm/selectable.py +++ b/test/orm/selectable.py @@ -1,6 +1,6 @@ """all tests involving generic mapping to Select statements""" -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -16,7 +16,7 @@ class SelectableNoFromsTest(ORMTest): Column('data', Integer), Column('extra', String(45)), ) - + def test_no_tables(self): class Subset(object): pass @@ -27,7 +27,7 @@ class SelectableNoFromsTest(ORMTest): assert False except exceptions.InvalidRequestError, e: assert str(e) == "Could not find any Table objects in mapped table 'SELECT x, y, z'", str(e) - + def test_basic(self): class Subset(Base): pass @@ -35,7 +35,7 @@ class SelectableNoFromsTest(ORMTest): subset_select = select([common_table.c.id, common_table.c.data]).alias('subset') subset_mapper = mapper(Subset, subset_select) - sess = create_session(bind=testbase.db) + sess = create_session(bind=testing.db) l = Subset() l.data = 1 sess.save(l) @@ -45,6 +45,6 @@ class SelectableNoFromsTest(ORMTest): assert [Subset(data=1)] == sess.query(Subset).all() # TODO: more tests mapping to selects - + if __name__ == '__main__': - testbase.main() + testenv.main() diff --git a/test/orm/session.py b/test/orm/session.py index 260f74876..0a6157e0a 100644 --- a/test/orm/session.py +++ b/test/orm/session.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -28,7 +28,7 @@ class SessionTest(AssertMixin): def test_close(self): """test that flush() doenst close a connection the session didnt open""" - c = testbase.db.connect() + c = testing.db.connect() class User(object):pass mapper(User, users) s = create_session(bind=c) @@ -45,7 +45,7 @@ class SessionTest(AssertMixin): s.flush() def test_close_two(self): - c = testbase.db.connect() + c = testing.db.connect() try: class User(object):pass mapper(User, users) @@ -86,7 +86,7 @@ class SessionTest(AssertMixin): @engines.close_open_connections def test_binds_from_expression(self): """test that Session can extract Table objects from ClauseElements and match them to tables.""" - Session = sessionmaker(binds={users:testbase.db, addresses:testbase.db}) + Session = sessionmaker(binds={users:testing.db, addresses:testing.db}) sess = Session() sess.execute(users.insert(), params=dict(user_id=1, user_name='ed')) assert sess.execute(users.select()).fetchall() == [(1, 'ed')] @@ -95,7 +95,7 @@ class SessionTest(AssertMixin): mapper(User, users, properties={ 'addresses':relation(Address, backref=backref("user", cascade="all"), cascade="all") }) - Session = sessionmaker(binds={User:testbase.db, Address:testbase.db}) + Session = sessionmaker(binds={User:testing.db, Address:testing.db}) sess.execute(users.insert(), params=dict(user_id=2, user_name='fred')) assert sess.execute(users.select()).fetchall() == [(1, 'ed'), (2, 'fred')] sess.close() @@ -105,8 +105,8 @@ class SessionTest(AssertMixin): def test_transaction(self): class User(object):pass mapper(User, users) - conn1 = testbase.db.connect() - conn2 = testbase.db.connect() + conn1 = testing.db.connect() + conn2 = testing.db.connect() sess = create_session(transactional=True, bind=conn1) u = User() @@ -116,7 +116,7 @@ class SessionTest(AssertMixin): assert conn2.execute("select count(1) from users").scalar() == 0 sess.commit() assert conn1.execute("select count(1) from users").scalar() == 1 - assert testbase.db.connect().execute("select count(1) from users").scalar() == 1 + assert testing.db.connect().execute("select count(1) from users").scalar() == 1 sess.close() @testing.unsupported('sqlite', 'mssql') # TEMP: test causes mssql to hang @@ -124,8 +124,8 @@ class SessionTest(AssertMixin): def test_autoflush(self): class User(object):pass mapper(User, users) - conn1 = testbase.db.connect() - conn2 = testbase.db.connect() + conn1 = testing.db.connect() + conn2 = testing.db.connect() sess = create_session(bind=conn1, transactional=True, autoflush=True) u = User() @@ -137,7 +137,7 @@ class SessionTest(AssertMixin): assert conn2.execute("select count(1) from users").scalar() == 0 sess.commit() assert conn1.execute("select count(1) from users").scalar() == 1 - assert testbase.db.connect().execute("select count(1) from users").scalar() == 1 + assert testing.db.connect().execute("select count(1) from users").scalar() == 1 sess.close() @testing.unsupported('sqlite', 'mssql') # TEMP: test causes mssql to hang @@ -154,10 +154,10 @@ class SessionTest(AssertMixin): u2 = sess.query(User).filter_by(user_name='ed').one() assert u2 is u assert sess.execute("select count(1) from users", mapper=User).scalar() == 1 - assert testbase.db.connect().execute("select count(1) from users").scalar() == 0 + assert testing.db.connect().execute("select count(1) from users").scalar() == 0 sess.commit() assert sess.execute("select count(1) from users", mapper=User).scalar() == 1 - assert testbase.db.connect().execute("select count(1) from users").scalar() == 1 + assert testing.db.connect().execute("select count(1) from users").scalar() == 1 sess.close() except: sess.rollback() @@ -167,8 +167,8 @@ class SessionTest(AssertMixin): def test_autoflush_2(self): class User(object):pass mapper(User, users) - conn1 = testbase.db.connect() - conn2 = testbase.db.connect() + conn1 = testing.db.connect() + conn2 = testing.db.connect() sess = create_session(bind=conn1, transactional=True, autoflush=True) u = User() @@ -176,7 +176,7 @@ class SessionTest(AssertMixin): sess.save(u) sess.commit() assert conn1.execute("select count(1) from users").scalar() == 1 - assert testbase.db.connect().execute("select count(1) from users").scalar() == 1 + assert testing.db.connect().execute("select count(1) from users").scalar() == 1 sess.commit() # TODO: not doing rollback of attributes right now. @@ -206,7 +206,7 @@ class SessionTest(AssertMixin): def test_external_joined_transaction(self): class User(object):pass mapper(User, users) - conn = testbase.db.connect() + conn = testing.db.connect() trans = conn.begin() sess = create_session(bind=conn, transactional=True, autoflush=True) sess.begin() @@ -225,7 +225,7 @@ class SessionTest(AssertMixin): class User(object):pass mapper(User, users) try: - conn = testbase.db.connect() + conn = testing.db.connect() trans = conn.begin() sess = create_session(bind=conn, transactional=True, autoflush=True) u1 = User() @@ -248,7 +248,7 @@ class SessionTest(AssertMixin): 'oracle', 'maxdb') @engines.close_open_connections def test_heavy_nesting(self): - session = create_session(bind=testbase.db) + session = create_session(bind=testing.db) session.begin() session.connection().execute("insert into users (user_name) values ('user1')") @@ -279,9 +279,9 @@ class SessionTest(AssertMixin): mapper(User, users) mapper(Address, addresses) - engine2 = create_engine(testbase.db.url) + engine2 = create_engine(testing.db.url) sess = create_session(transactional=False, autoflush=False, twophase=True) - sess.bind_mapper(User, testbase.db) + sess.bind_mapper(User, testing.db) sess.bind_mapper(Address, engine2) sess.begin() u1 = User() @@ -359,29 +359,29 @@ class SessionTest(AssertMixin): def test_bound_connection(self): class User(object):pass mapper(User, users) - c = testbase.db.connect() + c = testing.db.connect() sess = create_session(bind=c) sess.create_transaction() transaction = sess.transaction u = User() sess.save(u) sess.flush() - assert transaction.get_or_add(testbase.db) is transaction.get_or_add(c) is c + assert transaction.get_or_add(testing.db) is transaction.get_or_add(c) is c try: - transaction.add(testbase.db.connect()) + transaction.add(testing.db.connect()) assert False except exceptions.InvalidRequestError, e: assert str(e) == "Session already has a Connection associated for the given Connection's Engine" try: - transaction.get_or_add(testbase.db.connect()) + transaction.get_or_add(testing.db.connect()) assert False except exceptions.InvalidRequestError, e: assert str(e) == "Session already has a Connection associated for the given Connection's Engine" try: - transaction.add(testbase.db) + transaction.add(testing.db) assert False except exceptions.InvalidRequestError, e: assert str(e) == "Session already has a Connection associated for the given Engine" @@ -393,7 +393,7 @@ class SessionTest(AssertMixin): def test_bound_connection_transactional(self): class User(object):pass mapper(User, users) - c = testbase.db.connect() + c = testing.db.connect() sess = create_session(bind=c, transactional=True) u = User() @@ -413,7 +413,7 @@ class SessionTest(AssertMixin): c.execute("delete from users") assert c.scalar("select count(1) from users") == 0 - c = testbase.db.connect() + c = testing.db.connect() trans = c.begin() sess = create_session(bind=c, transactional=False) @@ -813,8 +813,8 @@ class SessionTest(AssertMixin): b = Bar() assert b in sess assert len(list(sess)) == 1 - - + + class ScopedSessionTest(ORMTest): def define_tables(self, metadata): @@ -851,7 +851,7 @@ class ScopedSessionTest(ORMTest): class ScopedMapperTest(PersistTest): def setUpAll(self): global metadata, table, table2 - metadata = MetaData(testbase.db) + metadata = MetaData(testing.db) table = Table('sometable', metadata, Column('id', Integer, primary_key=True), Column('data', String(30))) @@ -976,4 +976,4 @@ class ScopedMapperTest2(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/sessioncontext.py b/test/orm/sessioncontext.py index 4c09910e5..d6e652477 100644 --- a/test/orm/sessioncontext.py +++ b/test/orm/sessioncontext.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy.orm import * from sqlalchemy.ext.sessioncontext import SessionContext @@ -45,4 +45,4 @@ class SessionContextTest(AssertMixin): if __name__ == "__main__": - testbase.main() + testenv.main() diff --git a/test/orm/sharding/alltests.py b/test/orm/sharding/alltests.py index 0cdb838a9..aab3993f3 100644 --- a/test/orm/sharding/alltests.py +++ b/test/orm/sharding/alltests.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import unittest def suite(): @@ -15,4 +15,4 @@ def suite(): if __name__ == '__main__': - testbase.main(suite()) + testenv.main(suite()) diff --git a/test/orm/sharding/shard.py b/test/orm/sharding/shard.py index 7255155fa..283fac0e3 100644 --- a/test/orm/sharding/shard.py +++ b/test/orm/sharding/shard.py @@ -1,4 +1,4 @@ -import testbase +import testenv; testenv.configure_for_tests() import datetime, os from sqlalchemy import * from sqlalchemy import exceptions, sql @@ -12,7 +12,7 @@ from testlib import PersistTest class ShardTest(PersistTest): def setUpAll(self): global db1, db2, db3, db4, weather_locations, weather_reports - + db1 = create_engine('sqlite:///shard1.db') db2 = create_engine('sqlite:///shard2.db') db3 = create_engine('sqlite:///shard3.db') @@ -41,18 +41,18 @@ class ShardTest(PersistTest): Column('temperature', Float), Column('report_time', DateTime, default=datetime.datetime.now), ) - + for db in (db1, db2, db3, db4): meta.create_all(db) - + db1.execute(ids.insert(), nextid=1) self.setup_session() self.setup_mappers() - + def tearDownAll(self): for db in (db1, db2, db3, db4): - db.connect().invalidate() + db.connect().invalidate() for i in range(1,5): os.remove("shard%d.db" % i) @@ -65,7 +65,7 @@ class ShardTest(PersistTest): 'Europe':'europe', 'South America':'south_america' } - + def shard_chooser(mapper, instance, clause=None): if isinstance(instance, WeatherLocation): return shard_lookup[instance.continent] @@ -92,7 +92,7 @@ class ShardTest(PersistTest): return ['north_america', 'asia', 'europe', 'south_america'] else: return ids - + create_session = sessionmaker(class_=ShardedSession, autoflush=True, transactional=True) create_session.configure(shards={ @@ -101,11 +101,11 @@ class ShardTest(PersistTest): 'europe':db3, 'south_america':db4 }, shard_chooser=shard_chooser, id_chooser=id_chooser, query_chooser=query_chooser) - + def setup_mappers(self): global WeatherLocation, Report - + class WeatherLocation(object): def __init__(self, continent, city): self.continent = continent @@ -120,7 +120,7 @@ class ShardTest(PersistTest): 'city': deferred(weather_locations.c.city), }) - mapper(Report, weather_reports) + mapper(Report, weather_reports) def test_roundtrip(self): tokyo = WeatherLocation('Asia', 'Tokyo') @@ -144,7 +144,7 @@ class ShardTest(PersistTest): assert db2.execute(weather_locations.select()).fetchall() == [(1, 'Asia', 'Tokyo')] assert db1.execute(weather_locations.select()).fetchall() == [(2, 'North America', 'New York'), (3, 'North America', 'Toronto')] - + t = sess.query(WeatherLocation).get(tokyo.id) assert t.city == tokyo.city assert t.reports[0].temperature == 80.0 @@ -158,5 +158,4 @@ class ShardTest(PersistTest): if __name__ == '__main__': - testbase.main() - + testenv.main() diff --git a/test/orm/unitofwork.py b/test/orm/unitofwork.py index c31293e72..bcd4d2175 100644 --- a/test/orm/unitofwork.py +++ b/test/orm/unitofwork.py @@ -2,7 +2,7 @@ """Tests unitofwork operations.""" -import testbase +import testenv; testenv.configure_for_tests() import pickleable from sqlalchemy import * from sqlalchemy import exceptions, sql @@ -82,7 +82,7 @@ class VersioningTest(ORMTest): success = True # Only dialects with a sane rowcount can detect the ConcurrentModificationError - if testbase.db.dialect.supports_sane_rowcount: + if testing.db.dialect.supports_sane_rowcount: assert success s.close() @@ -100,7 +100,7 @@ class VersioningTest(ORMTest): except exceptions.ConcurrentModificationError, e: #print e success = True - if testbase.db.dialect.supports_sane_multi_rowcount: + if testing.db.dialect.supports_sane_multi_rowcount: assert success @engines.close_open_connections @@ -290,9 +290,9 @@ class MutableTypesTest(ORMTest): Session.commit() def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) f1.val = unicode('someothervalue') - self.assert_sql(testbase.db, lambda: Session.commit(), [ + self.assert_sql(testing.db, lambda: Session.commit(), [ ( "UPDATE mutabletest SET val=:val WHERE mutabletest.id = :mutabletest_id", {'mutabletest_id': f1.id, 'val': u'someothervalue'} @@ -300,7 +300,7 @@ class MutableTypesTest(ORMTest): ]) f1.val = unicode('hi') f1.data.x = 9 - self.assert_sql(testbase.db, lambda: Session.commit(), [ + self.assert_sql(testing.db, lambda: Session.commit(), [ ( "UPDATE mutabletest SET data=:data, val=:val WHERE mutabletest.id = :mutabletest_id", {'mutabletest_id': f1.id, 'val': u'hi', 'data':f1.data} @@ -317,7 +317,7 @@ class MutableTypesTest(ORMTest): def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) Session.close() @@ -325,12 +325,12 @@ class MutableTypesTest(ORMTest): def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) f2.data.y = 19 def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) Session.close() f3 = Session.query(Foo).filter_by(id=f1.id).one() @@ -339,7 +339,7 @@ class MutableTypesTest(ORMTest): def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_unicode(self): """test that two equivalent unicode values dont get flagged as changed. @@ -356,7 +356,7 @@ class MutableTypesTest(ORMTest): f1.val = u'hi' def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) class MutableTypesTest2(ORMTest): def define_tables(self, metadata): @@ -380,7 +380,7 @@ class MutableTypesTest2(ORMTest): Session.commit() def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) f1.data = [{'personne': {'nom': u'Smith', 'pers_id': 1, 'prenom': u'john', 'civilite': u'Mr', \ 'int_3': False, 'int_2': False, 'int_1': u'23', 'VenSoir': True, 'str_1': u'Test', \ @@ -388,12 +388,12 @@ class MutableTypesTest2(ORMTest): def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) f1.data[0]['personne']['VenSoir']= False def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) Session.clear() f = Session.query(Foo).get(f1.id) @@ -525,7 +525,7 @@ class ClauseAttributesTest(ORMTest): sess.flush() def go(): assert u.counter == 2 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_multi_update(self): class User(object): @@ -542,7 +542,7 @@ class ClauseAttributesTest(ORMTest): def go(): assert u.name == 'test2' assert u.counter == 2 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.clear() u = sess.query(User).get(u.id) @@ -680,10 +680,10 @@ class DefaultTest(ORMTest): defaults back from the engine.""" def define_tables(self, metadata): - db = testbase.db + db = testing.db use_string_defaults = db.engine.__module__.endswith('postgres') or db.engine.__module__.endswith('oracle') or db.engine.__module__.endswith('sqlite') global hohoval, althohoval - + if use_string_defaults: hohotype = String(30) hohoval = "im hoho" @@ -719,17 +719,17 @@ class DefaultTest(ORMTest): def go(): # test deferred load of attribues, one select per instance self.assert_(h2.hoho==h4.hoho==h5.hoho==hohoval) - self.assert_sql_count(testbase.db, go, 3) + self.assert_sql_count(testing.db, go, 3) def go(): self.assert_(h1.counter == h4.counter==h5.counter==7) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def go(): self.assert_(h3.counter == h2.counter == 12) self.assert_(h2.foober == h3.foober == h4.foober == 'im foober') self.assert_(h5.foober=='im the new foober') - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) Session.close() @@ -750,11 +750,11 @@ class DefaultTest(ORMTest): mapper(Hoho, default_table, eager_defaults=True) h1 = Hoho() Session.commit() - + def go(): self.assert_(h1.hoho==hohoval) - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_insert_nopostfetch(self): # populates the PassiveDefaults explicitly so there is no "post-update" class Hoho(object):pass @@ -767,7 +767,7 @@ class DefaultTest(ORMTest): self.assert_(h1.hoho=="15") self.assert_(h1.counter=="15") self.assert_(h1.foober=="im foober") - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_update(self): class Hoho(object):pass @@ -849,7 +849,7 @@ class OneToManyTest(ORMTest): u2.user_name = 'user2modified' u1.addresses.append(a3) del u1.addresses[0] - self.assert_sql(testbase.db, lambda: Session.commit(), + self.assert_sql(testing.db, lambda: Session.commit(), [ ( "UPDATE users SET user_name=:user_name WHERE users.user_id = :users_user_id", @@ -1121,7 +1121,7 @@ class SaveTest(ORMTest): u = Session.query(User).filter(User.user_name=='some name').one() def go(): u.user_name = 'some other name' - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) Session.flush() assert list(Session.execute(users.select(), mapper=User)) == [(42, 'some other name')] @@ -1147,7 +1147,7 @@ class SaveTest(ORMTest): u.user_name = "" def go(): Session.commit() - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) def test_multitable(self): """tests a save of an object where each instance spans two tables. also tests @@ -1269,7 +1269,7 @@ class ManyToOneTest(ORMTest): objects[2].email_address = 'imnew@foo.bar' objects[3].user = User() objects[3].user.user_name = 'imnewlyadded' - self.assert_sql(testbase.db, lambda: Session.commit(), [ + self.assert_sql(testing.db, lambda: Session.commit(), [ ( "INSERT INTO users (user_name) VALUES (:user_name)", {'user_name': 'imnewlyadded'} @@ -1459,7 +1459,7 @@ class ManyToManyTest(ORMTest): k = Keyword() k.name = 'yellow' objects[5].keywords.append(k) - self.assert_sql(testbase.db, lambda:Session.commit(), [ + self.assert_sql(testing.db, lambda:Session.commit(), [ { "UPDATE items SET item_name=:item_name WHERE items.item_id = :items_item_id": {'item_name': 'item4updated', 'items_item_id': objects[4].item_id} @@ -1488,7 +1488,7 @@ class ManyToManyTest(ORMTest): objects[2].keywords.append(k) dkid = objects[5].keywords[1].keyword_id del objects[5].keywords[1] - self.assert_sql(testbase.db, lambda:Session.commit(), [ + self.assert_sql(testing.db, lambda:Session.commit(), [ ( "DELETE FROM itemkeywords WHERE itemkeywords.item_id = :item_id AND itemkeywords.keyword_id = :keyword_id", [{'item_id': objects[5].item_id, 'keyword_id': dkid}] @@ -1667,7 +1667,7 @@ class SaveTest2(ORMTest): a.user = User() a.user.user_name = elem['user_name'] objects.append(a) - self.assert_sql(testbase.db, lambda: Session.commit(), [ + self.assert_sql(testing.db, lambda: Session.commit(), [ ( "INSERT INTO users (user_name) VALUES (:user_name)", {'user_name': 'thesub'} @@ -1929,4 +1929,4 @@ class RowSwitchTest(ORMTest): if __name__ == "__main__": - testbase.main() + testenv.main() |