Method execute_command

tests/test_asyncio/test_cluster.py:4153–4169

(self, *args, **kwargs)

                ) as execute_command_mock:

                    async def execute_command(self, *args, **kwargs):
                        if args[0] == "INFO":
                            return {"cluster_enabled": True}
                        if args[0] == "CLUSTER SLOTS":
                            return [[0, 16383, [ssl_host, ssl_port, "ssl_node"]]]
                        if args[0] == "COMMAND":
                            return {
                                "ping": {
                                    "name": "ping",
                                    "arity": -1,
                                    "flags": ["stale", "fast"],
                                    "first_key_pos": 0,
                                    "last_key_pos": 0,
                                    "step_count": 0,
                                }
                            }
                        raise NotImplementedError()

                    execute_command_mock.side_effect = execute_command