Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@mocketize
def test_gethostname(self):
self.assertEqual(socket.gethostname(), 'localhost')
@mocketize(record_truesocket=False)
def test_hm(self):
h = {b'f1': b'one', b'f2': b'two'}
self.assertTrue(self.rclient.hmset('hash', h))
self.assertEqual(self.rclient.hgetall('hash'), h)
@mocketize(record_truesocket=False)
def setUp(self):
self.rclient = redis.StrictRedis()
self.rclient.flushdb()
@mocketize
def test_incr(self):
Entry.register_responses('INCRBY counter 1', (1, 2, 3))
self.assertEqual(self.rclient.incr('counter'), 1)
self.assertEqual(self.rclient.incr('counter'), 2)
self.assertEqual(self.rclient.incr('counter'), 3)
self.assertEqual(len(Mocket._requests), 3)
self.assertEqual(Mocket._requests[0].data, b'*3\r\n$6\r\nINCRBY\r\n$7\r\ncounter\r\n$1\r\n1\r\n')
self.assertEqual(Mocket._requests[1].data, b'*3\r\n$6\r\nINCRBY\r\n$7\r\ncounter\r\n$1\r\n1\r\n')
self.assertEqual(Mocket._requests[2].data, b'*3\r\n$6\r\nINCRBY\r\n$7\r\ncounter\r\n$1\r\n1\r\n')
@mocketize(record_truesocket=False)
def test_shutdown(self):
rc = redis.StrictRedis(host='127.1.1.1')
try:
rc.get('foo')
except redis.ConnectionError:
pass
@mocketize
def test_set(self):
Entry.register_response('SET mocket "is awesome!"', OK)
self.assertTrue(self.rclient.set('mocket', 'is awesome!'))
self.assertEqual(len(Mocket._requests), 1)
self.assertEqual(Mocket.last_request().data, b'*3\r\n$3\r\nSET\r\n$6\r\nmocket\r\n$11\r\nis awesome!\r\n')
@mocketize(record_truesocket=False)
def test_select_db(self):
r = redis.StrictRedis(db=1)
r.set('foo', 10)
foo = r.get('foo')
self.assertEqual(foo, b'10')
@mocketize(record_truesocket=False)
def test_incr(self):
self.assertEqual(self.rclient.incr('counter'), 1)
self.assertEqual(self.rclient.incr('counter'), 2)
self.assertEqual(self.rclient.incr('counter'), 3)
@mocketize
def test_err(self):
Entry.register_response('INCRBY counter one', ERROR('ERR value is not an integer or out of range'))
self.assertRaises(redis.ResponseError, self.rclient.incr, 'counter', 'one')
self.assertEqual(len(Mocket._requests), 1)
self.assertEqual(Mocket.last_request().data, b'*3\r\n$6\r\nINCRBY\r\n$7\r\ncounter\r\n$3\r\none\r\n')
@mocketize
def test_write_and_read(self):
"""Tests writing-to and reading-from TCP Host."""
frame = "%s>%s:%s" % (
self.random(6),
','.join([self.random(6), self.random(6), self.random(6)]),
' '.join([
self.random(), 'test_write_and_read', self.random()])
)
aprs_frame = aprs.Frame(frame)
kiss_frame = aprs_frame.encode_kiss()
ks = kiss.TCPKISS(host=self.random_host, port=self.random_port)
a = (self.random_host, self.random_port)
entry = MocketEntry(a, [kiss_frame])
entry_1 = MocketEntry(('localhost', 80), True)