How to use the mocket.mockredis.Redisizer function in mocket

To help you get started, we’ve selected a few mocket examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github mindflayer / python-mocket / tests / main / test_redis.py View on Github external
def test_redisize_list(self):
        self.assertEqual(
            Redisizer.redisize(['snowman', '☃']),
            b'*2\r\n$7\r\nsnowman\r\n$3\r\n\xe2\x98\x83\r\n'
        )
github mindflayer / python-mocket / tests / main / test_redis.py View on Github external
def test_token(self):
        self.assertEqual(
            Redisizer.tokens(['SET', 'snowman', 'is ☃!']),
            [b'*3', b'$3', b'SET', b'$7', b'snowman', b'$7', b'is \xe2\x98\x83!']
        )
github mindflayer / python-mocket / tests / main / test_redis.py View on Github external
def test_token(self):
        self.assertEqual(
            Redisizer.tokens(['SET', 'snowman', 'is ☃!']),
            [b'*3', b'$3', b'SET', b'$7', b'snowman', b'$7', b'is \xe2\x98\x83!']
        )
github mindflayer / python-mocket / tests / main / test_redis.py View on Github external
def test_error(self):
        self.assertEqual(
            Redisizer.error('ERR: ☃ summer'),
            b'-ERR: \xe2\x98\x83 summer\r\n'
        )
github mindflayer / python-mocket / mocket / mockredis.py View on Github external
return data
        if isinstance(data, byte_type):
            data = decode_from_bytes(data)
        return Redisizer(get_conversion(data.__class__)(data) + b'\r\n')

    @staticmethod
    def command(description, _type='+'):
        return Redisizer('{0}{1}{2}'.format(_type, description, '\r\n').encode('utf-8'))

    @staticmethod
    def error(description):
        return Redisizer.command(description, _type='-')


OK = Redisizer.command('OK')
QUEUED = Redisizer.command('QUEUED')
ERROR = Redisizer.error


class Entry(MocketEntry):
    request_cls = Request
    response_cls = Response

    def __init__(self, addr, command, responses):
        super(Entry, self).__init__(addr or ('localhost', 6379), responses)
        d = shsplit(command)
        d[0] = d[0].upper()
        self.command = Redisizer.tokens(d)

    def can_handle(self, data):
        return data.splitlines() == self.command
github mindflayer / python-mocket / mocket / mockredis.py View on Github external
if isinstance(data, byte_type):
            data = decode_from_bytes(data)
        return Redisizer(get_conversion(data.__class__)(data) + b'\r\n')

    @staticmethod
    def command(description, _type='+'):
        return Redisizer('{0}{1}{2}'.format(_type, description, '\r\n').encode('utf-8'))

    @staticmethod
    def error(description):
        return Redisizer.command(description, _type='-')


OK = Redisizer.command('OK')
QUEUED = Redisizer.command('QUEUED')
ERROR = Redisizer.error


class Entry(MocketEntry):
    request_cls = Request
    response_cls = Response

    def __init__(self, addr, command, responses):
        super(Entry, self).__init__(addr or ('localhost', 6379), responses)
        d = shsplit(command)
        d[0] = d[0].upper()
        self.command = Redisizer.tokens(d)

    def can_handle(self, data):
        return data.splitlines() == self.command

    @classmethod
github mindflayer / python-mocket / mocket / mockredis.py View on Github external
def __init__(self, data=None):
        self.data = Redisizer.redisize(data or OK)
github mindflayer / python-mocket / mocket / mockredis.py View on Github external
def redisize(data):
        def get_conversion(t):
            return {
                dict: lambda x: b'\r\n'.join(Redisizer.tokens(list(chain(*tuple(x.items()))))),
                int: lambda x: ':{0}'.format(x).encode('utf-8'),
                text_type: lambda x: '${0}\r\n{1}'.format(len(x.encode('utf-8')), x).encode('utf-8'),
                list: lambda x: b'\r\n'.join(Redisizer.tokens(x)),
            }[t]
        if isinstance(data, Redisizer):
            return data
        if isinstance(data, byte_type):
            data = decode_from_bytes(data)
        return Redisizer(get_conversion(data.__class__)(data) + b'\r\n')
github mindflayer / python-mocket / mocket / mockredis.py View on Github external
def command(description, _type='+'):
        return Redisizer('{0}{1}{2}'.format(_type, description, '\r\n').encode('utf-8'))
github mindflayer / python-mocket / mocket / mockredis.py View on Github external
def redisize(data):
        def get_conversion(t):
            return {
                dict: lambda x: b'\r\n'.join(Redisizer.tokens(list(chain(*tuple(x.items()))))),
                int: lambda x: ':{0}'.format(x).encode('utf-8'),
                text_type: lambda x: '${0}\r\n{1}'.format(len(x.encode('utf-8')), x).encode('utf-8'),
                list: lambda x: b'\r\n'.join(Redisizer.tokens(x)),
            }[t]
        if isinstance(data, Redisizer):
            return data
        if isinstance(data, byte_type):
            data = decode_from_bytes(data)
        return Redisizer(get_conversion(data.__class__)(data) + b'\r\n')