--- /dev/null
+from test.lib.testing import assert_raises, assert_raises_message
+import sqlalchemy as sa
+from test.lib import testing
+from sqlalchemy import MetaData, Integer, String, ForeignKey, func, util
+from test.lib.schema import Table, Column
+from sqlalchemy.orm import mapper, relationship, backref, \
+ class_mapper, \
+ validates, aliased
+from sqlalchemy.orm import attributes, \
+ composite, relationship, \
+ Session
+from test.lib.testing import eq_
+from test.orm import _base, _fixtures
+
+
+class PointTest(_base.MappedTest):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('graphs', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('name', String(30)))
+
+ Table('edges', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('graph_id', Integer, ForeignKey('graphs.id'), nullable=False),
+ Column('x1', Integer),
+ Column('y1', Integer),
+ Column('x2', Integer),
+ Column('y2', Integer),
+ )
+
+ @classmethod
+ @testing.resolve_artifact_names
+ def setup_mappers(cls):
+ class Point(_base.BasicEntity):
+ def __init__(self, x, y):
+ self.x = x
+ self.y = y
+ def __composite_values__(self):
+ return [self.x, self.y]
+ __hash__ = None
+ def __eq__(self, other):
+ return isinstance(other, Point) and \
+ other.x == self.x and \
+ other.y == self.y
+ def __ne__(self, other):
+ return not isinstance(other, Point) or not self.__eq__(other)
+
+ class Graph(_base.BasicEntity):
+ pass
+ class Edge(_base.BasicEntity):
+ def __init__(self, start, end):
+ self.start = start
+ self.end = end
+
+ mapper(Graph, graphs, properties={
+ 'edges':relationship(Edge)
+ })
+ mapper(Edge, edges, properties={
+ 'start':sa.orm.composite(Point, edges.c.x1, edges.c.y1),
+ 'end': sa.orm.composite(Point, edges.c.x2, edges.c.y2)
+ })
+
+ @testing.resolve_artifact_names
+ def _fixture(self):
+ sess = Session()
+ g = Graph(id=1, edges=[
+ Edge(Point(3, 4), Point(5, 6)),
+ Edge(Point(14, 5), Point(2, 7))
+ ])
+ sess.add(g)
+ sess.commit()
+ return sess
+
+ @testing.resolve_artifact_names
+ def test_round_trip(self):
+
+ sess = self._fixture()
+
+ g1 = sess.query(Graph).first()
+ sess.close()
+
+ g = sess.query(Graph).get(g1.id)
+ eq_(
+ [(e.start, e.end) for e in g.edges],
+ [
+ (Point(3, 4), Point(5, 6)),
+ (Point(14, 5), Point(2, 7)),
+ ]
+ )
+
+ @testing.resolve_artifact_names
+ def test_detect_change(self):
+ sess = self._fixture()
+
+ g = sess.query(Graph).first()
+ g.edges[1].end = Point(18, 4)
+ sess.commit()
+
+ e = sess.query(Edge).get(g.edges[1].id)
+ eq_(e.end, Point(18, 4))
+
+ @testing.resolve_artifact_names
+ def test_eager_load(self):
+ sess = self._fixture()
+
+ g = sess.query(Graph).first()
+ sess.close()
+
+ def go():
+ g2 = sess.query(Graph).\
+ options(sa.orm.joinedload('edges')).\
+ get(g.id)
+
+ eq_(
+ [(e.start, e.end) for e in g2.edges],
+ [
+ (Point(3, 4), Point(5, 6)),
+ (Point(14, 5), Point(2, 7)),
+ ]
+ )
+ self.assert_sql_count(testing.db, go, 1)
+
+ @testing.resolve_artifact_names
+ def test_comparator(self):
+ sess = self._fixture()
+
+ g = sess.query(Graph).first()
+
+ assert sess.query(Edge).\
+ filter(Edge.start==Point(3, 4)).one() is \
+ g.edges[0]
+
+ assert sess.query(Edge).\
+ filter(Edge.start!=Point(3, 4)).first() is \
+ g.edges[1]
+
+ eq_(
+ sess.query(Edge).filter(Edge.start==None).all(),
+ []
+ )
+
+ @testing.resolve_artifact_names
+ def test_query_cols(self):
+ sess = self._fixture()
+
+ eq_(
+ sess.query(Edge.start, Edge.end).all(),
+ [(3, 4, 5, 6), (14, 5, 2, 7)]
+ )
+
+ @testing.resolve_artifact_names
+ def test_delete(self):
+ sess = self._fixture()
+ g = sess.query(Graph).first()
+
+ e = g.edges[1]
+ del e.end
+ sess.flush()
+ eq_(
+ sess.query(Edge.start, Edge.end).all(),
+ [(3, 4, 5, 6), (14, 5, None, None)]
+ )
+
+ @testing.resolve_artifact_names
+ def test_save_null(self):
+ """test saving a null composite value
+
+ See google groups thread for more context:
+ http://groups.google.com/group/sqlalchemy/browse_thread/thread/0c6580a1761b2c29
+
+ """
+ sess = Session()
+ g = Graph(id=1)
+ e = Edge(None, None)
+ g.edges.append(e)
+
+ sess.add(g)
+ sess.commit()
+
+ g2 = sess.query(Graph).get(1)
+ assert g2.edges[-1].start.x is None
+ assert g2.edges[-1].start.y is None
+
+class PrimaryKeyTest(_base.MappedTest):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('graphs', metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('version_id', Integer, primary_key=True, nullable=True),
+ Column('name', String(30)))
+
+ @classmethod
+ @testing.resolve_artifact_names
+ def setup_mappers(cls):
+ class Version(_base.BasicEntity):
+ def __init__(self, id, version):
+ self.id = id
+ self.version = version
+ def __composite_values__(self):
+ return (self.id, self.version)
+ __hash__ = None
+ def __eq__(self, other):
+ return isinstance(other, Version) and other.id == self.id and \
+ other.version == self.version
+ def __ne__(self, other):
+ return not self.__eq__(other)
+
+ class Graph(_base.BasicEntity):
+ def __init__(self, version):
+ self.version = version
+
+ mapper(Graph, graphs, properties={
+ 'version':sa.orm.composite(Version, graphs.c.id,
+ graphs.c.version_id)})
+
+
+ @testing.resolve_artifact_names
+ def _fixture(self):
+ sess = Session()
+ g = Graph(Version(1, 1))
+ sess.add(g)
+ sess.commit()
+ return sess
+
+ @testing.resolve_artifact_names
+ def test_get_by_col(self):
+
+ sess = self._fixture()
+ g = sess.query(Graph).first()
+
+ g2 = sess.query(Graph).get([g.id, g.version_id])
+ eq_(g.version, g2.version)
+
+ @testing.resolve_artifact_names
+ def test_get_by_composite(self):
+ sess = self._fixture()
+ g = sess.query(Graph).first()
+
+ g2 = sess.query(Graph).get(Version(g.id, g.version_id))
+ eq_(g.version, g2.version)
+
+ @testing.fails_on('mssql', 'Cannot update identity columns.')
+ @testing.resolve_artifact_names
+ def test_pk_mutation(self):
+ sess = self._fixture()
+
+ g = sess.query(Graph).first()
+
+ g.version = Version(2, 1)
+ sess.commit()
+ g2 = sess.query(Graph).get(Version(2, 1))
+ eq_(g.version, g2.version)
+
+ @testing.fails_on_everything_except("sqlite")
+ @testing.resolve_artifact_names
+ def test_null_pk(self):
+ sess = Session()
+
+ # test pk with one column NULL
+ # only sqlite can really handle this
+ g = Graph(Version(2, None))
+ sess.add(g)
+ sess.commit()
+ g2 = sess.query(Graph).filter_by(version=Version(2, None)).one()
+ eq_(g.version, g2.version)
+
+class DefaultsTest(_base.MappedTest):
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('foobars', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('x1', Integer, default=2),
+ Column('x2', Integer),
+ Column('x3', Integer, default=15),
+ Column('x4', Integer)
+ )
+
+ @classmethod
+ @testing.resolve_artifact_names
+ def setup_mappers(cls):
+ class Foobar(_base.BasicEntity):
+ pass
+
+ class FBComposite(_base.BasicEntity):
+ def __init__(self, x1, x2, x3, x4):
+ self.x1 = x1
+ self.x2 = x2
+ self.x3 = x3
+ self.x4 = x4
+ def __composite_values__(self):
+ return self.x1, self.x2, self.x3, self.x4
+ __hash__ = None
+ def __eq__(self, other):
+ return other.x1 == self.x1 and \
+ other.x2 == self.x2 and \
+ other.x3 == self.x3 and \
+ other.x4 == self.x4
+ def __ne__(self, other):
+ return not self.__eq__(other)
+
+ mapper(Foobar, foobars, properties=dict(
+ foob=sa.orm.composite(FBComposite,
+ foobars.c.x1,
+ foobars.c.x2,
+ foobars.c.x3,
+ foobars.c.x4)
+ ))
+
+ @testing.resolve_artifact_names
+ def test_attributes_with_defaults(self):
+
+ sess = Session()
+ f1 = Foobar()
+ f1.foob = FBComposite(None, 5, None, None)
+ sess.add(f1)
+ sess.flush()
+
+ assert f1.foob == FBComposite(2, 5, 15, None)
+
+ f2 = Foobar()
+ sess.add(f2)
+ sess.flush()
+ assert f2.foob == FBComposite(2, None, 15, None)
+
+ @testing.resolve_artifact_names
+ def test_set_composite_values(self):
+ sess = Session()
+ f1 = Foobar()
+ f1.foob = FBComposite(None, 5, None, None)
+ sess.add(f1)
+ sess.flush()
+
+ assert f1.foob == FBComposite(2, 5, 15, None)
+
self._test(thing)
-class CompositeTypesTest(_base.MappedTest):
-
- @classmethod
- def define_tables(cls, metadata):
- Table('graphs', metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('version_id', Integer, primary_key=True, nullable=True),
- Column('name', String(30)))
-
- Table('edges', metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('graph_id', Integer, nullable=False),
- Column('graph_version_id', Integer, nullable=False),
- Column('x1', Integer),
- Column('y1', Integer),
- Column('x2', Integer),
- Column('y2', Integer),
- sa.ForeignKeyConstraint(
- ['graph_id', 'graph_version_id'],
- ['graphs.id', 'graphs.version_id']))
-
- Table('foobars', metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('x1', Integer, default=2),
- Column('x2', Integer),
- Column('x3', Integer, default=15),
- Column('x4', Integer)
- )
-
- @testing.resolve_artifact_names
- def test_basic(self):
- class Point(object):
- def __init__(self, x, y):
- self.x = x
- self.y = y
- def __composite_values__(self):
- return [self.x, self.y]
- __hash__ = None
- def __eq__(self, other):
- return isinstance(other, Point) and other.x == self.x and other.y == self.y
- def __ne__(self, other):
- return not isinstance(other, Point) or not self.__eq__(other)
-
- class Graph(object):
- pass
- class Edge(object):
- def __init__(self, start, end):
- self.start = start
- self.end = end
-
- mapper(Graph, graphs, properties={
- 'edges':relationship(Edge)
- })
- mapper(Edge, edges, properties={
- 'start':sa.orm.composite(Point, edges.c.x1, edges.c.y1),
- 'end': sa.orm.composite(Point, edges.c.x2, edges.c.y2)
- })
-
- sess = Session()
- g = Graph()
- g.id = 1
- g.version_id=1
- g.edges.append(Edge(Point(3, 4), Point(5, 6)))
- g.edges.append(Edge(Point(14, 5), Point(2, 7)))
- sess.add(g)
- sess.commit()
-
- g2 = sess.query(Graph).get([g.id, g.version_id])
- for e1, e2 in zip(g.edges, g2.edges):
- eq_(e1.start, e2.start)
- eq_(e1.end, e2.end)
-
- g2.edges[1].end = Point(18, 4)
- sess.commit()
-
- e = sess.query(Edge).get(g2.edges[1].id)
- eq_(e.end, Point(18, 4))
-
- e.end = Point(19, 5)
- sess.commit()
- g.id, g.version_id, g.edges
- sess.expunge_all()
-
- def go():
- g2 = sess.query(Graph).\
- options(sa.orm.joinedload('edges')).get([g.id, g.version_id])
- for e1, e2 in zip(g.edges, g2.edges):
- eq_(e1.start, e2.start)
- eq_(e1.end, e2.end)
- self.assert_sql_count(testing.db, go, 1)
-
- # test comparison of CompositeProperties to their object instances
- g = sess.query(Graph).get([1, 1])
- assert sess.query(Edge).filter(Edge.start==Point(3, 4)).one() is g.edges[0]
-
- assert sess.query(Edge).filter(Edge.start!=Point(3, 4)).first() is g.edges[1]
-
- eq_(sess.query(Edge).filter(Edge.start==None).all(), [])
-
- # query by columns
- eq_(sess.query(Edge.start, Edge.end).all(), [(3, 4, 5, 6), (14, 5, 19, 5)])
-
- e = g.edges[1]
- del e.end
- sess.flush()
- eq_(sess.query(Edge.start, Edge.end).all(), [(3, 4, 5, 6), (14, 5, None, None)])
-
-
- @testing.resolve_artifact_names
- def test_pk(self):
- """Using a composite type as a primary key"""
-
- class Version(object):
- def __init__(self, id, version):
- self.id = id
- self.version = version
- def __composite_values__(self):
- return (self.id, self.version)
- __hash__ = None
- def __eq__(self, other):
- return isinstance(other, Version) and other.id == self.id and \
- other.version == self.version
- def __ne__(self, other):
- return not self.__eq__(other)
-
- class Graph(object):
- def __init__(self, version):
- self.version = version
-
- mapper(Graph, graphs, properties={
- 'version':sa.orm.composite(Version, graphs.c.id,
- graphs.c.version_id)})
-
- sess = create_session()
- g = Graph(Version(1, 1))
- sess.add(g)
- sess.flush()
-
- sess.expunge_all()
- g2 = sess.query(Graph).get([1, 1])
- eq_(g.version, g2.version)
- sess.expunge_all()
-
- g2 = sess.query(Graph).get(Version(1, 1))
- eq_(g.version, g2.version)
-
- # test pk mutation
- @testing.fails_on('mssql', 'Cannot update identity columns.')
- def update_pk():
- g2.version = Version(2, 1)
- sess.flush()
- g3 = sess.query(Graph).get(Version(2, 1))
- eq_(g2.version, g3.version)
- update_pk()
-
- # test pk with one column NULL
- # TODO: can't seem to get NULL in for a PK value
- # in either mysql or postgresql, autoincrement=False etc.
- # notwithstanding
- @testing.fails_on_everything_except("sqlite")
- def go():
- g = Graph(Version(2, None))
- sess.add(g)
- sess.flush()
- sess.expunge_all()
- g2 = sess.query(Graph).filter_by(version=Version(2, None)).one()
- eq_(g.version, g2.version)
- go()
-
- @testing.resolve_artifact_names
- def test_attributes_with_defaults(self):
- class Foobar(object):
- pass
-
- class FBComposite(object):
- def __init__(self, x1, x2, x3, x4):
- self.x1 = x1
- self.x2 = x2
- self.x3 = x3
- self.x4 = x4
- def __composite_values__(self):
- return self.x1, self.x2, self.x3, self.x4
- __hash__ = None
- def __eq__(self, other):
- return other.x1 == self.x1 and other.x2 == self.x2 and other.x3 == self.x3 and other.x4 == self.x4
- def __ne__(self, other):
- return not self.__eq__(other)
-
- mapper(Foobar, foobars, properties=dict(
- foob=sa.orm.composite(FBComposite, foobars.c.x1, foobars.c.x2, foobars.c.x3, foobars.c.x4)
- ))
-
- sess = create_session()
- f1 = Foobar()
- f1.foob = FBComposite(None, 5, None, None)
- sess.add(f1)
- sess.flush()
-
- assert f1.foob == FBComposite(2, 5, 15, None)
-
-
- f2 = Foobar()
- sess.add(f2)
- sess.flush()
- assert f2.foob == FBComposite(2, None, 15, None)
-
-
- @testing.resolve_artifact_names
- def test_set_composite_values(self):
- class Foobar(object):
- pass
-
- class FBComposite(object):
- def __init__(self, x1, x2, x3, x4):
- self.x1val = x1
- self.x2val = x2
- self.x3 = x3
- self.x4 = x4
- def __composite_values__(self):
- return self.x1val, self.x2val, self.x3, self.x4
- def __set_composite_values__(self, x1, x2, x3, x4):
- self.x1val = x1
- self.x2val = x2
- self.x3 = x3
- self.x4 = x4
- __hash__ = None
- def __eq__(self, other):
- return other.x1val == self.x1val and other.x2val == self.x2val and other.x3 == self.x3 and other.x4 == self.x4
- def __ne__(self, other):
- return not self.__eq__(other)
-
- mapper(Foobar, foobars, properties=dict(
- foob=sa.orm.composite(FBComposite, foobars.c.x1, foobars.c.x2, foobars.c.x3, foobars.c.x4)
- ))
-
- sess = create_session()
- f1 = Foobar()
- f1.foob = FBComposite(None, 5, None, None)
- sess.add(f1)
- sess.flush()
-
- assert f1.foob == FBComposite(2, 5, 15, None)
-
- @testing.resolve_artifact_names
- def test_save_null(self):
- """test saving a null composite value
-
- See google groups thread for more context:
- http://groups.google.com/group/sqlalchemy/browse_thread/thread/0c6580a1761b2c29
-
- """
- class Point(object):
- def __init__(self, x, y):
- self.x = x
- self.y = y
- def __composite_values__(self):
- return [self.x, self.y]
- __hash__ = None
- def __eq__(self, other):
- return other.x == self.x and other.y == self.y
- def __ne__(self, other):
- return not self.__eq__(other)
-
- class Graph(object):
- pass
- class Edge(object):
- def __init__(self, start, end):
- self.start = start
- self.end = end
-
- mapper(Graph, graphs, properties={
- 'edges':relationship(Edge)
- })
- mapper(Edge, edges, properties={
- 'start':sa.orm.composite(Point, edges.c.x1, edges.c.y1),
- 'end':sa.orm.composite(Point, edges.c.x2, edges.c.y2)
- })
-
- sess = create_session()
- g = Graph()
- g.id = 1
- g.version_id=1
- e = Edge(None, None)
- g.edges.append(e)
-
- sess.add(g)
- sess.flush()
-
- sess.expunge_all()
-
- g2 = sess.query(Graph).get([1, 1])
- assert g2.edges[-1].start.x is None
- assert g2.edges[-1].start.y is None
class NoLoadTest(_fixtures.FixtureTest):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
