Method test_basic_hybrid_search

tests/test_search.py:4221–4259

(self, client)

    @pytest.mark.redismod
    @skip_if_server_version_lt("8.4.0")
    def test_basic_hybrid_search(self, client):
        # Create index and add data
        self._create_hybrid_search_index(client)
        self._add_data_for_hybrid_search(client, items_sets=5)

        # set search query
        search_query = HybridSearchQuery("@color:{red} @color:{green}")

        vsim_query = HybridVsimQuery(
            vector_field_name="@embedding",
            vector_data="$vec",
        )

        hybrid_query = HybridQuery(search_query, vsim_query)

        res = client.ft().hybrid_search(
            query=hybrid_query,
            params_substitution={
                "vec": np.array([-100, -200, -200, -300], dtype=np.float32).tobytes()
            },
        )

        # the default results count limit is 10
        if expects_resp2_shape(client) or expects_unified_shape(client):
            assert res.total_results == 10
            assert len(res.results) == 10
            assert res.warnings == []
            assert res.execution_time > 0
            assert all(isinstance(res.results[i]["__score"], bytes) for i in range(10))
            assert all(isinstance(res.results[i]["__key"], bytes) for i in range(10))
        elif expects_resp3_shape(client):
            assert res["total_results"] == 10
            assert len(res["results"]) == 10
            assert res["warnings"] == []
            assert res["execution_time"] > 0
            assert all(
                isinstance(res["results"][i]["__score"], bytes) for i in range(10)
            )
            assert all(isinstance(res["results"][i]["__key"], bytes) for i in range(10))

    @pytest.mark.redismod
    @skip_if_server_version_lt("8.3.224")