Add unit tests

Author: rohansingh

tests/test_cluster.py

@@ -1692,23 +1692,23 @@ def test_cluster_bitop_not_empty_string(self, r):
 
     @skip_if_server_version_lt("2.6.0")
     def test_cluster_bitop_not(self, r):
-        test_str = b"\xAA\x00\xFF\x55"
+        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"
+        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"
+        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")
@@ -1719,8 +1719,8 @@ def test_cluster_bitop_single_string(self, r):
 
     @skip_if_server_version_lt("2.6.0")
     def test_cluster_bitop_string_operands(self, r):
-        r["{foo}a"] = b"\x01\x02\xFF\xFF"
-        r["{foo}b"] = b"\x01\x02\xFF"
+        r["{foo}a"] = b"\x01\x02\xff\xff"
+        r["{foo}b"] = b"\x01\x02\xff"
         r.bitop("and", "{foo}res1", "{foo}a", "{foo}b")
         r.bitop("or", "{foo}res2", "{foo}a", "{foo}b")
         r.bitop("xor", "{foo}res3", "{foo}a", "{foo}b")
@@ -3260,6 +3260,27 @@ def raise_ask_error():
             assert ask_node.redis_connection.connection.read_response.called
             assert res == ["MOCK_OK"]
 
+    def test_error_is_truncated(self, r):
+        """
+        Test that an error from the pipeline is truncated correctly.
+        """
+        key = "a" * 5000
+
+        pipe = r.pipeline(transaction=False)
+
+        with r.pipeline() as pipe:
+            pipe.set(key, 1)
+            pipe.llen(key)
+            pipe.expire(key, 100)
+
+            with pytest.raises(Exception) as ex:
+                pipe.execute()
+
+            expected = (
+                "Command # 2 (LLEN " + ("a" * 95) + "...) of pipeline caused error: "
+            )
+            assert str(ex.value).startswith(expected)
+
     def test_return_previously_acquired_connections(self, r):
         # in order to ensure that a pipeline will make use of connections
         #   from different nodes

tests/test_pipeline.py

@@ -369,6 +369,22 @@ def test_exec_error_in_no_transaction_pipeline_unicode_command(self, r):
 
         assert r[key] == b"1"
 
+    @pytest.mark.onlynoncluster
+    def test_exec_error_in_pipeline_truncated(self, r):
+        key = "a" * 5000
+        r[key] = 1
+        with r.pipeline() as pipe:
+            pipe.llen(key)
+            pipe.expire(key, 100)
+
+            with pytest.raises(redis.ResponseError) as ex:
+                pipe.execute()
+
+            expected = (
+                "Command # 1 (LLEN " + ("a" * 95) + "...) of pipeline caused error: "
+            )
+            assert str(ex.value).startswith(expected)
+
     def test_pipeline_with_bitfield(self, r):
         with r.pipeline() as pipe:
             pipe.set("a", "1")