Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get(self):
Entry.register_response('GET mocket', 'is awesome!')
self.assertEqual(self.rclient.get('mocket'), b'is awesome!')
self.assertEqual(len(Mocket._requests), 1)
self.assertEqual(Mocket._requests[0].data, b'*2\r\n$3\r\nGET\r\n$6\r\nmocket\r\n')
def test_init_byte(self):
entry = Entry(addr=None, command=b'SET snowman "is \xe2\x98\x83!"', responses=[])
self.assertEqual(
entry.command,
[b'*3', b'$3', b'SET', b'$7', b'snowman', b'$7', b'is \xe2\x98\x83!']
)
def test_init_text(self):
entry = Entry(addr=None, command='SET snowman "is ☃!"', responses=[])
self.assertEqual(
entry.command,
[b'*3', b'$3', b'SET', b'$7', b'snowman', b'$7', b'is \xe2\x98\x83!']
)
def test_can_handle(self):
entry = Entry(addr=None, command='SET snowman "is ☃!"', responses=[])
self.assertTrue(entry.can_handle(b'*3\r\n$3\r\nSET\r\n$7\r\nsnowman\r\n$7\r\nis \xe2\x98\x83!'))
def test_register(self):
Entry.register(('localhost', 6379), 'SET snowman "is ☃!"', OK)
self.assertEqual(
Mocket._entries[('localhost', 6379)][0].command,
[b'*3', b'$3', b'SET', b'$7', b'snowman', b'$7', b'is \xe2\x98\x83!']
)
self.assertEqual(
Mocket._entries[('localhost', 6379)][0].responses[0].data,
b'+OK\r\n'
)
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')
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')
def test_init_text(self):
entry = Entry(addr=None, command='SET snowman "is ☃!"', responses=[])
self.assertEqual(
entry.command,
[b'*3', b'$3', b'SET', b'$7', b'snowman', b'$7', b'is \xe2\x98\x83!']
)
def test_init_byte(self):
entry = Entry(addr=None, command=b'SET snowman "is \xe2\x98\x83!"', responses=[])
self.assertEqual(
entry.command,
[b'*3', b'$3', b'SET', b'$7', b'snowman', b'$7', b'is \xe2\x98\x83!']
)