import pytest
import redis

from redis.connection import Connection
from .conftest import _get_client


class TestEncoding:
    @pytest.fixture()
    def r(self, request):
        return _get_client(redis.Redis, request=request, decode_responses=True)

    @pytest.fixture()
    def r_no_decode(self, request):
        return _get_client(
            redis.Redis,
            request=request,
            decode_responses=False,
        )

    def test_simple_encoding(self, r_no_decode):
        unicode_string = chr(3456) + 'abcd' + chr(3421)
        r_no_decode['unicode-string'] = unicode_string.encode('utf-8')
        cached_val = r_no_decode['unicode-string']
        assert isinstance(cached_val, bytes)
        assert unicode_string == cached_val.decode('utf-8')

    def test_simple_encoding_and_decoding(self, r):
        unicode_string = chr(3456) + 'abcd' + chr(3421)
        r['unicode-string'] = unicode_string
        cached_val = r['unicode-string']
        assert isinstance(cached_val, str)
        assert unicode_string == cached_val

    def test_memoryview_encoding(self, r_no_decode):
        unicode_string = chr(3456) + 'abcd' + chr(3421)
        unicode_string_view = memoryview(unicode_string.encode('utf-8'))
        r_no_decode['unicode-string-memoryview'] = unicode_string_view
        cached_val = r_no_decode['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')

    def test_memoryview_encoding_and_decoding(self, r):
        unicode_string = chr(3456) + 'abcd' + chr(3421)
        unicode_string_view = memoryview(unicode_string.encode('utf-8'))
        r['unicode-string-memoryview'] = unicode_string_view
        cached_val = r['unicode-string-memoryview']
        assert isinstance(cached_val, str)
        assert unicode_string == cached_val

    def test_list_encoding(self, r):
        unicode_string = chr(3456) + 'abcd' + chr(3421)
        result = [unicode_string, unicode_string, unicode_string]
        r.rpush('a', *result)
        assert r.lrange('a', 0, -1) == result


class TestEncodingErrors:
    def test_ignore(self, request):
        r = _get_client(redis.Redis, request=request, decode_responses=True,
                        encoding_errors='ignore')
        r.set('a', b'foo\xff')
        assert r.get('a') == 'foo'

    def test_replace(self, request):
        r = _get_client(redis.Redis, request=request, decode_responses=True,
                        encoding_errors='replace')
        r.set('a', b'foo\xff')
        assert r.get('a') == 'foo\ufffd'


class TestMemoryviewsAreNotPacked:
    def test_memoryviews_are_not_packed(self):
        c = Connection()
        arg = memoryview(b'some_arg')
        arg_list = ['SOME_COMMAND', arg]
        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.fixture()
    def r(self, request):
        return _get_client(redis.Redis, request=request, encoding='utf-16')

    def test_basic_command(self, r):
        r.set('hello', 'world')


class TestInvalidUserInput:
    def test_boolean_fails(self, r):
        with pytest.raises(redis.DataError):
            r.set('a', True)

    def test_none_fails(self, r):
        with pytest.raises(redis.DataError):
            r.set('a', None)

    def test_user_type_fails(self, r):
        class Foo:
            def __str__(self):
                return 'Foo'

        with pytest.raises(redis.DataError):
            r.set('a', Foo())