diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/orm/eagertest3.py | 96 | ||||
| -rw-r--r-- | test/orm/mapper.py | 2 | ||||
| -rwxr-xr-x | test/sql/selectable.py | 29 |
3 files changed, 123 insertions, 4 deletions
diff --git a/test/orm/eagertest3.py b/test/orm/eagertest3.py index a731581d5..8e7735812 100644 --- a/test/orm/eagertest3.py +++ b/test/orm/eagertest3.py @@ -415,7 +415,101 @@ class EagerTest5(testbase.ORMTest): # object is not in the session; therefore the lazy load cant trigger here, # eager load had to succeed assert len([c for c in d2.comments]) == 1 + +class EagerTest6(testbase.ORMTest): + def define_tables(self, metadata): + global project_t, task_t, task_status_t, task_type_t, message_t, message_type_t - + project_t = Table('prj', metadata, + Column('id', Integer, primary_key=True), + Column('created', DateTime , ), + Column('title', Unicode(100)), + ) + + task_t = Table('task', metadata, + Column('id', Integer, primary_key=True), + Column('status_id', Integer, ForeignKey('task_status.id'), nullable=False), + Column('title', Unicode(100)), + Column('task_type_id', Integer , ForeignKey('task_type.id'), nullable=False), + Column('prj_id', Integer , ForeignKey('prj.id'), nullable=False), + ) + + task_status_t = Table('task_status', metadata, + Column('id', Integer, primary_key=True), + ) + + task_type_t = Table('task_type', metadata, + Column('id', Integer, primary_key=True), + ) + + message_t = Table('msg', metadata, + Column('id', Integer, primary_key=True), + Column('posted', DateTime, index=True,), + Column('type_id', Integer, ForeignKey('msg_type.id')), + Column('task_id', Integer, ForeignKey('task.id')), + ) + + message_type_t = Table('msg_type', metadata, + Column('id', Integer, primary_key=True), + Column('name', Unicode(20)), + Column('display_name', Unicode(20)), + ) + + def setUp(self): + testbase.db.execute("INSERT INTO prj (title) values('project 1');") + testbase.db.execute("INSERT INTO task_status (id) values(1);") + testbase.db.execute("INSERT INTO task_type(id) values(1);") + testbase.db.execute("INSERT INTO task (title, task_type_id, status_id, prj_id) values('task 1',1,1,1);") + + def test_nested_joins(self): + # this is testing some subtle column resolution stuff, + # concerning corresponding_column() being extremely accurate + # as well as how mapper sets up its column properties + + class Task(object):pass + class Task_Type(object):pass + class Message(object):pass + class Message_Type(object):pass + + tsk_cnt_join = outerjoin(project_t, task_t, task_t.c.prj_id==project_t.c.id) + + ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')], + from_obj=[tsk_cnt_join], group_by=[project_t.c.id]).alias('prj_tsk_cnt_s') + j = join(project_t, ss, project_t.c.id == ss.c.prj_id) + + mapper(Task_Type, task_type_t) + + mapper( Task, task_t, + properties=dict(type=relation(Task_Type, lazy=False), + )) + + mapper(Message_Type, message_type_t) + + mapper(Message, message_t, + properties=dict(type=relation(Message_Type, lazy=False, uselist=False), + )) + + tsk_cnt_join = outerjoin(project_t, task_t, task_t.c.prj_id==project_t.c.id) + ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')], + from_obj=[tsk_cnt_join], group_by=[project_t.c.id]).alias('prj_tsk_cnt_s') + j = join(project_t, ss, project_t.c.id == ss.c.prj_id) + + j = outerjoin( task_t, message_t, task_t.c.id==message_t.c.task_id) + jj = select([ task_t.c.id.label('task_id'), + func.count(message_t.c.id).label('props_cnt')], + from_obj=[j], group_by=[task_t.c.id]).alias('prop_c_s') + jjj = join(task_t, jj, task_t.c.id == jj.c.task_id) + + class cls(object):pass + + props =dict(type=relation(Task_Type, lazy=False)) + print [c.key for c in jjj.c] + cls.mapper = mapper( cls, jjj, properties=props) + + session = create_session() + + for t in session.query(cls.mapper).limit(10).offset(0).list(): + print t.id, t.title, t.props_cnt + if __name__ == "__main__": testbase.main() diff --git a/test/orm/mapper.py b/test/orm/mapper.py index 63af53b96..197f988b7 100644 --- a/test/orm/mapper.py +++ b/test/orm/mapper.py @@ -497,7 +497,7 @@ class MapperTest(MapperSuperTest): class_mapper(User) except exceptions.ArgumentError, e: assert str(e) == "Column '%s' is not represented in mapper's table. Use the `column_property()` function to force this column to be mapped as a read-only attribute." % str(f) - clear_mappers() + clear_mappers() mapper(User, users, properties={ 'concat': column_property(f), diff --git a/test/sql/selectable.py b/test/sql/selectable.py index 57ad99886..340c55837 100755 --- a/test/sql/selectable.py +++ b/test/sql/selectable.py @@ -27,17 +27,42 @@ table2 = Table('table2', db, )
class SelectableTest(testbase.AssertMixin):
+ def testdistance(self):
+ s = select([table.c.col1.label('c2'), table.c.col1, table.c.col1.label('c1')])
+
+ # didnt do this yet...col.label().make_proxy() has same "distance" as col.make_proxy() so far
+ #assert s.corresponding_column(table.c.col1) is s.c.col1
+ assert s.corresponding_column(s.c.col1) is s.c.col1
+ assert s.corresponding_column(s.c.c1) is s.c.c1
+
def testjoinagainstself(self):
jj = select([table.c.col1.label('bar_col1')])
jjj = join(table, jj, table.c.col1==jj.c.bar_col1)
+
+ # test column directly agaisnt itself
assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1
+ assert jjj.corresponding_column(jj.c.bar_col1) is jjj.c.bar_col1
+
+ # test alias of the join, targets the column with the least
+ # "distance" between the requested column and the returned column
+ # (i.e. there is less indirection between j2.c.table1_col1 and table.c.col1, than
+ # there is from j2.c.bar_col1 to table.c.col1)
+ j2 = jjj.alias('foo')
+ assert j2.corresponding_column(table.c.col1) is j2.c.table1_col1
+
+
def testjoinagainstjoin(self):
j = outerjoin(table, table2, table.c.col1==table2.c.col2)
jj = select([ table.c.col1.label('bar_col1')],from_obj=[j]).alias('foo')
jjj = join(table, jj, table.c.col1==jj.c.bar_col1)
assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1
+
+ j2 = jjj.alias('foo')
+ print j2.corresponding_column(jjj.c.table1_col1)
+ assert j2.corresponding_column(jjj.c.table1_col1) is j2.c.table1_col1
+ assert jjj.corresponding_column(jj.c.bar_col1) is jj.c.bar_col1
def testtablealias(self):
a = table.alias('a')
@@ -110,8 +135,8 @@ class SelectableTest(testbase.AssertMixin): j = join(a, table2)
criterion = a.c.col1 == table2.c.col2
- print
- print str(j)
+ print criterion
+ print j.onclause
self.assert_(criterion.compare(j.onclause))
def testselectlabels(self):
|