1 files changed, 126 insertions, 0 deletions
diff --git a/tests/test_asyncio/test_encoding.py b/tests/test_asyncio/test_encoding.py
new file mode 100644
index 0000000..da29837
--- /dev/null
+++ b/tests/test_asyncio/test_encoding.py
@@ -0,0 +1,126 @@
+import sys
+
+import pytest
+
+if sys.version_info[0:2] == (3, 6):
+    import pytest as pytest_asyncio
+else:
+    import pytest_asyncio
+
+import redis.asyncio as redis
+from redis.exceptions import DataError
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.onlynoncluster
+class TestEncoding:
+    @pytest_asyncio.fixture()
+    async def r(self, create_redis):
+        redis = await create_redis(decode_responses=True)
+        yield redis
+        await redis.flushall()
+
+    @pytest_asyncio.fixture()
+    async def r_no_decode(self, create_redis):
+        redis = await create_redis(decode_responses=False)
+        yield redis
+        await redis.flushall()
+
+    async def test_simple_encoding(self, r_no_decode: redis.Redis):
+        unicode_string = chr(3456) + "abcd" + chr(3421)
+        await r_no_decode.set("unicode-string", unicode_string.encode("utf-8"))
+        cached_val = await r_no_decode.get("unicode-string")
+        assert isinstance(cached_val, bytes)
+        assert unicode_string == cached_val.decode("utf-8")
+
+    async def test_simple_encoding_and_decoding(self, r: redis.Redis):
+        unicode_string = chr(3456) + "abcd" + chr(3421)
+        await r.set("unicode-string", unicode_string)
+        cached_val = await r.get("unicode-string")
+        assert isinstance(cached_val, str)
+        assert unicode_string == cached_val
+
+    async def test_memoryview_encoding(self, r_no_decode: redis.Redis):
+        unicode_string = chr(3456) + "abcd" + chr(3421)
+        unicode_string_view = memoryview(unicode_string.encode("utf-8"))
+        await r_no_decode.set("unicode-string-memoryview", unicode_string_view)
+        cached_val = await r_no_decode.get("unicode-string-memoryview")
+        # The cached value won't be a memoryview because it's a copy from Redis
+        assert isinstance(cached_val, bytes)
+        assert unicode_string == cached_val.decode("utf-8")
+
+    async def test_memoryview_encoding_and_decoding(self, r: redis.Redis):
+        unicode_string = chr(3456) + "abcd" + chr(3421)
+        unicode_string_view = memoryview(unicode_string.encode("utf-8"))
+        await r.set("unicode-string-memoryview", unicode_string_view)
+        cached_val = await r.get("unicode-string-memoryview")
+        assert isinstance(cached_val, str)
+        assert unicode_string == cached_val
+
+    async def test_list_encoding(self, r: redis.Redis):
+        unicode_string = chr(3456) + "abcd" + chr(3421)
+        result = [unicode_string, unicode_string, unicode_string]
+        await r.rpush("a", *result)
+        assert await r.lrange("a", 0, -1) == result
+
+
+@pytest.mark.onlynoncluster
+class TestEncodingErrors:
+    async def test_ignore(self, create_redis):
+        r = await create_redis(
+            decode_responses=True,
+            encoding_errors="ignore",
+        )
+        await r.set("a", b"foo\xff")
+        assert await r.get("a") == "foo"
+
+    async def test_replace(self, create_redis):
+        r = await create_redis(
+            decode_responses=True,
+            encoding_errors="replace",
+        )
+        await r.set("a", b"foo\xff")
+        assert await r.get("a") == "foo\ufffd"
+
+
+@pytest.mark.onlynoncluster
+class TestMemoryviewsAreNotPacked:
+    async def test_memoryviews_are_not_packed(self, r):
+        arg = memoryview(b"some_arg")
+        arg_list = ["SOME_COMMAND", arg]
+        c = r.connection or await r.connection_pool.get_connection("_")
+        cmd = c.pack_command(*arg_list)
+        assert cmd[1] is arg
+        cmds = c.pack_commands([arg_list, arg_list])
+        assert cmds[1] is arg
+        assert cmds[3] is arg
+
+
+class TestCommandsAreNotEncoded:
+    @pytest_asyncio.fixture()
+    async def r(self, create_redis):
+        redis = await create_redis(encoding="utf-16")
+        yield redis
+        await redis.flushall()
+
+    async def test_basic_command(self, r: redis.Redis):
+        await r.set("hello", "world")
+
+
+class TestInvalidUserInput:
+    async def test_boolean_fails(self, r: redis.Redis):
+        with pytest.raises(DataError):
+            await r.set("a", True)  # type: ignore
+
+    async def test_none_fails(self, r: redis.Redis):
+        with pytest.raises(DataError):
+            await r.set("a", None)  # type: ignore
+
+    async def test_user_type_fails(self, r: redis.Redis):
+        class Foo:
+            def __str__(self):
+                return "Foo"
+
+        with pytest.raises(DataError):
+            await r.set("a", Foo())  # type: ignore