Method test_pickled_rows

test/sql/test_resultset.py:513–568

(self, connection, use_pickle, use_labels)

    @testing.variation("use_pickle", [True, False])
    @testing.variation("use_labels", [True, False])
    def test_pickled_rows(self, connection, use_pickle, use_labels):
        users = self.tables.users
        addresses = self.tables.addresses

        result = self._pickle_row_data(connection, use_labels)

        if use_pickle:
            result = pickle.loads(pickle.dumps(result))

        eq_(result, [(7, "jack"), (8, "ed"), (9, "fred")])
        if use_labels:
            eq_(result[0]._mapping["users_user_id"], 7)
            eq_(
                list(result[0]._fields),
                ["users_user_id", "users_user_name"],
            )
        else:
            eq_(result[0]._mapping["user_id"], 7)
            eq_(list(result[0]._fields), ["user_id", "user_name"])

        eq_(result[0][0], 7)

        assert_raises(
            exc.NoSuchColumnError,
            lambda: result[0]._mapping["fake key"],
        )

        # previously would warn

        if use_pickle:
            with expect_raises_message(
                exc.NoSuchColumnError,
                "Row was unpickled; lookup by ColumnElement is unsupported",
            ):
                result[0]._mapping[users.c.user_id]
        else:
            eq_(result[0]._mapping[users.c.user_id], 7)

        if use_pickle:
            with expect_raises_message(
                exc.NoSuchColumnError,
                "Row was unpickled; lookup by ColumnElement is unsupported",
            ):
                result[0]._mapping[users.c.user_name]
        else:
            eq_(result[0]._mapping[users.c.user_name], "jack")

        assert_raises(
            exc.NoSuchColumnError,
            lambda: result[0]._mapping[addresses.c.user_id],
        )

        assert_raises(
            exc.NoSuchColumnError,
            lambda: result[0]._mapping[addresses.c.address_id],
        )

    @testing.variation("use_labels", [True, False])