Method test_bulk_insert_heterogeneous

test/orm/test_composites.py:378–451

(self, type_)

    @testing.combinations("legacy", "statement")
    def test_bulk_insert_heterogeneous(self, type_):
        Edge, Point = (self.classes.Edge, self.classes.Point)
        Graph = self.classes.Graph

        sess = self._fixture()

        graph = Graph(id=2)
        sess.add(graph)
        sess.flush()
        graph_id = 2

        d1 = [
            {
                "start": Point(random.randint(1, 50), random.randint(1, 50)),
                "end": Point(random.randint(1, 50), random.randint(1, 50)),
                "graph_id": graph_id,
            }
            for i in range(3)
        ]
        d2 = [
            {
                "start": Point(random.randint(1, 50), random.randint(1, 50)),
                "graph_id": graph_id,
            }
            for i in range(2)
        ]
        d3 = [
            {
                "x2": random.randint(1, 50),
                "y2": random.randint(1, 50),
                "graph_id": graph_id,
            }
            for i in range(2)
        ]
        data = d1 + d2 + d3
        random.shuffle(data)

        assert_data = [
            {
                "start": d["start"] if "start" in d else None,
                "end": (
                    d["end"]
                    if "end" in d
                    else Point(d["x2"], d["y2"]) if "x2" in d else None
                ),
                "graph_id": d["graph_id"],
            }
            for d in data
        ]

        if type_ == "statement":
            sess.execute(insert(Edge), data)
        elif type_ == "legacy":
            sess.bulk_insert_mappings(Edge, data)
        else:
            assert False

        edges = self.tables.edges