Method test_user_option_propagation

test/orm/test_events.py:167–246

test #6887. The one case here which would break before the bug was fixed is: use_query_cache=True, loader_opt=selectinload, key_type="fixed" Meaning with a fixed cache key and query caching in place, the user defined loader option would be cached under the same nam

(
        self, key_type, loader_opt, num_opts, use_query_cache
    )

    )
    @testing.combinations((True,), (False,), argnames="use_query_cache")
    def test_user_option_propagation(
        self, key_type, loader_opt, num_opts, use_query_cache
    ):
        """test #6887.

        The one case here which would break before the bug was fixed is:

        use_query_cache=True, loader_opt=selectinload, key_type="fixed"

        Meaning with a fixed cache key and query caching in place, the user
        defined loader option would be cached under the same name each time,
        leading to use of the stale option when using selectinload, as this
        strategy transfers query options from the cached ORM select into the
        one that it generates. no other strategy currently does this.

        """
        User, Address, Dinagling = self.classes("User", "Address", "Dingaling")

        class SetShardOption(UserDefinedOption):
            propagate_to_loaders = True

            def _gen_cache_key(self, anon_map, bindparams):
                if key_type == "fixed":
                    return (None,)
                elif key_type == "payload":
                    return (self.payload,)
                elif key_type == "dont_cache":
                    anon_map[NO_CACHE] = True
                    return None
                else:
                    assert False

        if use_query_cache:
            testing.db._compiled_cache.clear()
            s = fixture_session()
        else:
            s = fixture_session(
                bind=testing.db.execution_options(compiled_cache=None)
            )

        m1 = Mock()

        @event.listens_for(s, "do_orm_execute")
        def go(context):
            for elem in context.user_defined_options:
                if isinstance(elem, SetShardOption):
                    m1.update_execution_options(_sa_shard_id=elem.payload)

        stmt = select(User).options(
            loader_opt(User.addresses).options(loader_opt(Address.dingaling)),
            SetShardOption("some_shard"),
        )
        for u in s.execute(stmt).unique().scalars():
            for a in u.addresses:
                a.dingaling

        s.close()