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

Code before:
            for client in r.client_list(target_nodes=node)
            if client.get("name") in ["redis-py-c1", "redis-py-c2"]
        ]
        assert len(clients) == 2
        clients_by_name = {client.get("name"): client for client in clients}

        client_addr = clients_by_name["redis-py-c2"].get("addr")
        assert r.client_kill(client_addr, target_nodes=node) is True

        clients = [
            client
            for client in r.client_list(target_nodes=node)
            if client.get("name") in ["redis-py-c1", "redis-py-c2"]
        ]
        assert len(clients) == 1
        assert clients[0].get("name") == "redis-py-c1"

    @skip_if_server_version_lt("2.6.0")
    def test_cluster_bitop_not_empty_string(self, r):
        r["{foo}a"] = ""
        r.bitop("not", "{foo}r", "{foo}a")
        assert r.get("{foo}r") is None

    @skip_if_server_version_lt("2.6.0")
    def test_cluster_bitop_not(self, r):
        test_str = b"\xAA\x00\xFF\x55"
        correct = ~0xAA00FF55 & 0xFFFFFFFF
        r["{foo}a"] = test_str
        r.bitop("not", "{foo}r", "{foo}a")
        assert int(binascii.hexlify(r["{foo}r"]), 16) == correct

    @skip_if_server_version_lt("2.6.0")
    def test_cluster_bitop_not_in_place(self, r):
        test_str = b"\xAA\x00\xFF\x55"
        correct = ~0xAA00FF55 & 0xFFFFFFFF
        r["{foo}a"] = test_str
        r.bitop("not", "{foo}a", "{foo}a")
        assert int(binascii.hexlify(r["{foo}a"]), 16) == correct

    @skip_if_server_version_lt("2.6.0")
    def test_cluster_bitop_single_string(self, r):
        test_str = b"\x01\x02\xFF"
        r["{foo}a"] = test_str
        r.bitop("and", "{foo}res1", "{foo}a")
        r.bitop("or", "{foo}res2", "{foo}a")
        r.bitop("xor", "{foo}res3", "{foo}a")
        assert r["{foo}res1"] == test_str
        assert r["{foo}res2"] == test_str
        assert r["{foo}res3"] == test_str

    @skip_if_server_version_lt("2.6.0")

Review comment (suggestion):
```
% grep -R -E '\\x[0-9a-f][a-f]|\\x[a-f][0-9a-f]' tests
Binary file tests/testdata/will_play_text.csv.bz2 matches
tests/test_encoding.py:        r.set("a", b"foo\xff")
tests/test_encoding.py:        r.set("a", b"foo\xff")
tests/test_commands.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_commands.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_commands.py:        test_str = b"\x01\x02\xff"
tests/test_commands.py:        r["a"] = b"\x01\x02\xff\xff"
tests/test_commands.py:        r["b"] = b"\x01\x02\xff"
tests/test_commands.py:        r.set(key, b"\xff\xf0\x00")
tests/test_commands.py:        r.set(key, b"\x00\xff\xf0")
tests/test_commands.py:        r.set(key, b"\xff\xf0\x00")
tests/test_commands.py:        r.set("mykey", b"\x00\xff\xf0")
tests/test_cluster.py:        assert r.keyslot("大奖") == r.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96")
tests/test_cluster.py:        assert r.keyslot("大奖") == r.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96")
tests/test_cluster.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_cluster.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_cluster.py:        test_str = b"\x01\x02\xff"
tests/test_cluster.py:        r["{foo}a"] = b"\x01\x02\xff\xff"
tests/test_cluster.py:        r["{foo}b"] = b"\x01\x02\xff"
Binary file tests/test_asyncio/testdata/will_play_text.csv.bz2 matches
tests/test_asyncio/test_encoding.py:        await r.set("a", b"foo\xff")
tests/test_asyncio/test_encoding.py:        await r.set("a", b"foo\xff")
tests/test_asyncio/test_commands.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_asyncio/test_commands.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_asyncio/test_commands.py:        test_str = b"\x01\x02\xff"
tests/test_asyncio/test_commands.py:        await r.set("a", b"\x01\x02\xff\xff")
tests/test_asyncio/test_commands.py:        await r.set("b", b"\x01\x02\xff")
tests/test_asyncio/test_commands.py:        await r.set(key, b"\xff\xf0\x00")
tests/test_asyncio/test_commands.py:        await r.set(key, b"\x00\xff\xf0")
tests/test_asyncio/test_commands.py:        await r.set(key, b"\xff\xf0\x00")
tests/test_asyncio/test_cluster.py:        assert r.keyslot("大奖") == r.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96")
tests/test_asyncio/test_cluster.py:        assert r.keyslot("大奖") == r.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96")
tests/test_asyncio/test_cluster.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_asyncio/test_cluster.py:        test_str = b"\xaa\x00\xff\x55"
tests/test_asyncio/test_cluster.py:        test_str = b"\x01\x02\xff"
tests/test_asyncio/test_cluster.py:        await r.set("{foo}a", b"\x01\x02\xff\xff")
tests/test_asyncio/test_cluster.py:        await r.set("{foo}b", b"\x01\x02\xff")
tests/test_asyncio/test_scripting.py:        msgpack_message_1 = b"\x81\xa4name\xa3Joe"
tests/test_scripting.py:        return b"\xcf\x84o\xcf\x81\xce\xbdo\xcf\x82"
tests/test_scripting.py:        assert Script(r, script_bytes).script == b"\xcf\x84o\xcf\x81\xce\xbdo\xcf\x82"
tests/test_scripting.py:        msgpack_message_dumped = b"\x81\xa4name\xa3Joe"
tests/test_scripting.py:        msgpack_message_1 = b"\x81\xa4name\xa3Joe"
```

Code after:
        node = r.get_primaries()[0]
        assert r.echo("foo bar", target_nodes=node) == b"foo bar"

    @skip_if_server_version_lt("1.0.0")
    def test_debug_segfault(self, r):
        with pytest.raises(NotImplementedError):
            r.debug_segfault()

    def test_config_resetstat(self, r):
        node = r.get_primaries()[0]
        r.ping(target_nodes=node)
        prior_commands_processed = int(
            r.info(target_nodes=node)["total_commands_processed"]
        )
        assert prior_commands_processed >= 1
        r.config_resetstat(target_nodes=node)
        reset_commands_processed = int(
            r.info(target_nodes=node)["total_commands_processed"]
        )
        assert reset_commands_processed < prior_commands_processed

    @skip_if_server_version_lt("6.2.0")
    def test_client_trackinginfo(self, r):
        node = r.get_primaries()[0]
        res = r.client_trackinginfo(target_nodes=node)
        assert len(res) > 2
        assert "prefixes" in res or b"prefixes" in res

    @skip_if_server_version_lt("2.9.50")
    def test_client_pause(self, r):
        node = r.get_primaries()[0]
        assert r.client_pause(1, target_nodes=node)
        assert r.client_pause(timeout=1, target_nodes=node)
        with pytest.raises(RedisError):
            r.client_pause(timeout="not an integer", target_nodes=node)

    @skip_if_server_version_lt("6.2.0")
    @skip_if_redis_enterprise()
    def test_client_unpause(self, r):
        assert r.client_unpause()

    @skip_if_server_version_lt("5.0.0")
    def test_client_id(self, r):
        node = r.get_primaries()[0]
        assert r.client_id(target_nodes=node) > 0

    @skip_if_server_version_lt("5.0.0")
    def test_client_unblock(self, r):
        node = r.get_primaries()[0]
        myid = r.client_id(target_nodes=node)
        assert not r.client_unblock(myid, target_nodes=node)