-
Notifications
You must be signed in to change notification settings - Fork 52
/
cas_base.py
22 lines (20 loc) · 915 Bytes
/
cas_base.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from basetestcase import BaseTestCase
from memcached.helper.data_helper import MemcachedClientHelper
from membase.api.rest_client import Bucket
class CasBaseTest(BaseTestCase):
def setUp(self):
super(CasBaseTest, self).setUp()
self.value_size = self.input.param("value_size", 256)
self.doc_ops = self.input.param("doc_ops", None)
self.mutate_times = self.input.param("mutate_times", 10)
self.expire_time = self.input.param("expire_time", 5)
if self.doc_ops is not None:
self.doc_ops = self.doc_ops.split(";")
self.clients = {}
for bucket in self.buckets:
client = MemcachedClientHelper.direct_client(self.master, bucket.name)
self.clients[bucket.name]=client
def tearDown(self):
super(CasBaseTest, self).tearDown()
for bucket in self.buckets:
self.clients[bucket.name].close()