Repository: redis/redis-py
File: tests/test_asyncio/test_commands.py (Python)

Code before:
        assert (await r.config_get())["timeout"] == "70"
        assert await r.config_set("timeout", 0)
        assert (await r.config_get())["timeout"] == "0"

    @pytest.mark.redismod
    @skip_if_server_version_lt("7.9.0")
    async def test_config_get_for_modules(self, r: redis.Redis):
        search_module_configs = await r.config_get("search-*")
        assert "search-timeout" in search_module_configs

        ts_module_configs = await r.config_get("ts-*")
        assert "ts-num-threads" in ts_module_configs

        bf_module_configs = await r.config_get("bf-*")
        assert "bf-initial-size" in bf_module_configs

        cf_module_configs = await r.config_get("cf-*")
        assert "cf-max-iterations" in cf_module_configs

    @pytest.mark.redismod
    @skip_if_server_version_lt("7.9.0")
    async def test_config_set_for_search_module(self, r: redis.Redis):
        search_timeout_initial = (await r.config_get())["search-timeout"]
        search_timeout_new = int(search_timeout_initial) + 100

        assert await r.config_set("search-timeout", search_timeout_new)
        assert (
            int((await r.config_get("search-*"))["search-timeout"])
            == search_timeout_new
        )
        assert (
            int((await r.ft().config_get("TIMEOUT"))[b"TIMEOUT"]) == search_timeout_new
        )
        with pytest.raises(exceptions.ResponseError):
            await r.config_set("search-max-doctablesize", 2000000)

    @pytest.mark.onlynoncluster
    async def test_dbsize(self, r: redis.Redis):
        await r.set("a", "foo")
        await r.set("b", "bar")
        assert await r.dbsize() == 2

    @pytest.mark.onlynoncluster
    async def test_echo(self, r: redis.Redis):
        assert await r.echo("foo bar") == b"foo bar"

    @pytest.mark.onlynoncluster
    async def test_info(self, r: redis.Redis):
        await r.set("a", "foo")
        await r.set("b", "bar")
        info = await r.info()

Review comment (suggestion):
Let's deprecate following commands (FT.CONFIG SET, FT.CONFIG GET) as they will be replaced by top level CONFIG since 8.0. We have a custom deprecation annotation `@deprecate_function` for this purpose

Code after:
        assert (await r.config_get())["timeout"] == "70"
        assert await r.config_set("timeout", 0)
        assert (await r.config_get())["timeout"] == "0"

    @pytest.mark.redismod
    @skip_if_server_version_lt("7.9.0")
    async def test_config_get_for_modules(self, r: redis.Redis):
        search_module_configs = await r.config_get("search-*")
        assert "search-timeout" in search_module_configs

        ts_module_configs = await r.config_get("ts-*")
        assert "ts-retention-policy" in ts_module_configs

        bf_module_configs = await r.config_get("bf-*")
        assert "bf-error-rate" in bf_module_configs

        cf_module_configs = await r.config_get("cf-*")
        assert "cf-initial-size" in cf_module_configs

    @pytest.mark.redismod
    @skip_if_server_version_lt("7.9.0")
    async def test_config_set_for_search_module(self, r: redis.Redis):
        config = await r.config_get("*")
        initial_default_search_dialect = config["search-default-dialect"]
        try:
            default_dialect_new = "3"
            assert await r.config_set("search-default-dialect", default_dialect_new)
            assert (await r.config_get("*"))[
                "search-default-dialect"
            ] == default_dialect_new
            assert (
                (await r.ft().config_get("*"))[b"DEFAULT_DIALECT"]
            ).decode() == default_dialect_new
        except AssertionError as ex:
            raise ex
        finally:
            assert await r.config_set(
                "search-default-dialect", initial_default_search_dialect
            )
        with pytest.raises(exceptions.ResponseError):
            await r.config_set("search-max-doctablesize", 2000000)

    @pytest.mark.onlynoncluster
    async def test_dbsize(self, r: redis.Redis):
        await r.set("a", "foo")
        await r.set("b", "bar")
        assert await r.dbsize() == 2

    @pytest.mark.onlynoncluster
    async def test_echo(self, r: redis.Redis):
        assert await r.echo("foo bar") == b"foo bar"