-
Notifications
You must be signed in to change notification settings - Fork 137
/
Copy pathtest_list.py
140 lines (119 loc) · 4.66 KB
/
test_list.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
#
# Copyright © 2018 yongman <yming0221@gmail.com>
#
# Distributed under terms of the MIT license.
"""
unit test for list type
"""
import unittest
import time
import string
import random
from rediswrap import RedisWrapper
class ListTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
print 'connect to 127.0.0.1:5379\n'
cls.r = RedisWrapper('127.0.0.1', 5379).get_instance()
cls.k1 = '__list1__'
cls.k2 = '__list2__'
cls.v1 = 'value1'
cls.v2 = 'value2'
def setUp(self):
self.r.execute_command('del', self.k1)
self.r.execute_command('del', self.k2)
pass
def random_string(n):
return ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(n))
def test_lpop(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
for i in range(200):
self.assertEqual(self.r.lpop(self.k1), str(i))
def test_lpush(self):
for i in range(200):
self.assertTrue(self.r.lpush(self.k1, str(i)))
for i in range(200):
self.assertEqual(self.r.rpop(self.k1), str(i))
def test_rpop(self):
for i in range(200):
self.assertTrue(self.r.lpush(self.k1, str(i)))
for i in range(200):
self.assertEqual(self.r.rpop(self.k1), str(i))
def test_rpush(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
for i in range(200):
self.assertEqual(self.r.lpop(self.k1), str(i))
def test_llen(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
self.assertEqual(self.r.llen(self.k1), 200)
def test_lindex(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
for i in range(200):
self.assertEqual(self.r.lindex(self.k1, i), str(i))
def test_lrange(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
self.assertListEqual(self.r.lrange(self.k1, 10, 100), [str(i) for i in range(10, 101)])
def test_lset(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
self.assertTrue(self.r.lset(self.k1, 100, 'hello'))
self.assertEqual(self.r.lindex(self.k1, 100), 'hello')
def test_ltrim(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
self.assertTrue(self.r.ltrim(self.k1, 0, 100))
self.assertListEqual(self.r.lrange(self.k1, 0, -1), [str(i) for i in range(0, 101)])
self.assertEqual(self.r.llen(self.k1), 101)
def test_del(self):
for i in range(200):
self.assertTrue(self.r.rpush(self.k1, str(i)))
self.assertEqual(self.r.execute_command('del', self.k1), 1)
def test_pexpire(self):
self.assertTrue(self.r.lpush(self.k1, self.v1))
# expire in 5s
self.assertTrue(self.r.execute_command('pexpire', self.k1, 5000))
self.assertLessEqual(self.r.execute_command('pttl', self.k1), 5000)
self.assertEqual(self.r.llen(self.k1), 1)
time.sleep(6)
self.assertEqual(self.r.llen(self.k1), 0)
def test_pexpireat(self):
self.assertTrue(self.r.lpush(self.k1, self.v1))
# expire in 5s
ts = int(round(time.time()*1000)) + 5000
self.assertTrue(self.r.execute_command('pexpireat', self.k1, ts))
self.assertLessEqual(self.r.execute_command('pttl', self.k1), 5000)
self.assertEqual(self.r.llen(self.k1), 1)
time.sleep(6)
self.assertEqual(self.r.llen(self.k1), 0)
def test_expire(self):
self.assertTrue(self.r.lpush(self.k1, self.v1))
# expire in 5s
self.assertTrue(self.r.execute_command('expire', self.k1, 5))
self.assertLessEqual(self.r.execute_command('ttl', self.k1), 5)
self.assertEqual(self.r.llen(self.k1), 1)
time.sleep(6)
self.assertEqual(self.r.llen(self.k1), 0)
def test_expireat(self):
self.assertTrue(self.r.lpush(self.k1, self.v1))
# expire in 5s
ts = int(round(time.time())) + 5
self.assertTrue(self.r.execute_command('expireat', self.k1, ts))
self.assertLessEqual(self.r.execute_command('ttl', self.k1), 5)
self.assertEqual(self.r.llen(self.k1), 1)
time.sleep(6)
self.assertEqual(self.r.llen(self.k1), 0)
def tearDown(self):
pass
@classmethod
def tearDownClass(cls):
cls.r.execute_command('del', cls.k1)
cls.r.execute_command('del', cls.k2)
print '\nclean up\n'