Version 2.1.7

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.1.6
+current_version = 2.1.7
 commit = False
 tag = False
 message = 'Version {new_version}'

Makefile

@@ -35,11 +35,4 @@ test_sgw_cli:
 		python -m pytest tests/test_5.py
 test_capella:
 		python -m pytest tests/test_6.py
-test:
-		python -m pytest \
-		tests/test_1.py \
-		tests/test_2.py \
-		tests/test_3.py \
-		tests/test_4.py \
-		tests/test_5.py \
-		tests/test_6.py
+test: test_sync_drv test_async_drv test_cbc_cli test_random test_sgw_cli test_capella

README.md

@@ -1,4 +1,4 @@
-# cb-util 2.1.6
+# cb-util 2.1.7
 
 ## Couchbase Utilities
 Couchbase connection manager. Simplifies connecting to a Couchbase cluster and performing data and management operations.

VERSION

@@ -1 +1 @@
-2.1.6
+2.1.7

cbcmgr/__init__.py

@@ -2,7 +2,7 @@
 from pkg_resources import parse_version
 
 _ROOT = os.path.abspath(os.path.dirname(__file__))
-__version__ = "2.1.6"
+__version__ = "2.1.7"
 VERSION = parse_version(__version__)
 
 

cbcmgr/cb_connect.py

@@ -95,7 +95,7 @@ def bucket_stats(self, bucket: str):
         except Exception as err:
             raise BucketStatsError(f"can not get bucket {bucket} stats: {err}")
 
-    @retry(factor=0.5)
+    @retry()
     def bucket_wait(self, bucket: str, count: int = 0):
         try:
             bucket_stats = self.bucket_stats(bucket)
@@ -104,7 +104,7 @@ def bucket_wait(self, bucket: str, count: int = 0):
         except Exception as err:
             raise BucketWaitException(f"bucket_wait: error: {err}")
 
-    @retry(factor=0.5)
+    @retry()
     def scope_wait(self, bucket: str, scope: str):
         bucket = self._cluster.bucket(bucket)
         cm = bucket.collections()

cbcmgr/cb_connect_lite.py

@@ -130,6 +130,13 @@ def get_collection(self, bucket: Bucket, scope: Scope, name: str = "_default") -
             raise CollectionNotFoundException(f"collection {name} does not exist")
         return scope.collection(name)
 
+    @retry()
+    def collection_wait(self, bucket: Bucket, scope: Scope, name: str = "_default"):
+        if name is None:
+            raise TypeError("name can not be None")
+        if not self.is_collection(bucket, scope.name, name):
+            raise CollectionNotFoundException(f"wait timeout: collection {name} does not exist")
+
     @retry()
     def create_collection(self, bucket: Bucket, scope: Scope, name: str):
         if name is None:
@@ -140,6 +147,7 @@ def create_collection(self, bucket: Bucket, scope: Scope, name: str):
                 collection_spec = CollectionSpec(name, scope_name=scope.name)
                 cm = bucket.collections()
                 cm.create_collection(collection_spec)
+                self.collection_wait(bucket, scope, name)
         except CollectionAlreadyExistsException:
             pass
 
@@ -240,8 +248,6 @@ def create_group(self, name: str = None, description: str = None, roles: List[Ro
         if self.capella_project and self.capella_db:
             logger.warning("Skipping group creation on Capella")
         else:
-            for role in roles:
-                print(role.as_dict())
             um = self._cluster.users()
             # noinspection PyTypeChecker
             group = Group(name=name, description=description, roles=roles)
@@ -287,13 +293,13 @@ def index_by_query(self, sql: str):
             raise IndexInternalError(f"can not determine index for query")
 
     @retry()
-    def index_create(self, index: CBQueryIndex, timeout: int = 480):
+    def index_create(self, index: CBQueryIndex, timeout: int = 480, deferred: bool = True):
         if index.is_primary:
             index_options = CreatePrimaryQueryIndexOptions()
         else:
             index_options = CreateQueryIndexOptions()
 
-        index_options.update(deferred=True)
+        index_options.update(deferred=deferred)
         index_options.update(timeout=timedelta(seconds=timeout))
         index_options.update(num_replicas=index.num_replica)
         index_options.update(ignore_if_exists=True)

cbcmgr/cb_management.py

@@ -168,7 +168,7 @@ def wait_for_query_ready(self):
         cluster = Cluster.connect(self.cb_connect_string, self.cluster_options)
         cluster.wait_until_ready(timedelta(seconds=30), WaitUntilReadyOptions(service_types=[ServiceType.Query, ServiceType.Management]))
 
-    @retry(factor=0.5)
+    @retry()
     def wait_for_index_ready(self):
         value = []
         query_str = r"SELECT * FROM system:indexes;"
@@ -425,7 +425,7 @@ def is_index(self, index_name: str = None):
 
         return False
 
-    @retry(factor=0.5, allow_list=(IndexNotReady,))
+    @retry(allow_list=(IndexNotReady,))
     def index_wait(self, index_name: str = None):
         record_count = self.collection_count()
         try:
@@ -482,7 +482,7 @@ def index_online(self, name=None, primary=False, timeout=480):
         except WatchQueryIndexTimeoutException:
             raise IndexNotReady(f"Indexes not build within {timeout} seconds...")
 
-    @retry(factor=0.5, allow_list=(IndexNotReady,))
+    @retry(allow_list=(IndexNotReady,))
     def index_list(self):
         return_list = {}
         try:
@@ -494,7 +494,7 @@ def index_list(self):
         except Exception as err:
             raise IndexNotReady(f"index_list: bucket {self._bucket.name} error: {err}")
 
-    @retry(factor=0.5, allow_list=(IndexNotReady,))
+    @retry(allow_list=(IndexNotReady,))
     def delete_wait(self, index_name: str = None):
         if self.is_index(index_name=index_name):
             raise IndexNotReady(f"delete_wait: index still exists")

cbcmgr/cb_operation_s.py

@@ -185,6 +185,7 @@ def _collection_(self, name: str = "_default"):
         except CollectionNotFoundException:
             if self.create:
                 self.create_collection(self._bucket, self._scope, name)
+                self._collection_name = name
                 self.reconnect()
                 return self._collection_(name)
         self._collection_name = name

cbcmgr/cb_session.py

@@ -142,7 +142,7 @@ def cb_connect_string(self):
         logger.debug(f"Connect string: {connect_string}")
         return connect_string
 
-    @retry(retry_count=5)
+    @retry(retry_count=7)
     def is_reachable(self):
         resolver = dns.resolver.Resolver()
         resolver.timeout = 5
@@ -209,7 +209,7 @@ def process_cluster_data(self):
         self.cbas_memory_quota = self.cluster_info.get('cbasMemoryQuota', 0)
         self.eventing_memory_quota = self.cluster_info.get('eventingMemoryQuota', 0)
 
-    @retry(retry_count=5)
+    @retry(retry_count=7)
     def check_node_connectivity(self, hostname, port):
         try:
             sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -226,7 +226,7 @@ def check_node_connectivity(self, hostname, port):
         else:
             raise NodeConnectionFailed(f"node {hostname}:{port} unreachable")
 
-    @retry(factor=0.5)
+    @retry()
     def wait_until_ready(self):
         nodes = []
         cluster = Cluster(self.cb_connect_string, ClusterOptions(self.auth,

cbcmgr/cli/cbcutil.py

@@ -61,6 +61,7 @@ def local_args(self):
         opt_parser.add_argument('--docid', action='store', help="Import document ID field", default="doc_id")
         opt_parser.add_argument('--tls', action='store_true', help="Enable SSL")
         opt_parser.add_argument('--safe', action='store_true', help="Do not overwrite data")
+        opt_parser.add_argument('--defer', action='store_true', help="Defer index build")
         opt_parser.add_argument('-e', '--external', action='store_true', help='Use external network')
         opt_parser.add_argument('-f', '--file', action='store', help="File based collection schema JSON")
         opt_parser.add_argument('--outfile', action='store', help="Output file", default="output.dat")
@@ -118,7 +119,7 @@ def run(self):
             if self.options.replicate_command == 'source':
                 Replicator(self.options.filter).source()
             elif self.options.replicate_command == 'target':
-                Replicator().target()
+                Replicator(deferred=self.options.defer).target()
         else:
             if config.op_mode == OperatingMode.LOAD.value and self.options.schema:
                 MainLoop().schema_load()

cbcmgr/cli/replicate.py

@@ -47,12 +47,13 @@ def default(self, obj):
 
 class Replicator(object):
 
-    def __init__(self, filters=None):
+    def __init__(self, filters=None, deferred=True):
         self.output = {}
         if filters:
             self.filters = filters
         else:
             self.filters = []
+        self.deferred = deferred
         self.bucket_filters = []
         self.scope_filters = []
         self.collection_filters = []
@@ -210,7 +211,7 @@ def read_schema_from_input(self):
                         for index in index_list:
                             entry = CBQueryIndex.from_dict(index)
                             logger.info(f"Replicating index [{entry.keyspace_id}] {entry.name}")
-                            operator.index_create(entry)
+                            operator.index_create(entry, deferred=self.deferred)
             except Empty:
                 time.sleep(0.1)
                 continue

cbcmgr/cli/sgwutil.py

@@ -208,7 +208,7 @@ def resync(self, name):
             logger.error(f"Resync failed for database {name}: {err}")
             sys.exit(1)
 
-    @retry(factor=0.5, retry_count=20)
+    @retry(retry_count=12)
     def resync_wait(self, name):
         self.api_post(f"/{name}/_online", None)
 
@@ -237,7 +237,7 @@ def list_all(self):
             logger.error(f"Database list failed: {err}")
             sys.exit(1)
 
-    @retry(factor=0.5, retry_count=20)
+    @retry(retry_count=12)
     def ready_wait(self, name):
         self.api_get(f"/{name}/_config").json()