<commit_before>from distutils.version import StrictVersion
import unittest
import datetime
import time
import binascii

from redis._compat import (unichr, u, b, ascii_letters, iteritems, dictkeys,
                           dictvalues)
from redis.client import parse_info
import redis


class ServerCommandsTestCase(unittest.TestCase):
    def get_client(self, cls=redis.Redis):
        return cls(host='localhost', port=6379, db=9)

    def setUp(self):
        self.client = self.get_client()
        self.client.flushdb()

    def tearDown(self):
        self.client.flushdb()
        self.client.connection_pool.disconnect()

    def test_response_callbacks(self):
        self.assertEquals(
            self.client.response_callbacks,
            redis.Redis.RESPONSE_CALLBACKS)
        self.assertNotEquals(
            id(self.client.response_callbacks),
            id(redis.Redis.RESPONSE_CALLBACKS))
        self.client.set_response_callback('GET', lambda x: 'static')
        self.client.set('a', 'foo')
        self.assertEquals(self.client.get('a'), 'static')

    # GENERAL SERVER COMMANDS
    def test_dbsize(self):
        self.client['a'] = 'foo'
        self.client['b'] = 'bar'
        self.assertEquals(self.client.dbsize(), 2)

    def test_get_and_set(self):
        # get and set can't be tested independently of each other
        client = redis.Redis(host='localhost', port=6379, db=9)
        self.assertEquals(client.get('a'), None)
        byte_string = b('value')
        integer = 5
        unicode_string = unichr(3456) + u('abcd') + unichr(3421)
        self.assert_(client.set('byte_string', byte_string))
        self.assert_(client.set('integer', 5))
        self.assert_(client.set('unicode_string', unicode_string))
        self.assertEquals(client.get('byte_string'), byte_string)
        self.assertEquals(client.get('integer'), b(str(integer)))
        self.assertEquals(
            client.get('unicode_string').decode('utf-8'),
            unicode_string)

    def test_getitem_and_setitem(self):
        self.client['a'] = 'bar'
        self.assertEquals(self.client['a'], b('bar'))
        self.assertRaises(KeyError, self.client.__getitem__, 'b')

    def test_delete(self):
        self.assertEquals(self.client.delete('a'), False)
        self.client['a'] = 'foo'
        self.assertEquals(self.client.delete('a'), True)

    def test_delitem(self):
        self.client['a'] = 'foo'
        del self.client['a']
        self.assertEquals(self.client.get('a'), None)

    def test_client_list(self):
        clients = self.client.client_list()
        self.assert_(isinstance(clients[0], dict))
        self.assert_('addr' in clients[0])

    def test_client_getname(self):
        version = self.client.info()['redis_version']
        if StrictVersion(version) < StrictVersion('2.6.9'):
            try:
                raise unittest.SkipTest()
            except AttributeError:
                return

        name = self.client.client_getname()
        self.assertEquals(name, None)

    def test_client_setname(self):
        version = self.client.info()['redis_version']
        if StrictVersion(version) < StrictVersion('2.6.9'):
            try:
                raise unittest.SkipTest()
            except AttributeError:
                return

        self.assert_(self.client.client_setname('redis_py_test'))
        self.assertEquals(
            self.client.client_getname(),
            'redis_py_test'
        )

    def test_config_get(self):
        data = self.client.config_get()
        self.assert_('maxmemory' in data)
        self.assert_(data['maxmemory'].isdigit())

    def test_config_set(self):
        data = self.client.config_get()
        rdbname = data['dbfilename']
        self.assert_(self.client.config_set('dbfilename', 'redis_py_test.rdb'))
        self.assertEquals(
            self.client.config_get()['dbfilename'],
            'redis_py_test.rdb'
        )
        self.assert_(self.client.config_set('dbfilename', rdbname))
        self.assertEquals(self.client.config_get()['dbfilename'], rdbname)

    def test_debug_object(self):
        self.client['a'] = 'foo'
        debug_info = self.client.debug_object('a')
        self.assert_(len(debug_info) > 0)
        self.assertEquals(debug_info['refcount'], 1)
        self.assert_(debug_info['serializedlength'] > 0)
        self.client.rpush('b', 'a1')
        debug_info = self.client.debug_object('a')

    def test_echo(self):
        self.assertEquals(self.client.echo('foo bar'), b('foo bar'))

    def test_info(self):
        self.client['a'] = 'foo'
        self.client['b'] = 'bar'
        info = self.client.info()
        self.assert_(isinstance(info, dict))
        self.assertEquals(info['db9']['keys'], 2)

        import sys
        if sys.version_info[0] < 3:
            print
            print info['redis_version']
            print
        else:
            print('')
            print(info['redis_version'])
            print('')

    def test_lastsave(self):
        self.assert_(isinstance(self.client.lastsave(), datetime.datetime))

    def test_object(self):
        self.client['a'] = 'foo'
        self.assert_(isinstance(self.client.object('refcount', 'a'), int))
        self.assert_(isinstance(self.client.object('idletime', 'a'), int))
        self.assertEquals(self.client.object('encoding', 'a'), b('raw'))

    def test_ping(self):
        self.assertEquals(self.client.ping(), True)

    def test_time(self):
        version = self.client.info()['redis_version']
        if StrictVersion(version) < StrictVersion('2.6.0'):
            try:
                raise unittest.SkipTest()
            except AttributeError:
                return

        t = self.client.time()
        self.assertEquals(len(t), 2)<commit_msg>travis runs 2.6.12, great.<commit_after>from distutils.version import StrictVersion
import unittest
import datetime
import time
import binascii

from redis._compat import (unichr, u, b, ascii_letters, iteritems, dictkeys,
                           dictvalues)
from redis.client import parse_info
import redis


class ServerCommandsTestCase(unittest.TestCase):
    def get_client(self, cls=redis.Redis):
        return cls(host='localhost', port=6379, db=9)

    def setUp(self):
        self.client = self.get_client()
        self.client.flushdb()

    def tearDown(self):
        self.client.flushdb()
        self.client.connection_pool.disconnect()

    def test_response_callbacks(self):
        self.assertEquals(
            self.client.response_callbacks,
            redis.Redis.RESPONSE_CALLBACKS)
        self.assertNotEquals(
            id(self.client.response_callbacks),
            id(redis.Redis.RESPONSE_CALLBACKS))
        self.client.set_response_callback('GET', lambda x: 'static')
        self.client.set('a', 'foo')
        self.assertEquals(self.client.get('a'), 'static')

    # GENERAL SERVER COMMANDS
    def test_dbsize(self):
        self.client['a'] = 'foo'
        self.client['b'] = 'bar'
        self.assertEquals(self.client.dbsize(), 2)

    def test_get_and_set(self):
        # get and set can't be tested independently of each other
        client = redis.Redis(host='localhost', port=6379, db=9)
        self.assertEquals(client.get('a'), None)
        byte_string = b('value')
        integer = 5
        unicode_string = unichr(3456) + u('abcd') + unichr(3421)
        self.assert_(client.set('byte_string', byte_string))
        self.assert_(client.set('integer', 5))
        self.assert_(client.set('unicode_string', unicode_string))
        self.assertEquals(client.get('byte_string'), byte_string)
        self.assertEquals(client.get('integer'), b(str(integer)))
        self.assertEquals(
            client.get('unicode_string').decode('utf-8'),
            unicode_string)

    def test_getitem_and_setitem(self):
        self.client['a'] = 'bar'
        self.assertEquals(self.client['a'], b('bar'))
        self.assertRaises(KeyError, self.client.__getitem__, 'b')

    def test_delete(self):
        self.assertEquals(self.client.delete('a'), False)
        self.client['a'] = 'foo'
        self.assertEquals(self.client.delete('a'), True)

    def test_delitem(self):
        self.client['a'] = 'foo'
        del self.client['a']
        self.assertEquals(self.client.get('a'), None)

    def test_client_list(self):
        clients = self.client.client_list()
        self.assert_(isinstance(clients[0], dict))
        self.assert_('addr' in clients[0])

    def test_client_getname(self):
        version = self.client.info()['redis_version']
        if StrictVersion(version) < StrictVersion('2.6.9'):
            try:
                raise unittest.SkipTest()
            except AttributeError:
                return

        name = self.client.client_getname()
        self.assertEquals(name, None)

    def test_client_setname(self):
        version = self.client.info()['redis_version']
        if StrictVersion(version) < StrictVersion('2.6.9'):
            try:
                raise unittest.SkipTest()
            except AttributeError:
                return

        self.assert_(self.client.client_setname('redis_py_test'))
        self.assertEquals(
            self.client.client_getname(),
            'redis_py_test'
        )

    def test_config_get(self):
        data = self.client.config_get()
        self.assert_('maxmemory' in data)
        self.assert_(data['maxmemory'].isdigit())

    def test_config_set(self):
        data = self.client.config_get()
        rdbname = data['dbfilename']
        self.assert_(self.client.config_set('dbfilename', 'redis_py_test.rdb'))
        self.assertEquals(
            self.client.config_get()['dbfilename'],
            'redis_py_test.rdb'
        )
        self.assert_(self.client.config_set('dbfilename', rdbname))
        self.assertEquals(self.client.config_get()['dbfilename'], rdbname)

    def test_debug_object(self):
        self.client['a'] = 'foo'
        debug_info = self.client.debug_object('a')
        self.assert_(len(debug_info) > 0)
        self.assertEquals(debug_info['refcount'], 1)
        self.assert_(debug_info['serializedlength'] > 0)
        self.client.rpush('b', 'a1')
        debug_info = self.client.debug_object('a')

    def test_echo(self):
        self.assertEquals(self.client.echo('foo bar'), b('foo bar'))

    def test_info(self):
        self.client['a'] = 'foo'
        self.client['b'] = 'bar'
        info = self.client.info()
        self.assert_(isinstance(info, dict))
        self.assertEquals(info['db9']['keys'], 2)

    def test_lastsave(self):
        self.assert_(isinstance(self.client.lastsave(), datetime.datetime))

    def test_object(self):
        self.client['a'] = 'foo'
        self.assert_(isinstance(self.client.object('refcount', 'a'), int))
        self.assert_(isinstance(self.client.object('idletime', 'a'), int))
        self.assertEquals(self.client.object('encoding', 'a'), b('raw'))

    def test_ping(self):
        self.assertEquals(self.client.ping(), True)

    def test_time(self):
        version = self.client.info()['redis_version']
        if StrictVersion(version) < StrictVersion('2.6.0'):
            try:
                raise unittest.SkipTest()
            except AttributeError:
                return

        t = self.client.time()
        self.assertEquals(len(t), 2)
{'fraction_non_alphanumeric': 0.136307746355173, 'fraction_numerical': 0.007553135429474793, 'mean_word_length': 2.579691920779629, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6781695', 'n_tokens_mistral': 3551, 'n_tokens_neox': 3438, 'n_words': 565}