1import testconstants
2from basetestcase import BaseTestCase
3from xdcr.xdcrbasetests import XDCRReplicationBaseTest
4from couchbase_helper.document import View
5from couchbase_helper.documentgenerator import BlobGenerator
6from membase.api.rest_client import RestConnection, Bucket
7from remote.remote_util import RemoteMachineShellConnection
8from testconstants import LINUX_BACKUP_PATH
9from testconstants import WIN_BACKUP_PATH, WIN_BACKUP_C_PATH
10from testconstants import LINUX_COUCHBASE_BIN_PATH
11from testconstants import WIN_COUCHBASE_BIN_PATH
12
13
14class CommunityBaseTest(BaseTestCase):
15
16    def setUp(self):
17        super(CommunityBaseTest, self).setUp()
18        self.product = self.input.param("product", "cb")
19        self.vbuckets = self.input.param("vbuckets", 128)
20        self.version = self.input.param("version", "2.5.1-1082")
21        self.type = self.input.param('type', 'community')
22        self.doc_ops = self.input.param("doc_ops", None)
23        if self.doc_ops is not None:
24            self.doc_ops = self.doc_ops.split(";")
25        self.defaul_map_func = "function (doc) {\n  emit(doc._id, doc);\n}"
26        self.couchbase_login = "%s:%s" % (self.input.membase_settings.rest_username,
27                                          self.input.membase_settings.rest_password)
28        self.backup_option = self.input.param("backup_option", '')
29        #define the data that will be used to test
30        self.blob_generator = self.input.param("blob_generator", True)
31        self.cli_test = self.input.param("cli_test", False)
32        self.rest = RestConnection(self.master)
33        if self.rest.is_enterprise_edition():
34            raise Exception("This couchbase server is not Community Edition."
35                  "Tests require Community Edition to test")
36        self.version = self.rest.get_nodes_version()[:5]
37
38        if self.blob_generator:
39            #gen_load data is used for upload before each test(1000 items by default)
40            self.gen_load = BlobGenerator('test', 'test-', self.value_size, end=self.num_items)
41            #gen_update is used for doing mutation for 1/2th of uploaded data
42            self.gen_update = BlobGenerator('test', 'test-', self.value_size, end=(self.num_items / 2 - 1))
43            #upload data before each test
44            self._load_all_buckets(self.servers[0], self.gen_load, "create", 0)
45        else:
46            self._load_doc_data_all_buckets()
47        self.remote = RemoteMachineShellConnection(self.master)
48        type = self.remote.extract_remote_info().distribution_type
49        self.backup_location = LINUX_BACKUP_PATH
50        self.backup_c_location = LINUX_BACKUP_PATH
51        self.bin_path = LINUX_COUCHBASE_BIN_PATH
52        self.file_extension = ""
53        if type.lower() == 'windows':
54            self.is_linux = False
55            self.backup_location = WIN_BACKUP_PATH
56            self.backup_c_location = WIN_BACKUP_C_PATH
57            self.bin_path = WIN_COUCHBASE_BIN_PATH
58            self.file_extension = ".exe"
59        else:
60            self.is_linux = True
61
62    def tearDown(self):
63        """ Some test involve kill couchbase server.  If the test steps failed
64            right after kill erlang process, we need to start couchbase server
65            in teardown so that the next test will not be false failed """
66        super(CommunityBaseTest, self).tearDown()
67
68
69class CommunityXDCRBaseTest(XDCRReplicationBaseTest):
70    def setup(self):
71        super(CommunityXDCRBaseTest, self).setUp()
72
73
74    def tearDown(self):
75        super(CommunityXDCRBaseTest, self).tearDown()
76