summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/orm/test_rel_fn.py246
-rw-r--r--test/orm/test_relationships.py79
-rw-r--r--test/sql/test_generative.py86
-rw-r--r--test/sql/test_selectable.py4
4 files changed, 303 insertions, 112 deletions
diff --git a/test/orm/test_rel_fn.py b/test/orm/test_rel_fn.py
index d3d346bba..346cb90c1 100644
--- a/test/orm/test_rel_fn.py
+++ b/test/orm/test_rel_fn.py
@@ -6,9 +6,8 @@ from sqlalchemy import MetaData, Table, Column, ForeignKey, Integer, \
select, ForeignKeyConstraint, exc
from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE, MANYTOMANY
-class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
- __dialect__ = 'default'
+class _JoinFixtures(object):
@classmethod
def setup_class(cls):
m = MetaData()
@@ -36,6 +35,28 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
['composite_selfref.id', 'composite_selfref.group_id']
)
)
+ cls.m2mleft = Table('m2mlft', m,
+ Column('id', Integer, primary_key=True),
+ )
+ cls.m2mright = Table('m2mrgt', m,
+ Column('id', Integer, primary_key=True),
+ )
+ cls.m2msecondary = Table('m2msecondary', m,
+ Column('lid', Integer, ForeignKey('m2mlft.id'), primary_key=True),
+ Column('rid', Integer, ForeignKey('m2mrgt.id'), primary_key=True),
+ )
+
+ def _join_fixture_m2m_selfref(self, **kw):
+ return relationships.JoinCondition(
+ self.m2mleft,
+ self.m2mright,
+ self.m2mleft,
+ self.m2mright,
+ secondary=self.m2msecondary,
+ primaryjoin=self.m2mleft.c.id==self.m2msecondary.c.lid,
+ secondaryjoin=self.m2mright.c.id==self.m2msecondary.c.rid,
+ **kw
+ )
def _join_fixture_o2m(self, **kw):
return relationships.JoinCondition(
@@ -120,6 +141,7 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
**kw
)
+class ColumnCollectionsTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
def test_determine_remote_columns_compound_1(self):
joincond = self._join_fixture_compound_expression_1(
support_sync=False)
@@ -133,7 +155,25 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
support_sync=False)
eq_(
joincond.local_remote_pairs,
- []
+ [
+ (self.left.c.x, self.right.c.x),
+ (self.left.c.x, self.right.c.y),
+ (self.left.c.y, self.right.c.x),
+ (self.left.c.y, self.right.c.y)
+ ]
+ )
+
+ def test_determine_local_remote_compound_2(self):
+ joincond = self._join_fixture_compound_expression_2(
+ support_sync=False)
+ eq_(
+ joincond.local_remote_pairs,
+ [
+ (self.left.c.x, self.right.c.x),
+ (self.left.c.x, self.right.c.y),
+ (self.left.c.y, self.right.c.x),
+ (self.left.c.y, self.right.c.y)
+ ]
)
def test_err_local_remote_compound_1(self):
@@ -160,14 +200,71 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
set([self.right.c.x, self.right.c.y])
)
- def test_determine_local_remote_compound_2(self):
- joincond = self._join_fixture_compound_expression_2(
- support_sync=False)
+
+ def test_determine_remote_columns_o2m(self):
+ joincond = self._join_fixture_o2m()
+ eq_(
+ joincond.remote_columns,
+ set([self.right.c.lid])
+ )
+
+ def test_determine_remote_columns_o2m_selfref(self):
+ joincond = self._join_fixture_o2m_selfref()
+ eq_(
+ joincond.remote_columns,
+ set([self.selfref.c.sid])
+ )
+
+ def test_determine_remote_columns_o2m_composite_selfref(self):
+ joincond = self._join_fixture_o2m_composite_selfref()
+ eq_(
+ joincond.remote_columns,
+ set([self.composite_selfref.c.parent_id,
+ self.composite_selfref.c.group_id])
+ )
+
+ def test_determine_remote_columns_m2o_composite_selfref(self):
+ joincond = self._join_fixture_m2o_composite_selfref()
+ eq_(
+ joincond.remote_columns,
+ set([self.composite_selfref.c.id,
+ self.composite_selfref.c.group_id])
+ )
+
+ def test_determine_remote_columns_m2o(self):
+ joincond = self._join_fixture_m2o()
+ eq_(
+ joincond.remote_columns,
+ set([self.left.c.id])
+ )
+
+ def test_determine_local_remote_pairs_o2m(self):
+ joincond = self._join_fixture_o2m()
eq_(
joincond.local_remote_pairs,
- []
+ [(self.left.c.id, self.right.c.lid)]
+ )
+
+ def test_determine_synchronize_pairs_m2m_selfref(self):
+ joincond = self._join_fixture_m2m_selfref()
+ eq_(
+ joincond.synchronize_pairs,
+ [(self.m2mleft.c.id, self.m2msecondary.c.lid)]
+ )
+ eq_(
+ joincond.secondary_synchronize_pairs,
+ [(self.m2mright.c.id, self.m2msecondary.c.rid)]
)
+ def test_determine_remote_columns_m2o_selfref(self):
+ joincond = self._join_fixture_m2o_selfref()
+ eq_(
+ joincond.remote_columns,
+ set([self.selfref.c.id])
+ )
+
+
+class DirectionTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
def test_determine_direction_compound_2(self):
joincond = self._join_fixture_compound_expression_2(
support_sync=False)
@@ -176,60 +273,46 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
ONETOMANY
)
- def test_determine_join_o2m(self):
- joincond = self._join_fixture_o2m()
- self.assert_compile(
- joincond.primaryjoin,
- "lft.id = rgt.lid"
- )
-
def test_determine_direction_o2m(self):
joincond = self._join_fixture_o2m()
is_(joincond.direction, ONETOMANY)
- def test_determine_remote_columns_o2m(self):
- joincond = self._join_fixture_o2m()
- eq_(
- joincond.remote_columns,
- set([self.right.c.lid])
- )
-
- def test_determine_join_o2m_selfref(self):
- joincond = self._join_fixture_o2m_selfref()
- self.assert_compile(
- joincond.primaryjoin,
- "selfref.id = selfref.sid"
- )
-
def test_determine_direction_o2m_selfref(self):
joincond = self._join_fixture_o2m_selfref()
is_(joincond.direction, ONETOMANY)
- def test_determine_remote_columns_o2m_selfref(self):
- joincond = self._join_fixture_o2m_selfref()
- eq_(
- joincond.remote_columns,
- set([self.selfref.c.sid])
- )
+ def test_determine_direction_m2o_selfref(self):
+ joincond = self._join_fixture_m2o_selfref()
+ is_(joincond.direction, MANYTOONE)
- def test_join_targets_o2m_selfref(self):
- joincond = self._join_fixture_o2m_selfref()
- left = select([joincond.parent_selectable]).alias('pj')
- pj, sj, sec, adapter = joincond.join_targets(
- left,
- joincond.child_selectable,
- True)
+ def test_determine_direction_o2m_composite_selfref(self):
+ joincond = self._join_fixture_o2m_composite_selfref()
+ is_(joincond.direction, ONETOMANY)
+
+ def test_determine_direction_m2o_composite_selfref(self):
+ joincond = self._join_fixture_m2o_composite_selfref()
+ is_(joincond.direction, MANYTOONE)
+
+ def test_determine_direction_m2o(self):
+ joincond = self._join_fixture_m2o()
+ is_(joincond.direction, MANYTOONE)
+
+
+class DetermineJoinTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def test_determine_join_o2m(self):
+ joincond = self._join_fixture_o2m()
self.assert_compile(
- pj, "pj.id = selfref.sid"
+ joincond.primaryjoin,
+ "lft.id = rgt.lid"
)
- right = select([joincond.child_selectable]).alias('pj')
- pj, sj, sec, adapter = joincond.join_targets(
- joincond.parent_selectable,
- right,
- True)
+ def test_determine_join_o2m_selfref(self):
+ joincond = self._join_fixture_o2m_selfref()
self.assert_compile(
- pj, "selfref.id = pj.sid"
+ joincond.primaryjoin,
+ "selfref.id = selfref.sid"
)
def test_determine_join_m2o_selfref(self):
@@ -239,17 +322,6 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
"selfref.id = selfref.sid"
)
- def test_determine_direction_m2o_selfref(self):
- joincond = self._join_fixture_m2o_selfref()
- is_(joincond.direction, MANYTOONE)
-
- def test_determine_remote_columns_m2o_selfref(self):
- joincond = self._join_fixture_m2o_selfref()
- eq_(
- joincond.remote_columns,
- set([self.selfref.c.id])
- )
-
def test_determine_join_o2m_composite_selfref(self):
joincond = self._join_fixture_o2m_composite_selfref()
self.assert_compile(
@@ -258,18 +330,6 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
"AND composite_selfref.id = composite_selfref.parent_id"
)
- def test_determine_direction_o2m_composite_selfref(self):
- joincond = self._join_fixture_o2m_composite_selfref()
- is_(joincond.direction, ONETOMANY)
-
- def test_determine_remote_columns_o2m_composite_selfref(self):
- joincond = self._join_fixture_o2m_composite_selfref()
- eq_(
- joincond.remote_columns,
- set([self.composite_selfref.c.parent_id,
- self.composite_selfref.c.group_id])
- )
-
def test_determine_join_m2o_composite_selfref(self):
joincond = self._join_fixture_m2o_composite_selfref()
self.assert_compile(
@@ -278,17 +338,7 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
"AND composite_selfref.id = composite_selfref.parent_id"
)
- def test_determine_direction_m2o_composite_selfref(self):
- joincond = self._join_fixture_m2o_composite_selfref()
- is_(joincond.direction, MANYTOONE)
- def test_determine_remote_columns_m2o_composite_selfref(self):
- joincond = self._join_fixture_m2o_composite_selfref()
- eq_(
- joincond.remote_columns,
- set([self.composite_selfref.c.id,
- self.composite_selfref.c.group_id])
- )
def test_determine_join_m2o(self):
joincond = self._join_fixture_m2o()
@@ -297,24 +347,30 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
"lft.id = rgt.lid"
)
- def test_determine_direction_m2o(self):
- joincond = self._join_fixture_m2o()
- is_(joincond.direction, MANYTOONE)
+class AdaptedJoinTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
- def test_determine_remote_columns_m2o(self):
- joincond = self._join_fixture_m2o()
- eq_(
- joincond.remote_columns,
- set([self.left.c.id])
+ def test_join_targets_o2m_selfref(self):
+ joincond = self._join_fixture_o2m_selfref()
+ left = select([joincond.parent_selectable]).alias('pj')
+ pj, sj, sec, adapter = joincond.join_targets(
+ left,
+ joincond.child_selectable,
+ True)
+ self.assert_compile(
+ pj, "pj.id = selfref.sid"
)
- def test_determine_local_remote_pairs_o2m(self):
- joincond = self._join_fixture_o2m()
- eq_(
- joincond.local_remote_pairs,
- [(self.left.c.id, self.right.c.lid)]
+ right = select([joincond.child_selectable]).alias('pj')
+ pj, sj, sec, adapter = joincond.join_targets(
+ joincond.parent_selectable,
+ right,
+ True)
+ self.assert_compile(
+ pj, "selfref.id = pj.sid"
)
+
def test_join_targets_o2m_plain(self):
joincond = self._join_fixture_o2m()
pj, sj, sec, adapter = joincond.join_targets(
@@ -347,6 +403,8 @@ class JoinCondTest(fixtures.TestBase, AssertsCompiledSQL):
pj, "lft.id = pj.lid"
)
+class LazyClauseTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+
def _test_lazy_clause_o2m(self):
joincond = self._join_fixture_o2m()
self.assert_compile(
diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index 0a02cbf9a..d2dcbe312 100644
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -7,8 +7,9 @@ from test.lib.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, relation, \
backref, create_session, configure_mappers, \
clear_mappers, sessionmaker, attributes,\
- Session, composite, column_property
-from test.lib.testing import eq_, startswith_, AssertsCompiledSQL
+ Session, composite, column_property, foreign
+from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE, MANYTOMANY
+from test.lib.testing import eq_, startswith_, AssertsCompiledSQL, is_
from test.lib import fixtures
from test.orm import _fixtures
@@ -141,12 +142,12 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
Table('company_t', metadata,
Column('company_id', Integer, primary_key=True,
test_needs_autoincrement=True),
- Column('name', sa.Unicode(30)))
+ Column('name', String(30)))
Table('employee_t', metadata,
Column('company_id', Integer, primary_key=True),
Column('emp_id', Integer, primary_key=True),
- Column('name', sa.Unicode(30)),
+ Column('name', String(30)),
Column('reports_to_id', Integer),
sa.ForeignKeyConstraint(
['company_id'],
@@ -158,7 +159,8 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
@classmethod
def setup_classes(cls):
class Company(cls.Basic):
- pass
+ def __init__(self, name):
+ self.name = name
class Employee(cls.Basic):
def __init__(self, name, company, emp_id, reports_to=None):
@@ -248,11 +250,16 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
self._test()
def _test(self):
+ sess = Session()
+ self._setup_data(sess)
+ self._test_lazy_relations(sess)
+ self._test_join_aliasing(sess)
+
+ def _setup_data(self, sess):
Employee, Company = self.classes.Employee, self.classes.Company
- sess = create_session()
- c1 = Company()
- c2 = Company()
+ c1 = Company('c1')
+ c2 = Company('c2')
e1 = Employee(u'emp1', c1, 1)
e2 = Employee(u'emp2', c1, 2, e1)
@@ -263,10 +270,17 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
e7 = Employee(u'emp7', c2, 3, e5)
sess.add_all((c1, c2))
- sess.flush()
- sess.expunge_all()
+ sess.commit()
+ sess.close()
+
+ def _test_lazy_relations(self, sess):
+ Employee, Company = self.classes.Employee, self.classes.Company
+
+ c1 = sess.query(Company).filter_by(name='c1').one()
+ c2 = sess.query(Company).filter_by(name='c2').one()
+ e1 = sess.query(Employee).filter_by(name='emp1').one()
+ e5 = sess.query(Employee).filter_by(name='emp5').one()
- test_c1 = sess.query(Company).get(c1.company_id)
test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
assert test_e1.name == 'emp1', test_e1.name
test_e5 = sess.query(Employee).get([c2.company_id, e5.emp_id])
@@ -277,6 +291,16 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
assert sess.query(Employee).\
get([c2.company_id, 3]).reports_to.name == 'emp5'
+ def _test_join_aliasing(self, sess):
+ Employee, Company = self.classes.Employee, self.classes.Company
+ eq_(
+ [n for n, in sess.query(Employee.name).\
+ join(Employee.reports_to, aliased=True).\
+ filter_by(name='emp5').\
+ reset_joinpoint().\
+ order_by(Employee.name)],
+ ['emp6', 'emp7']
+ )
class CompositeJoinPartialFK(fixtures.MappedTest, AssertsCompiledSQL):
__dialect__ = 'default'
@@ -839,7 +863,6 @@ class AmbiguousJoinInterpretedAsSelfRef(fixtures.MappedTest):
def test_mapping(self):
Subscriber, Address = self.classes.Subscriber, self.classes.Address
- from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE
sess = create_session()
assert Subscriber.addresses.property.direction is ONETOMANY
assert Address.customer.property.direction is MANYTOONE
@@ -1733,21 +1756,45 @@ class ExplicitLocalRemoteTest(fixtures.MappedTest):
class T2(cls.Comparable):
pass
- def test_onetomany_funcfk(self):
+ def test_onetomany_funcfk_oldstyle(self):
T2, T1, t2, t1 = (self.classes.T2,
self.classes.T1,
self.tables.t2,
self.tables.t1)
- # use a function within join condition. but specifying
- # local_remote_pairs overrides all parsing of the join condition.
+ # old _local_remote_pairs
mapper(T1, t1, properties={
't2s':relationship(T2,
primaryjoin=t1.c.id==sa.func.lower(t2.c.t1id),
_local_remote_pairs=[(t1.c.id, t2.c.t1id)],
- foreign_keys=[t2.c.t1id])})
+ foreign_keys=[t2.c.t1id]
+ )
+ })
+ mapper(T2, t2)
+ self._test_onetomany()
+
+ def test_onetomany_funcfk_annotated(self):
+ T2, T1, t2, t1 = (self.classes.T2,
+ self.classes.T1,
+ self.tables.t2,
+ self.tables.t1)
+
+ # use annotation
+ mapper(T1, t1, properties={
+ 't2s':relationship(T2,
+ primaryjoin=t1.c.id==
+ foreign(sa.func.lower(t2.c.t1id)),
+ )})
mapper(T2, t2)
+ self._test_onetomany()
+ def _test_onetomany(self):
+ T2, T1, t2, t1 = (self.classes.T2,
+ self.classes.T1,
+ self.tables.t2,
+ self.tables.t1)
+ is_(T1.t2s.property.direction, ONETOMANY)
+ eq_(T1.t2s.property.local_remote_pairs, [(t1.c.id, t2.c.t1id)])
sess = create_session()
a1 = T1(id='number1', data='a1')
a2 = T1(id='number2', data='a2')
diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py
index f9333dbf5..d4f324dd7 100644
--- a/test/sql/test_generative.py
+++ b/test/sql/test_generative.py
@@ -1,5 +1,5 @@
from sqlalchemy import *
-from sqlalchemy.sql import table, column, ClauseElement
+from sqlalchemy.sql import table, column, ClauseElement, operators
from sqlalchemy.sql.expression import _clone, _from_objects
from test.lib import *
from sqlalchemy.sql.visitors import *
@@ -166,6 +166,90 @@ class TraversalTest(fixtures.TestBase, AssertsExecutionResults):
s = set(ClauseVisitor().iterate(bin))
assert set(ClauseVisitor().iterate(bin)) == set([foo, bar, bin])
+class BinaryEndpointTraversalTest(fixtures.TestBase):
+ """test the special binary product visit"""
+
+ def _assert_traversal(self, expr, expected):
+ canary = []
+ def visit(binary, l, r):
+ canary.append((binary.operator, l, r))
+ print binary.operator, l, r
+ sql_util.visit_binary_product(visit, expr)
+ eq_(
+ canary, expected
+ )
+
+ def test_basic(self):
+ a, b = column("a"), column("b")
+ self._assert_traversal(
+ a == b,
+ [
+ (operators.eq, a, b)
+ ]
+ )
+
+ def test_with_tuples(self):
+ a, b, c, d, b1, b1a, b1b, e, f = (
+ column("a"),
+ column("b"),
+ column("c"),
+ column("d"),
+ column("b1"),
+ column("b1a"),
+ column("b1b"),
+ column("e"),
+ column("f")
+ )
+ expr = tuple_(
+ a, b, b1==tuple_(b1a, b1b == d), c
+ ) > tuple_(
+ func.go(e + f)
+ )
+ self._assert_traversal(
+ expr,
+ [
+ (operators.gt, a, e),
+ (operators.gt, a, f),
+ (operators.gt, b, e),
+ (operators.gt, b, f),
+ (operators.eq, b1, b1a),
+ (operators.eq, b1b, d),
+ (operators.gt, c, e),
+ (operators.gt, c, f)
+ ]
+ )
+
+ def test_composed(self):
+ a, b, e, f, q, j, r = (
+ column("a"),
+ column("b"),
+ column("e"),
+ column("f"),
+ column("q"),
+ column("j"),
+ column("r"),
+ )
+ expr = and_(
+ (a + b) == q + func.sum(e + f),
+ and_(
+ j == r,
+ f == q
+ )
+ )
+ self._assert_traversal(
+ expr,
+ [
+ (operators.eq, a, q),
+ (operators.eq, a, e),
+ (operators.eq, a, f),
+ (operators.eq, b, q),
+ (operators.eq, b, e),
+ (operators.eq, b, f),
+ (operators.eq, j, r),
+ (operators.eq, f, q),
+ ]
+ )
+
class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
"""test copy-in-place behavior of various ClauseElements."""
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 6d85f7c4f..4f1f39014 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -1151,5 +1151,7 @@ class AnnotationsTest(fixtures.TestBase):
assert b2.left is not bin.left
assert b3.left is not b2.left is not bin.left
assert b4.left is bin.left # since column is immutable
- assert b4.right is not bin.right is not b2.right is not b3.right
+ assert b4.right is bin.right
+ assert b2.right is not bin.right
+ assert b3.right is b4.right is bin.right
View Lorry Controller Status