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

Code before:
        A dirty connection is one where we've written commands but haven't
        read all responses. If such a connection is returned to the pool,
        the next caller will read responses from the previous request.
        """
        # Ensure we're using multiple nodes to test the dirty connection scenario
        assert r.keyslot("a") != r.keyslot("b")

        # Mock the write method to raise an error after writing to only some nodes
        orig_write = redis.cluster.NodeCommands.write

        write_count = 0

        def mock_write(self):
            nonlocal write_count
            write_count += 1
            # Allow the first write to succeed
            if write_count == 1:
                return orig_write(self)
            # Simulate a failure after the first write (leaving connection dirty)
            else:
                raise RuntimeError("Simulated write error")

        with patch.object(redis.cluster.NodeCommands, "write", mock_write):
            with pytest.raises(RuntimeError):
                r.pipeline().get("a").get("b").execute()

        # After the error, verify that no connections are in the available pool
        # with dirty state (unread responses). If a connection is dirty, it should
        # have been disconnected before being returned to the pool.
        # We verify this by checking the connections can be reused successfully.
        try:
            # Try to execute a command on each connection to verify
            # they're clean (not holding responses from previous requests)
            result = r.ping()
            assert result is True
        except Exception as e:
            pytest.fail(
                f"Connection reuse after dirty state failed: {e}. "
                f"This indicates a dirty connection was returned to the pool."
            )


@pytest.mark.onlycluster
class TestReadOnlyPipeline:
    """
    Tests for ClusterPipeline class in readonly mode
    """

    def test_pipeline_readonly(self, r):
        """
        On readonly mode, we supports get related stuff only.

Review comment (bug):
The test patches the 'write' method on the NodeCommands class but doesn't verify that connections were actually disconnected. Consider adding an assertion to check that at least one connection has been disconnected (e.g., by verifying n.connection.disconnect() was called) to make the test more explicit about what behavior is being validated.
```suggestion
        # Patch Connection.disconnect so we can assert that at least one
        # connection was disconnected when the write error occurred.
        with patch.object(Connection, "disconnect", wraps=Connection.disconnect) as mock_disconnect:
            with patch.object(redis.cluster.NodeCommands, "write", mock_write):
                with pytest.raises(RuntimeError):
                    r.pipeline().get("a").get("b").execute()

            # Ensure that at least one connection was disconnected as part of
            # handling the dirty connection created by the write failure.
            assert mock_disconnect.called, (
                "Expected at least one connection to be disconnected when "
                "handling a dirty connection, but disconnect() was not called."
            )
```

Code after:
        """
        with r.pipeline(transaction=True) as pipe:
            r["{key}:a"] = 1
            r["{key}:b"] = 2
            pipe.delete("{key}:a", "{key}:b")
            assert pipe.execute()

    def test_unlink_single(self, r):
        """
        Test a single unlink operation
        """
        r["a"] = 1
        with r.pipeline(transaction=False) as pipe:
            pipe.unlink("a")
            assert pipe.execute() == [1]

    def test_multi_unlink_unsupported(self, r):
        """
        Test that multi unlink operation is unsupported
        """
        with r.pipeline(transaction=False) as pipe:
            r["a"] = 1
            r["b"] = 2
            with pytest.raises(RedisClusterException):
                pipe.unlink("a", "b")

    def test_brpoplpush_disabled(self, r):
        """
        Test that brpoplpush is disabled for ClusterPipeline
        """
        with r.pipeline(transaction=False) as pipe:
            with pytest.raises(RedisClusterException):
                pipe.brpoplpush()

    def test_rpoplpush_disabled(self, r):
        """
        Test that rpoplpush is disabled for ClusterPipeline
        """
        with r.pipeline(transaction=False) as pipe:
            with pytest.raises(RedisClusterException):
                pipe.rpoplpush()

    def test_sort_disabled(self, r):
        """
        Test that sort is disabled for ClusterPipeline
        """
        with r.pipeline(transaction=False) as pipe:
            with pytest.raises(RedisClusterException):
                pipe.sort()

    def test_sdiff_disabled(self, r):