class Group( Principal ):
pass
-class InheritTest(testbase.AssertMixin):
+class InheritTest(testbase.ORMTest):
"""deals with inheritance and many-to-many relationships"""
- def setUpAll(self):
+ def define_tables(self, metadata):
global principals
global users
global groups
global user_group_map
- global metadata
- metadata = BoundMetaData(testbase.db)
+
principals = Table(
'principals',
metadata,
)
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
-
- def setUp(self):
- clear_mappers()
-
def testbasic(self):
mapper( Principal, principals )
mapper(
sess.flush()
# TODO: put an assertion
-class InheritTest2(testbase.AssertMixin):
+class InheritTest2(testbase.ORMTest):
"""deals with inheritance and many-to-many relationships"""
- def setUpAll(self):
- global foo, bar, foo_bar, metadata
- metadata = BoundMetaData(testbase.db)
+ def define_tables(self, metadata):
+ global foo, bar, foo_bar
foo = Table('foo', metadata,
Column('id', Integer, Sequence('foo_id_seq'), primary_key=True),
Column('data', String(20)),
foo_bar = Table('foo_bar', metadata,
Column('foo_id', Integer, ForeignKey('foo.id')),
Column('bar_id', Integer, ForeignKey('bar.bid')))
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
def testbasic(self):
class Foo(object):
{'id':b.id, 'data':'barfoo', 'foos':(Foo, [{'id':f1.id,'data':'subfoo1'}, {'id':f2.id,'data':'subfoo2'}])},
)
-class InheritTest3(testbase.AssertMixin):
+class InheritTest3(testbase.ORMTest):
"""deals with inheritance and many-to-many relationships"""
- def setUpAll(self):
- global foo, bar, blub, bar_foo, blub_bar, blub_foo,metadata
- metadata = BoundMetaData(testbase.db)
+ def define_tables(self, metadata):
+ global foo, bar, blub, bar_foo, blub_bar, blub_foo
+
# the 'data' columns are to appease SQLite which cant handle a blank INSERT
foo = Table('foo', metadata,
Column('id', Integer, Sequence('foo_seq'), primary_key=True),
blub_foo = Table('blub_foo', metadata,
Column('blub_id', Integer, ForeignKey('blub.id')),
Column('foo_id', Integer, ForeignKey('foo.id')))
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
-
- def tearDown(self):
- for table in metadata.table_iterator():
- table.delete().execute()
def testbasic(self):
class Foo(object):
self.echo(x)
self.assert_(repr(x) == compare)
-class InheritTest4(testbase.AssertMixin):
+class InheritTest4(testbase.ORMTest):
"""deals with inheritance and one-to-many relationships"""
- def setUpAll(self):
- global foo, bar, blub, metadata
- metadata = BoundMetaData(testbase.db)
+ def define_tables(self, metadata):
+ global foo, bar, blub
# the 'data' columns are to appease SQLite which cant handle a blank INSERT
foo = Table('foo', metadata,
Column('id', Integer, Sequence('foo_seq'), primary_key=True),
Column('id', Integer, ForeignKey('bar.id'), primary_key=True),
Column('foo_id', Integer, ForeignKey('foo.id'), nullable=False),
Column('data', String(20)))
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
-
- def tearDown(self):
- for table in metadata.table_iterator():
- table.delete().execute()
def testbasic(self):
class Foo(object):
self.assert_(compare == result)
self.assert_(l[0].parent_foo.data == 'foo #1' and l[1].parent_foo.data == 'foo #1')
-class InheritTest5(testbase.AssertMixin):
+class InheritTest5(testbase.ORMTest):
"""testing that construction of inheriting mappers works regardless of when extra properties
are added to the superclass mapper"""
- def setUpAll(self):
- global content_type, content, product, metadata
- metadata = BoundMetaData(testbase.db)
+ def define_tables(self, metadata):
+ global content_type, content, product
content_type = Table('content_type', metadata,
Column('id', Integer, primary_key=True)
)
product = Table('product', metadata,
Column('id', Integer, ForeignKey('content.id'), primary_key=True)
)
- def tearDownAll(self):
- pass
- def tearDown(self):
- pass
def testbasic(self):
class ContentType(object): pass
p.contenttype = ContentType()
# TODO: assertion ??
-class InheritTest6(testbase.AssertMixin):
+class InheritTest6(testbase.ORMTest):
"""tests eager load/lazy load of child items off inheritance mappers, tests that
LazyLoader constructs the right query condition."""
- def setUpAll(self):
- global foo, bar, bar_foo, metadata
- metadata=BoundMetaData(testbase.db)
+ def define_tables(self, metadata):
+ global foo, bar, bar_foo
foo = Table('foo', metadata, Column('id', Integer, Sequence('foo_seq'), primary_key=True),
Column('data', String(30)))
bar = Table('bar', metadata, Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
Column('bar_id', Integer, ForeignKey('bar.id')),
Column('foo_id', Integer, ForeignKey('foo.id'))
)
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
-
+
def testbasic(self):
class Foo(object): pass
class Bar(Foo): pass
self.assert_(len(q.selectfirst().eager) == 1)
-class InheritTest7(testbase.AssertMixin):
+class InheritTest7(testbase.ORMTest):
"""test dependency sorting among inheriting mappers"""
- def setUpAll(self):
- global users, roles, user_roles, admins, metadata
- metadata=BoundMetaData(testbase.db)
+ def define_tables(self, metadata):
+ global users, roles, user_roles, admins
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('email', String(128)),
Column('id', Integer, primary_key=True),
Column('user_id', Integer, ForeignKey('users.id'))
)
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
- def tearDown(self):
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
def testone(self):
class User(object):pass
from sqlalchemy import *
from datetime import datetime
-class InheritTest(testbase.AssertMixin):
+class InheritTest(testbase.ORMTest):
"""tests some various inheritance round trips involving a particular set of polymorphic inheritance relationships"""
- def setUpAll(self):
- global metadata, products_table, specification_table, documents_table
+ def define_tables(self, metadata):
+ global products_table, specification_table, documents_table
global Product, Detail, Assembly, SpecLine, Document, RasterDocument
- metadata = BoundMetaData(testbase.db)
products_table = Table('products', metadata,
Column('product_id', Integer, primary_key=True),
Column('size', Integer, default=0),
)
- metadata.create_all()
-
class Product(object):
def __init__(self, name, mark=''):
self.name = name
class RasterDocument(Document):
pass
- def tearDown(self):
- clear_mappers()
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
-
- def tearDownAll(self):
- metadata.drop_all()
-
def testone(self):
product_mapper = mapper(Product, products_table,
polymorphic_on=products_table.c.product_type,
class ClassifiedPage(MagazinePage):
pass
-class InheritTest(testbase.AssertMixin):
+class InheritTest(testbase.ORMTest):
"""tests a large polymorphic relationship"""
- def setUpAll(self):
- global metadata, publication_table, issue_table, location_table, location_name_table, magazine_table, \
+ def define_tables(self, metadata):
+ global publication_table, issue_table, location_table, location_name_table, magazine_table, \
page_table, magazine_page_table, classified_page_table, page_size_table
-
- metadata = BoundMetaData(testbase.db)
zerodefault = {} #{'default':0}
publication_table = Table('publication', metadata,
Column('name', String(45), default=''),
)
- metadata.create_all()
-
publication_mapper = mapper(Publication, publication_table)
issue_mapper = mapper(Issue, issue_table, properties = {
classified_page_mapper = mapper(ClassifiedPage, classified_page_table, inherits=magazine_page_mapper, polymorphic_identity='c')
- def tearDown(self):
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
-
- def tearDownAll(self):
- metadata.drop_all()
- clear_mappers()
-
def testone(self):
session = create_session()
from sqlalchemy import *
import testbase
-class ConcreteTest1(testbase.AssertMixin):
- def setUpAll(self):
- global managers_table, engineers_table, metadata
- metadata = BoundMetaData(testbase.db)
+class ConcreteTest1(testbase.ORMTest):
+ def define_tables(self, metadata):
+ global managers_table, engineers_table
managers_table = Table('managers', metadata,
Column('employee_id', Integer, primary_key=True),
Column('name', String(50)),
Column('engineer_info', String(50)),
)
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
-
def testbasic(self):
class Employee(object):
def __init__(self, name):
return self.__class__.__name__ + ' ' + ','.join(["%s=%s" % (k,v) for k, v in self.__dict__.iteritems() if k[0] != '_'])
-class RelationTest1(testbase.PersistTest):
+class RelationTest1(testbase.ORMTest):
"""test self-referential relationships on polymorphic mappers"""
- def setUpAll(self):
- global people, managers, metadata
- metadata = BoundMetaData(testbase.db)
+ def define_tables(self, metadata):
+ global people, managers
people = Table('people', metadata,
Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
Column('manager_name', String(50))
)
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
-
- def tearDown(self):
- clear_mappers()
- people.update().execute(manager_id=None)
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
-
def testbasic(self):
class Person(AttrSettable):
pass
print p, m, p.manager
assert p.manager is m
-class RelationTest2(testbase.AssertMixin):
+class RelationTest2(testbase.ORMTest):
"""test self-referential relationships on polymorphic mappers"""
- def setUpAll(self):
- global people, managers, metadata
- metadata = BoundMetaData(testbase.db)
-
+ def define_tables(self, metadata):
+ global people, managers
people = Table('people', metadata,
Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
Column('name', String(50)),
Column('status', String(30)),
)
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
-
- def tearDown(self):
- clear_mappers()
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
-
def testrelationonsubclass(self):
class Person(AttrSettable):
pass
print m
assert m.colleague is p
-class RelationTest3(testbase.AssertMixin):
+class RelationTest3(testbase.ORMTest):
"""test self-referential relationships on polymorphic mappers"""
- def setUpAll(self):
- global people, managers, metadata
- metadata = BoundMetaData(testbase.db)
-
+ def define_tables(self, metadata):
+ global people, managers
people = Table('people', metadata,
Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
Column('colleague_id', Integer, ForeignKey('people.person_id')),
Column('status', String(30)),
)
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
-
- def tearDown(self):
- clear_mappers()
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
-
def testrelationonbaseclass(self):
class Person(AttrSettable):
pass
assert len(p.colleagues) == 1
assert p.colleagues == [p2]
-class RelationTest4(testbase.AssertMixin):
- def setUpAll(self):
- global metadata, people, engineers, managers, cars
- metadata = BoundMetaData(testbase.db)
+class RelationTest4(testbase.ORMTest):
+ def define_tables(self, metadata):
+ global people, engineers, managers, cars
people = Table('people', metadata,
Column('person_id', Integer, primary_key=True),
Column('name', String(50)))
cars = Table('cars', metadata,
Column('car_id', Integer, primary_key=True),
Column('owner', Integer, ForeignKey('people.person_id')))
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
- def tearDown(self):
- clear_mappers()
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
def testmanytoonepolymorphic(self):
"""in this test, the polymorphic union is between two subclasses, but does not include the base table by itself
car1 = session.query(Car).options(eagerload('employee')).get(car1.car_id)
assert str(car1.employee) == "Engineer E4, status X"
-class RelationTest5(testbase.AssertMixin):
- def setUpAll(self):
- global metadata, people, engineers, managers, cars
- metadata = BoundMetaData(testbase.db)
+class RelationTest5(testbase.ORMTest):
+ def define_tables(self, metadata):
+ global people, engineers, managers, cars
people = Table('people', metadata,
Column('person_id', Integer, primary_key=True),
Column('name', String(50)),
cars = Table('cars', metadata,
Column('car_id', Integer, primary_key=True),
Column('owner', Integer, ForeignKey('people.person_id')))
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
- def tearDown(self):
- clear_mappers()
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
def testeagerempty(self):
"""an easy one...test parent object with child relation to an inheriting mapper, using eager loads,
carlist = sess.query(Car).select()
assert carlist[0].manager is None
assert carlist[1].manager.person_id == car2.manager.person_id
+
+class MultiLevelTest(testbase.ORMTest):
+ def define_tables(self, metadata):
+ global table_Employee, table_Engineer, table_Manager
+ table_Employee = Table( 'Employee', metadata,
+ Column( 'name', type= String, ),
+ Column( 'id', primary_key= True, type= Integer, ),
+ Column( 'atype', type= String, ),
+ )
+
+ table_Engineer = Table( 'Engineer', metadata,
+ Column( 'machine', type= String, ),
+ Column( 'id', Integer, ForeignKey( 'Employee.id', ), primary_key= True, ),
+ )
+
+ table_Manager = Table( 'Manager', metadata,
+ Column( 'duties', type= String, ),
+ Column( 'id', Integer, ForeignKey( 'Engineer.id', ), primary_key= True, ),
+ )
+ def test_threelevels(self):
+ class Employee( object):
+ def set( me, **kargs):
+ for k,v in kargs.iteritems(): setattr( me, k, v)
+ return me
+ def __str__(me): return str(me.__class__.__name__)+':'+str(me.name)
+ __repr__ = __str__
+ class Engineer( Employee): pass
+ class Manager( Engineer): pass
+ pu_Employee = polymorphic_union( {
+ 'Manager': table_Employee.join( table_Engineer).join( table_Manager),
+ 'Engineer': select([table_Employee, table_Engineer.c.machine], table_Employee.c.atype == 'Engineer', from_obj=[table_Employee.join(table_Engineer)]),
+ 'Employee': table_Employee.select( table_Employee.c.atype == 'Employee'),
+ }, None, 'pu_employee', )
+
+ mapper_Employee = mapper( Employee, table_Employee,
+ polymorphic_identity= 'Employee',
+ polymorphic_on= pu_Employee.c.atype,
+ select_table= pu_Employee,
+ )
+
+ pu_Engineer = polymorphic_union( {
+ 'Manager': table_Employee.join( table_Engineer).join( table_Manager),
+ 'Engineer': select([table_Employee, table_Engineer.c.machine], table_Employee.c.atype == 'Engineer', from_obj=[table_Employee.join(table_Engineer)]),
+ }, None, 'pu_engineer', )
+ mapper_Engineer = mapper( Engineer, table_Engineer,
+ inherit_condition= table_Engineer.c.id == table_Employee.c.id,
+ inherits= mapper_Employee,
+ polymorphic_identity= 'Engineer',
+ polymorphic_on= pu_Engineer.c.atype,
+ select_table= pu_Engineer,
+ )
+
+ mapper_Manager = mapper( Manager, table_Manager,
+ inherit_condition= table_Manager.c.id == table_Engineer.c.id,
+ inherits= mapper_Engineer,
+ polymorphic_identity= 'Manager',
+ )
+
+ a = Employee().set( name= 'one')
+ b = Engineer().set( egn= 'two', machine= 'any')
+ c = Manager().set( name= 'head', machine= 'fast', duties= 'many')
+
+ session = create_session()
+ session.save(a)
+ session.save(b)
+ session.save(c)
+ session.flush()
+ assert set(session.query(Employee).select()) == set([a,b,c])
+ assert set(session.query( Engineer).select()) == set([b,c])
+ assert session.query( Manager).select() == [c]
if __name__ == "__main__":
testbase.main()
def __repr__(self):
return object.__repr__(self)+ " " + repr(self.inputs) + " " + repr(self.outputs)
-class M2MTest(testbase.AssertMixin):
- def setUpAll(self):
- global metadata
- metadata = testbase.metadata
+class M2MTest(testbase.ORMTest):
+ def define_tables(self, metadata):
global place
place = Table('place', metadata,
Column('place_id', Integer, Sequence('pid_seq', optional=True), primary_key=True),
Column('pl1_id', Integer, ForeignKey('place.place_id')),
Column('pl2_id', Integer, ForeignKey('place.place_id')),
)
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
- clear_mappers()
- #testbase.db.tables.clear()
-
- def setUp(self):
- clear_mappers()
-
- def tearDown(self):
- place_place.delete().execute()
- place_input.delete().execute()
- place_output.delete().execute()
- transition.delete().execute()
- place.delete().execute()
def testcircular(self):
"""tests a many-to-many relationship from a table to itself."""
self.assert_result([t1], Transition, {'outputs': (Place, [{'name':'place3'}, {'name':'place1'}])})
self.assert_result([p2], Place, {'inputs': (Transition, [{'name':'transition1'},{'name':'transition2'}])})
-class M2MTest2(testbase.AssertMixin):
- def setUpAll(self):
- metadata = testbase.metadata
+class M2MTest2(testbase.ORMTest):
+ def define_tables(self, metadata):
global studentTbl
studentTbl = Table('student', metadata, Column('name', String(20), primary_key=True))
global courseTbl
enrolTbl = Table('enrol', metadata,
Column('student_id', String(20), ForeignKey('student.name'),primary_key=True),
Column('course_id', String(20), ForeignKey('course.name'), primary_key=True))
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
- clear_mappers()
-
- def setUp(self):
- clear_mappers()
-
- def tearDown(self):
- enrolTbl.delete().execute()
- courseTbl.delete().execute()
- studentTbl.delete().execute()
def testcircular(self):
class Student(object):
del s.courses[1]
self.assert_(len(s.courses) == 2)
-class M2MTest3(testbase.AssertMixin):
- def setUpAll(self):
- metadata = testbase.metadata
+class M2MTest3(testbase.ORMTest):
+ def define_tables(self, metadata):
global c, c2a1, c2a2, b, a
c = Table('c', metadata,
Column('c1', Integer, primary_key = True),
Column('a1', Integer, ForeignKey('a.a1')),
Column('b2', Boolean)
)
- metadata.create_all()
-
- def tearDownAll(self):
- b.drop()
- c2a2.drop()
- c2a1.drop()
- a.drop()
- c.drop()
- clear_mappers()
def testbasic(self):
class C(object):pass
import re
import sqlalchemy
import optparse
-
+from sqlalchemy.schema import BoundMetaData
+from sqlalchemy.orm import clear_mappers
db = None
metadata = None
finally:
self.assert_(db.sql_count == count, "desired statement count %d does not match %d" % (count, db.sql_count))
+class ORMTest(AssertMixin):
+ def setUpAll(self):
+ global metadata
+ metadata = BoundMetaData(db)
+ self.define_tables(metadata)
+ metadata.create_all()
+ def define_tables(self, metadata):
+ raise NotImplementedError()
+ def get_metadata(self):
+ return metadata
+ def tearDownAll(self):
+ metadata.drop_all()
+ def tearDown(self):
+ clear_mappers()
+ for t in metadata.table_iterator(reverse=True):
+ t.delete().execute().close()
+
+
class EngineAssert(proxy.BaseProxyEngine):
"""decorates a SQLEngine object to match the incoming queries against a set of assertions."""
def __init__(self, engine):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
