From d5b562fc7efd2214dd66dadd3dbe2c0dd5870ceb Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Mon, 20 Dec 2010 15:18:58 -0500 Subject: [PATCH] - migrate composites to its own test suite, break up tests into individual feature tests --- test/orm/test_composites.py | 337 ++++++++++++++++++++++++++++++++++++ test/orm/test_mapper.py | 293 ------------------------------- 2 files changed, 337 insertions(+), 293 deletions(-) create mode 100644 test/orm/test_composites.py diff --git a/test/orm/test_composites.py b/test/orm/test_composites.py new file mode 100644 index 0000000000..54a1d571a2 --- /dev/null +++ b/test/orm/test_composites.py @@ -0,0 +1,337 @@ +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) + diff --git a/test/orm/test_mapper.py b/test/orm/test_mapper.py index 621e5f47ce..f06173d322 100644 --- a/test/orm/test_mapper.py +++ b/test/orm/test_mapper.py @@ -2155,299 +2155,6 @@ class DeferredPopulationTest(_base.MappedTest): 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): -- 2.47.2