diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index 6bf0034146..eb207270ae 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -112,3 +112,37 @@ async def test_connect_timeout_error_without_retry():
         await conn.connect()
     assert conn._connect.call_count == 1
     assert str(e.value) == "Timeout connecting to server"
+
+
+@pytest.mark.parametrize(
+    "exc_type",
+    [
+        Exception,
+        pytest.param(
+            BaseException,
+            marks=pytest.mark.xfail(
+                reason="https://github.com/redis/redis-py/issues/360",
+            ),
+        ),
+    ],
+)
+async def test_read_response__interrupt_does_not_corrupt(exc_type):
+    conn = Connection()
+
+    await conn.send_command("GET non_existent_key")
+    resp = await conn.read_response()
+    assert resp is None
+
+    with pytest.raises(exc_type):
+        await conn.send_command("EXISTS non_existent_key")
+        # due to the interrupt, the integer '0' result of EXISTS will remain
+        # on the socket's buffer
+        with patch.object(socket.socket, "recv", side_effect=exc_type) as mock_recv:
+            await conn.read_response()
+        mock_recv.assert_called_once()
+
+    await conn.send_command("GET non_existent_key")
+    resp = await conn.read_response()
+    # If working properly, this will get a None.
+    # If not, it will get a zero (the integer result of the previous EXISTS command).
+    assert resp is None
diff --git a/tests/test_connection.py b/tests/test_connection.py
index d9251c31dc..44ef9b93cc 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -122,3 +122,48 @@ def test_connect_timeout_error_without_retry(self):
         assert conn._connect.call_count == 1
         assert str(e.value) == "Timeout connecting to server"
         self.clear(conn)
+
+    @pytest.mark.parametrize(
+        "exc_type",
+        [
+            Exception,
+            pytest.param(
+                BaseException,
+                marks=pytest.mark.xfail(
+                    reason="https://github.com/redis/redis-py/issues/360",
+                ),
+            ),
+        ],
+    )
+    def test_read_response__interrupt_does_not_corrupt(self, exc_type):
+        conn = Connection()
+
+        # A note on BaseException:
+        # While socket.recv is not supposed to raise BaseException, gevent's version
+        # of socket (which, when using gevent + redis-py, one would monkey-patch in)
+        # can raise BaseException on a timer elapse, since `gevent.Timeout` derives
+        # from BaseException. This design suggests that a timeout should
+        # not be suppressed but rather allowed to propagate.
+        # asyncio.exceptions.CancelledError also derives from BaseException
+        # for same reason.
+        #
+        # What we should ensure, one way or another, is that the connection is
+        # not left in a corrupted state.
+
+        conn.send_command("GET non_existent_key")
+        resp = conn.read_response()
+        assert resp is None
+
+        with pytest.raises(exc_type):
+            conn.send_command("EXISTS non_existent_key")
+            # due to the interrupt, the integer '0' result of EXISTS will remain
+            # on the socket's buffer
+            with patch.object(socket.socket, "recv", side_effect=exc_type) as mock_recv:
+                _ = conn.read_response()
+            mock_recv.assert_called_once()
+
+        conn.send_command("GET non_existent_key")
+        resp = conn.read_response()
+        # If working properly, this will get a None.
+        # If not, it will get a zero (the integer result of the previous command).
+        assert resp is None