# Path: aredis/exceptions.py
# class RedisClusterException(Exception):
#     pass
# 
# class ResponseError(RedisError):
#     pass
# 
# class DataError(RedisError):
#     pass
# 
# class RedisError(Exception):
#     pass
# 
# Path: aredis/utils.py
# def b(x):
#     return x.encode('latin-1') if not isinstance(x, bytes) else x
# 
# def iteritems(x):
#     return iter(x.items())
# 
# def iterkeys(x):
#     return iter(x.keys())
# 
# def itervalues(x):
#     return iter(x.values())
# 
# Path: aredis/commands/server.py
# def parse_info(response):
#     """Parses the result of Redis's INFO command into a Python dict"""
#     info = {}
#     response = nativestr(response)
# 
#     def get_value(value):
#         if ',' not in value or '=' not in value:
#             try:
#                 if '.' in value:
#                     return float(value)
#                 else:
#                     return int(value)
#             except ValueError:
#                 return value
#         else:
#             sub_dict = {}
#             for item in value.split(','):
#                 k, v = item.rsplit('=', 1)
#                 sub_dict[k] = get_value(v)
#             return sub_dict
# 
#     for line in response.splitlines():
#         if line and not line.startswith('#'):
#             if line.find(':') != -1:
#                 key, value = line.split(':', 1)
#                 info[key] = get_value(value)
#             else:
#                 # if the line isn't splittable, append it to the "__raw__" key
#                 info.setdefault('__raw__', []).append(line)
# 
#     return info
# 
# Path: tests/cluster/conftest.py
# def skip_if_server_version_lt(min_version):
#     """
#     """
#     versions = get_versions()
#     for version in versions.values():
#         if StrictVersion(version) < StrictVersion(min_version):
#             return pytest.mark.skipif(True, reason="")
#     return pytest.mark.skipif(False, reason="")
# 
# def skip_if_redis_py_version_lt(min_version):
#     """
#     """
#     import aredis
#     version = aredis.__version__
#     if StrictVersion(version) < StrictVersion(min_version):
#         return pytest.mark.skipif(True, reason="")
#     return pytest.mark.skipif(False, reason="")


import datetime
import re
import time
import pytest
from string import ascii_letters
from aredis.exceptions import RedisClusterException, ResponseError, DataError, RedisError
from aredis.utils import b, iteritems, iterkeys, itervalues
from aredis.commands.server import parse_info
from tests.cluster.conftest import skip_if_server_version_lt, skip_if_redis_py_version_lt
    @pytest.mark.asyncio
    async def test_bitop_not_supported(self, r):
        await r.flushdb()
        await r.set('a', '')
        with pytest.raises(RedisClusterException):
            await r.bitop('not', 'r', 'a')

    @pytest.mark.asyncio
    @skip_if_server_version_lt('2.8.7')
    @skip_if_redis_py_version_lt("2.10.2")
    async def test_bitpos(self, r):
        """
        Bitpos was added in redis-py in version 2.10.2

        # TODO: Added b() around keys but i think they should not have to be
                there for this command to work properly.
        """
        await r.flushdb()
        key = 'key:bitpos'
        await r.set(key, b('\xff\xf0\x00'))
        assert await r.bitpos(key, 0) == 12
        assert await r.bitpos(key, 0, 2, -1) == 16
        assert await r.bitpos(key, 0, -2, -1) == 12
        await r.set(key, b('\x00\xff\xf0'))
        assert await r.bitpos(key, 1, 0) == 8
        assert await r.bitpos(key, 1, 1) == 8
        await r.set(key, '\x00\x00\x00')
        assert await r.bitpos(key, 1) == -1

    @pytest.mark.asyncio

# Path: aredis/exceptions.py
# class RedisClusterException(Exception):
#     pass
# 
# class ResponseError(RedisError):
#     pass
# 
# class DataError(RedisError):
#     pass
# 
# class RedisError(Exception):
#     pass
# 
# Path: aredis/utils.py
# def b(x):
#     return x.encode('latin-1') if not isinstance(x, bytes) else x
# 
# def iteritems(x):
#     return iter(x.items())
# 
# def iterkeys(x):
#     return iter(x.keys())
# 
# def itervalues(x):
#     return iter(x.values())
# 
# Path: aredis/commands/server.py
# def parse_info(response):
#     """Parses the result of Redis's INFO command into a Python dict"""
#     info = {}
#     response = nativestr(response)
# 
#     def get_value(value):
#         if ',' not in value or '=' not in value:
#             try:
#                 if '.' in value:
#                     return float(value)
#                 else:
#                     return int(value)
#             except ValueError:
#                 return value
#         else:
#             sub_dict = {}
#             for item in value.split(','):
#                 k, v = item.rsplit('=', 1)
#                 sub_dict[k] = get_value(v)
#             return sub_dict
# 
#     for line in response.splitlines():
#         if line and not line.startswith('#'):
#             if line.find(':') != -1:
#                 key, value = line.split(':', 1)
#                 info[key] = get_value(value)
#             else:
#                 # if the line isn't splittable, append it to the "__raw__" key
#                 info.setdefault('__raw__', []).append(line)
# 
#     return info
# 
# Path: tests/cluster/conftest.py
# def skip_if_server_version_lt(min_version):
#     """
#     """
#     versions = get_versions()
#     for version in versions.values():
#         if StrictVersion(version) < StrictVersion(min_version):
#             return pytest.mark.skipif(True, reason="")
#     return pytest.mark.skipif(False, reason="")
# 
# def skip_if_redis_py_version_lt(min_version):
#     """
#     """
#     import aredis
#     version = aredis.__version__
#     if StrictVersion(version) < StrictVersion(min_version):
#         return pytest.mark.skipif(True, reason="")
#     return pytest.mark.skipif(False, reason="")


# Path: tests/cluster/test_commands.py
import datetime
import re
import time
import pytest
from string import ascii_letters
from aredis.exceptions import RedisClusterException, ResponseError, DataError, RedisError
from aredis.utils import b, iteritems, iterkeys, itervalues
from aredis.commands.server import parse_info
from tests.cluster.conftest import skip_if_server_version_lt, skip_if_redis_py_version_lt

    @pytest.mark.asyncio
    async def test_bitop_not_supported(self, r):
        await r.flushdb()
        await r.set('a', '')
        with pytest.raises(RedisClusterException):
            await r.bitop('not', 'r', 'a')

    @pytest.mark.asyncio
    @skip_if_server_version_lt('2.8.7')
    @skip_if_redis_py_version_lt("2.10.2")
    async def test_bitpos(self, r):
        """
        Bitpos was added in redis-py in version 2.10.2

        # TODO: Added b() around keys but i think they should not have to be
                there for this command to work properly.
        """
        await r.flushdb()
        key = 'key:bitpos'
        await r.set(key, b('\xff\xf0\x00'))
        assert await r.bitpos(key, 0) == 12
        assert await r.bitpos(key, 0, 2, -1) == 16
        assert await r.bitpos(key, 0, -2, -1) == 12
        await r.set(key, b('\x00\xff\xf0'))
        assert await r.bitpos(key, 1, 0) == 8
        assert await r.bitpos(key, 1, 1) == 8
        await r.set(key, '\x00\x00\x00')
        assert await r.bitpos(key, 1) == -1

    @pytest.mark.asyncio

    @skip_if_server_version_lt('2.8.7')