Migrating SGW tests related to XATTRS (#294)

* Added a new test. Wrote a function for doc deletion in CBS. Added a delete database similar to put since the older functionality was flaky and a failed test-mid-run was still running into errors in reruns.

* Added another new test. Everything works fine, ran it standalone and amongst all the others too.

* A 3rd test related to xattrs

* fixed mypy errors

* addressed Ritesh's comments, using public APIs now for tests focusing on SG Client based verification

Author: vipbhardwaj

client/src/cbltest/api/couchbaseserver.py

@@ -23,6 +23,7 @@
 from couchbase.management.collections import CollectionSpec
 from couchbase.management.options import CreatePrimaryQueryIndexOptions
 from couchbase.options import ClusterOptions
+from couchbase.subdocument import upsert
 from opentelemetry.trace import get_tracer
 
 from cbltest.api.error import CblTestError
@@ -292,6 +293,36 @@ def upsert_document(
                     f"Failed to insert document '{doc_id}' into {bucket}.{scope}.{collection}: {e}"
                 )
 
+    def delete_document(
+        self,
+        bucket: str,
+        doc_id: str,
+        scope: str = "_default",
+        collection: str = "_default",
+    ) -> None:
+        """
+        Deletes a document from the specified bucket.scope.collection.
+        """
+        with self.__tracer.start_as_current_span(
+            "delete_document",
+            attributes={
+                "cbl.bucket.name": bucket,
+                "cbl.scope.name": scope,
+                "cbl.collection.name": collection,
+                "cbl.document.id": doc_id,
+            },
+        ):
+            try:
+                bucket_obj = _try_n_times(10, 1, False, self.__cluster.bucket, bucket)
+                coll = bucket_obj.scope(scope).collection(collection)
+                coll.remove(doc_id)
+            except DocumentNotFoundException:
+                pass
+            except Exception as e:
+                raise CblTestError(
+                    f"Failed to delete document '{doc_id}' from {bucket}.{scope}.{collection}: {e}"
+                )
+
     def get_document(
         self,
         bucket: str,
@@ -329,6 +360,84 @@ def get_document(
                     f"Failed to get document '{doc_id}' from {bucket}.{scope}.{collection}: {e}"
                 )
 
+    def upsert_document_xattr(
+        self,
+        bucket: str,
+        doc_id: str,
+        xattr_key: str,
+        xattr_value: str,
+        scope: str = "_default",
+        collection: str = "_default",
+    ) -> None:
+        """
+        Upserts an xattr on a document using subdocument operations
+
+        :param bucket: The bucket containing the document
+        :param doc_id: The ID of the document to update
+        :param xattr_key: The xattr key to upsert
+        :param xattr_value: The value to set for the xattr
+        :param scope: The scope containing the document (default '_default')
+        :param collection: The collection containing the document (default '_default')
+        """
+        with self.__tracer.start_as_current_span(
+            "upsert_document_xattr",
+            attributes={
+                "cbl.bucket": bucket,
+                "cbl.scope": scope,
+                "cbl.collection": collection,
+                "cbl.document.id": doc_id,
+                "cbl.xattr.key": xattr_key,
+            },
+        ):
+            try:
+                col = self.__cluster.bucket(bucket).scope(scope).collection(collection)
+                col.mutate_in(
+                    doc_id,
+                    [upsert(xattr_key, xattr_value, xattr=True, create_parents=True)],
+                )
+            except Exception as e:
+                raise CblTestError(
+                    f"Failed to upsert xattr '{xattr_key}' on document '{doc_id}' in {bucket}.{scope}.{collection}: {e}"
+                )
+
+    def delete_document_xattr(
+        self,
+        bucket: str,
+        doc_id: str,
+        xattr_key: str,
+        scope: str = "_default",
+        collection: str = "_default",
+    ) -> None:
+        """
+        Deletes an xattr from a document using subdocument operations
+
+        :param bucket: The bucket containing the document
+        :param doc_id: The ID of the document
+        :param xattr_key: The xattr key to delete
+        :param scope: The scope containing the document (default '_default')
+        :param collection: The collection containing the document (default '_default')
+        """
+        with self.__tracer.start_as_current_span(
+            "delete_document_xattr",
+            attributes={
+                "cbl.bucket": bucket,
+                "cbl.scope": scope,
+                "cbl.collection": collection,
+                "cbl.document.id": doc_id,
+                "cbl.xattr.key": xattr_key,
+            },
+        ):
+            try:
+                from couchbase.subdocument import remove
+
+                col = self.__cluster.bucket(bucket).scope(scope).collection(collection)
+                col.mutate_in(
+                    doc_id,
+                    [remove(xattr_key, xattr=True)],
+                )
+            except Exception:
+                pass
+
     def start_xdcr(self, target: "CouchbaseServer", bucket_name: str) -> None:
         """
         Starts an XDCR replication from this cluster to the target cluster

client/src/cbltest/api/syncgateway.py

@@ -425,11 +425,74 @@ def __init__(
         self.__tracer = get_tracer(__name__, VERSION)
         self.__secure: bool = secure
         self.__hostname: str = url
+        self.__port: int = port
         self.__admin_port: int = admin_port
         self.__admin_session: ClientSession = self._create_session(
             secure, scheme, url, admin_port, BasicAuth(username, password, "ascii")
         )
 
+    @property
+    def hostname(self) -> str:
+        """Gets the hostname of the Sync Gateway instance"""
+        return self.__hostname
+
+    @property
+    def port(self) -> int:
+        """Gets the public port of the Sync Gateway instance"""
+        return self.__port
+
+    @property
+    def admin_port(self) -> int:
+        """Gets the admin port of the Sync Gateway instance"""
+        return self.__admin_port
+
+    @property
+    def secure(self) -> bool:
+        """Gets whether the Sync Gateway instance uses TLS"""
+        return self.__secure
+
+    @classmethod
+    async def create_user_client(
+        cls,
+        admin_sg: "SyncGateway",
+        db_name: str,
+        username: str,
+        password: str,
+        channels: list[str],
+    ) -> "SyncGateway":
+        """
+        Helper method to create a user with channel access and return a user-specific SG client.
+
+        This is a convenience method for tests that need to verify user-level access control.
+
+        :param admin_sg: The admin SyncGateway instance
+        :param db_name: The database name
+        :param username: The username to create
+        :param password: The password for the user
+        :param channels: List of channels the user should have access to
+        :return: A SyncGateway instance authenticated as the user for public API access
+        """
+        # Clean up user if exists from previous run
+        await admin_sg.delete_user(db_name, username)
+
+        # Create user with channel access
+        await admin_sg.add_user(
+            db_name,
+            username,
+            password=password,
+            collection_access={"_default": {"_default": {"admin_channels": channels}}},
+        )
+
+        # Return user-specific SG client for public API access
+        return cls(
+            admin_sg.hostname,
+            username,
+            password,
+            port=admin_sg.port,
+            admin_port=admin_sg.admin_port,
+            secure=admin_sg.secure,
+        )
+
     def _create_session(
         self, secure: bool, scheme: str, url: str, port: int, auth: BasicAuth | None
     ) -> ClientSession:
@@ -583,31 +646,39 @@ async def database_exists(self, db_name: str) -> bool:
                 return False
             raise
 
-    async def delete_database(self, db_name: str) -> None:
-        """
-        Deletes a database from Sync Gateway's configuration.
-
-        .. warning:: This will not delete the data from the Couchbase Server bucket.
-            To delete the data see the
-            :func:`drop_bucket()<cbltest.api.couchbaseserver.CouchbaseServer.drop_bucket>` function
-
-        :param db_name: The name of the Database to delete
-        """
+    async def _delete_database(self, db_name: str, retry_count: int = 0) -> None:
         with self.__tracer.start_as_current_span(
             "delete_database", attributes={"cbl.database.name": db_name}
-        ):
+        ) as current_span:
             try:
                 await self._send_request("delete", f"/{db_name}")
             except CblSyncGatewayBadResponseError as e:
-                if e.code == 500:
+                if e.code == 500 and retry_count < 3:
                     cbl_warning(
-                        f"SGW returned 500 when deleting {db_name}, database may be in transitional state. Ignoring."
+                        f"Sync gateway returned 500 from DELETE database call, retrying ({retry_count + 1})..."
                     )
-                elif e.code == 404:
+                    current_span.add_event("SGW returned 500, retry")
+                    import asyncio
+
+                    await asyncio.sleep(2)
+                    await self._delete_database(db_name, retry_count + 1)
+                elif e.code == 403:
                     pass
                 else:
                     raise
 
+    async def delete_database(self, db_name: str) -> None:
+        """
+        Deletes a database from Sync Gateway's configuration.
+
+        .. warning:: This will not delete the data from the Couchbase Server bucket.
+            To delete the data see the
+            :func:`drop_bucket()<cbltest.api.couchbaseserver.CouchbaseServer.drop_bucket>` function
+
+        :param db_name: The name of the Database to delete
+        """
+        await self._delete_database(db_name, 0)
+
     def create_collection_access_dict(self, input: dict[str, list[str]]) -> dict:
         """
         Creates a collection access dictionary in the format that Sync Gateway expects,
@@ -689,6 +760,25 @@ async def add_user(
                 "put", f"/{db_name}/_user/{name}", JSONDictionary(body)
             )
 
+    async def delete_user(self, db_name: str, name: str) -> None:
+        """
+        Deletes a user from a Sync Gateway database
+
+        :param db_name: The name of the Database
+        :param name: The username to delete
+        """
+        with self.__tracer.start_as_current_span(
+            "delete_user", attributes={"cbl.user.name": name}
+        ):
+            try:
+                await self._send_request("delete", f"/{db_name}/_user/{name}")
+            except CblSyncGatewayBadResponseError as e:
+                if e.code == 404:
+                    # User doesn't exist, that's fine
+                    pass
+                else:
+                    raise
+
     async def add_role(self, db_name: str, role: str, collection_access: dict) -> None:
         """
         Adds the specified role to a Sync Gateway database with the specified collection access
@@ -785,14 +875,19 @@ async def load_dataset(self, db_name: str, path: Path) -> None:
                 self._analyze_dataset_response(cast(list, resp))
 
     async def get_all_documents(
-        self, db_name: str, scope: str = "_default", collection: str = "_default"
+        self,
+        db_name: str,
+        scope: str = "_default",
+        collection: str = "_default",
+        use_public_api: bool = False,
     ) -> AllDocumentsResponse:
         """
         Gets all the documents in the given collection from Sync Gateway (id and revid)
 
         :param db_name: The name of the Sync Gateway database to query
         :param scope: The scope to use when querying Sync Gateway
         :param collection: The collection to use when querying Sync Gateway
+        :param use_public_api: If True, uses public port (4984) with user auth instead of admin port (4985)
         """
         with self.__tracer.start_as_current_span(
             "get_all_documents",
@@ -802,9 +897,28 @@ async def get_all_documents(
                 "cbl.collection.name": collection,
             },
         ):
-            resp = await self._send_request(
-                "get", f"/{db_name}.{scope}.{collection}/_all_docs"
-            )
+            if use_public_api:
+                # Use public port (4984) - required for regular user access
+                scheme = "https://" if self.__secure else "http://"
+                # Create session with user's credentials on public port
+                async with self._create_session(
+                    self.__secure,
+                    scheme,
+                    self.__hostname,
+                    4984,
+                    self.__admin_session.auth,
+                ) as session:
+                    resp = await self._send_request(
+                        "get",
+                        f"/{db_name}.{scope}.{collection}/_all_docs",
+                        session=session,
+                    )
+            else:
+                # Use admin port (4985) - default behavior
+                resp = await self._send_request(
+                    "get", f"/{db_name}.{scope}.{collection}/_all_docs"
+                )
+
             assert isinstance(resp, dict)
             return AllDocumentsResponse(cast(dict, resp))
 
@@ -814,6 +928,7 @@ async def get_changes(
         scope: str = "_default",
         collection: str = "_default",
         version_type: str = "rev",
+        use_public_api: bool = False,
     ) -> ChangesResponse:
         """
         Gets the changes feed from Sync Gateway, including deleted documents
@@ -822,6 +937,7 @@ async def get_changes(
         :param scope: The scope to use when querying Sync Gateway
         :param collection: The collection to use when querying Sync Gateway
         :param version_type: The version type to use ('rev' for revision IDs, 'cv' for version vectors in SGW 4.0+)
+        :param use_public_api: If True, uses public port (4984) with user auth instead of admin port (4985)
         """
         with self.__tracer.start_as_current_span(
             "get_changes",
@@ -832,9 +948,29 @@ async def get_changes(
             },
         ):
             query_params = f"version_type={version_type}"
-            resp = await self._send_request(
-                "get", f"/{db_name}.{scope}.{collection}/_changes?{query_params}"
-            )
+
+            if use_public_api:
+                # Use public port (4984) - required for regular user access
+                scheme = "https://" if self.__secure else "http://"
+                # Create session with user's credentials on public port
+                async with self._create_session(
+                    self.__secure,
+                    scheme,
+                    self.__hostname,
+                    4984,
+                    self.__admin_session.auth,
+                ) as session:
+                    resp = await self._send_request(
+                        "get",
+                        f"/{db_name}.{scope}.{collection}/_changes?{query_params}",
+                        session=session,
+                    )
+            else:
+                # Use admin port (4985) - default behavior
+                resp = await self._send_request(
+                    "get", f"/{db_name}.{scope}.{collection}/_changes?{query_params}"
+                )
+
             assert isinstance(resp, dict)
             return ChangesResponse(cast(dict, resp))