Add CE restriction tests

Implement automated tests for Community Edition 5-node limit enforcement:
- test_ce_5_node_limit_restart_persistence: Validates node limit, persistence across restart, and recovery path
- test_ce_service_topology_restrictions: Validates CE service restrictions

Change-Id: I4a7318364ca945c2f9c0ccc00bbaa669c9e33998
Reviewed-on: https://review.couchbase.org/c/TAF/+/244683
Reviewed-by: <pulkit.matta@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: kushagra1502

conf/ns_server/ce_restrictions.conf

@@ -0,0 +1,6 @@
+# CE Restriction Tests
+# Jenkins install param: edition=community
+ns_server.ce_restrictions.CommunityEditionRestrictions:
+    test_ce_5_node_limit_restart_persistence,nodes_init=5,services_init=kv-kv-kv-kv-kv,default_bucket=True,bucket_size=256,replicas=0,bucket_storage=couchstore,GROUP=P0
+    test_ce_reject_ee_only_services,nodes_init=1,services_init=kv,default_bucket=True,bucket_size=256,replicas=0,bucket_storage=couchstore,GROUP=P0
+    test_ce_allow_valid_service_combinations,nodes_init=1,services_init=kv,default_bucket=True,bucket_size=256,replicas=0,bucket_storage=couchstore,kv_quota_percent=25,index_quota_percent=25,fts_quota_percent=25,GROUP=P0

pytests/ns_server/ce_restrictions.py

@@ -0,0 +1,136 @@
+"""
+Community Edition Restrictions Test Module
+
+Validates ns_server enforcement of Community Edition limitations:
+- 5-node cluster limit
+- EE-only service restrictions (analytics, eventing, backup)
+- CE topology restrictions (no query-only nodes)
+"""
+
+from basetestcase import ClusterSetup
+from ns_server.ce_restrictions_util import CommunityEditionRestrictionsUtil
+
+
+class CommunityEditionRestrictions(ClusterSetup):
+    """Test class for Community Edition restriction enforcement."""
+
+    def setUp(self):
+        super(CommunityEditionRestrictions, self).setUp()
+        self.ce_util = CommunityEditionRestrictionsUtil(
+            self.cluster, self.cluster_util, self.log, self.sleep)
+
+        if self.cluster_util.is_enterprise_edition(self.cluster):
+            self.fail("Test requires Community Edition cluster. "
+                      "Install with edition=community parameter.")
+
+        self.log.info("Cluster edition: %s", self.cluster.edition)
+
+    def tearDown(self):
+        self.ce_util.cleanup_pending_nodes()
+        super(CommunityEditionRestrictions, self).tearDown()
+
+    def test_ce_5_node_limit_restart_persistence(self):
+        """
+        Validates:
+        1. CE cluster cannot rebalance with >5 nodes
+        2. Restriction persists across master restart
+        3. Recovery path (scale-down) works
+        """
+        if len(self.cluster.servers) < 6:
+            self.fail("Requires 6 servers (5 cluster + 1 test node)")
+
+        self.ce_util.verify_ce_edition_via_diag_eval()
+
+        # Verify cluster has 5 active nodes
+        active_nodes = self.ce_util.get_active_node_count()
+        self.assertEqual(active_nodes, self.ce_util.CE_NODE_LIMIT,
+                         "Expected %d nodes, found %d"
+                         % (self.ce_util.CE_NODE_LIMIT, active_nodes))
+
+        node_to_add = self.cluster.servers[self.nodes_init]
+        self.log.info("Adding 6th node: %s", node_to_add.ip)
+        self.ce_util.add_node_without_rebalance(node_to_add, services=["kv"])
+        self.ce_util.verify_node_in_pending_state(node_to_add.ip)
+
+        # Rebalance should fail
+        self.log.info("Verifying rebalance fails with >5 nodes")
+        self.ce_util.attempt_rebalance_expect_failure()
+
+        # Restart master and verify persistence
+        self.log.info("Restarting master to verify persistence")
+        self.ce_util.restart_couchbase_server(self.cluster.master)
+        self.ce_util.verify_node_in_pending_state(node_to_add.ip)
+
+        self.log.info("Verifying restriction persists after restart")
+        self.ce_util.attempt_rebalance_expect_failure()
+
+        self.log.info("Removing extra node to restore compliance")
+        self.ce_util.rebalance_out_node(node_to_add)
+
+        final_count = self.ce_util.get_active_node_count()
+        self.assertEqual(final_count, self.ce_util.CE_NODE_LIMIT,
+                         "Expected %d nodes after recovery, found %d"
+                         % (self.ce_util.CE_NODE_LIMIT, final_count))
+
+        self.log.info("CE 5-node limit enforcement validated")
+
+    def test_ce_reject_ee_only_services(self):
+        """
+        Validates CE rejects EE-only services via CLI:
+        - analytics: EE only
+        - eventing: EE only
+        - backup: EE only
+        - query-only: Invalid CE topology
+        """
+        test_node = self._get_available_node()
+
+        # EE-only services with expected error messages
+        ee_services = [
+            ("analytics", "analytics service is only available on Enterprise Edition"),
+            ("eventing", "eventing service is only available on Enterprise Edition"),
+            ("backup", "backup service is only available on Enterprise Edition"),
+            ("data,analytics", "analytics service is only available on Enterprise Edition"),
+        ]
+
+        for services, expected_error in ee_services:
+            success, error_msg = self.ce_util.add_node_via_cli(
+                test_node, services, expect_success=False)
+            self.assertTrue(success, "Services '%s' should be rejected" % services)
+            self.assertIn(expected_error, error_msg,
+                          "Expected '%s' in: %s" % (expected_error, error_msg))
+            self.log.info("Rejected '%s': %s", services, error_msg)
+            self.sleep(8, "Wait after ejection")
+
+        # Query-only topology restriction
+        success, error_msg = self.ce_util.add_node_via_cli(
+            test_node, "query", expect_success=False)
+        self.assertTrue(success, "Query-only should be rejected")
+        self.assertIn("Community Edition only supports", error_msg)
+        self.log.info("Rejected 'query-only': %s", error_msg)
+        self.sleep(8, "Wait after ejection")
+
+    def test_ce_allow_valid_service_combinations(self):
+        """
+        Validates CE allows valid service combinations:
+        - data (kv)
+        - data,query,index (kv,n1ql,index)
+        - data,query,index,fts (kv,n1ql,index,fts)
+        """
+        test_node = self._get_available_node()
+        services = "kv,index,n1ql,fts"
+
+        self.log.info("Adding node %s with services: %s", test_node.ip, services)
+        self.ce_util.add_node_via_rest_and_rebalance_in(test_node, services)
+        self.log.info("Rebalance-in succeeded")
+
+        # Cleanup
+        self.ce_util.rebalance_out_node(test_node)
+        self.log.info("Cleanup completed")
+
+    def _get_available_node(self):
+        """Get a node not currently in the cluster."""
+        available = [s for s in self.cluster.servers
+                     if s not in self.cluster.nodes_in_cluster]
+        if not available:
+            self.fail("No available nodes. Requires at least 2 nodes in node.ini")
+        return available[0]

pytests/ns_server/ce_restrictions_util.py

@@ -0,0 +1,228 @@
+"""
+Community Edition Restrictions Utility Module
+
+Helper methods for CE restriction testing:
+- Node management (add, eject, rebalance)
+- CE edition verification
+- Service restriction validation
+"""
+
+from cb_server_rest_util.cluster_nodes.cluster_nodes_api import ClusterRestAPI
+from cb_tools.cb_cli import CbCli
+from couchbase_utils.cluster_utils.cluster_ready_functions import ClusterUtils
+from couchbase_utils.rebalance_utils.rebalance_util import RebalanceUtil
+from custom_exceptions.exception import RebalanceFailedException
+from shell_util.remote_connection import RemoteMachineShellConnection
+
+
+class CommunityEditionRestrictionsUtil:
+    """Utility class for CE restriction test operations."""
+
+    CE_NODE_LIMIT = 5
+    CE_REBALANCE_ERROR_MSG = "Cannot rebalance with more than 5"
+
+    def __init__(self, cluster, cluster_util, log, sleep_func):
+        self.cluster = cluster
+        self.cluster_util = cluster_util
+        self.log = log
+        self.sleep = sleep_func
+        self.rest = ClusterRestAPI(cluster.master)
+
+    def verify_ce_edition_via_diag_eval(self):
+        """Verify cluster is running CE via diag/eval."""
+        shell = RemoteMachineShellConnection(self.cluster.master)
+        shell.enable_diag_eval_on_non_local_hosts()
+        shell.disconnect()
+
+        status, content = self.rest.diag_eval(
+            "cluster_compat_mode:is_enterprise().")
+        if not status:
+            raise AssertionError("Failed to execute diag/eval")
+
+        is_ce = content is False if isinstance(content, bool) \
+            else str(content).strip().lower() == "false"
+
+        if not is_ce:
+            raise AssertionError("Expected CE edition. Got: %s" % content)
+
+        self.log.info("Verified CE edition via diag/eval")
+        return True
+
+    def add_node_without_rebalance(self, node, services=None):
+        """Add a node without triggering rebalance."""
+        services = services or ["kv"]
+        self.log.info("Adding node %s with services %s", node.ip, services)
+        try:
+            return self.cluster_util.add_node(
+                self.cluster, node, services=services, rebalance=False)
+        except Exception as e:
+            raise AssertionError("Failed to add node %s: %s" % (node.ip, e))
+
+    def verify_node_in_pending_state(self, node_ip):
+        """Verify node is in inactiveAdded (pending) state."""
+        nodes = ClusterUtils.get_nodes(self.cluster.master, inactive_added=True)
+        for node in nodes:
+            if node.ip == node_ip:
+                if node.clusterMembership != "inactiveAdded":
+                    raise AssertionError(
+                        "Node %s expected 'inactiveAdded', got '%s'"
+                        % (node_ip, node.clusterMembership))
+                self.log.info("Node %s is in pending state", node_ip)
+                return True
+        raise AssertionError("Node %s not found in cluster" % node_ip)
+
+    def attempt_rebalance_expect_failure(self, expected_error=None):
+        """Attempt rebalance expecting CE restriction failure."""
+        expected_error = expected_error or self.CE_REBALANCE_ERROR_MSG
+        nodes = ClusterUtils.get_nodes(self.cluster.master, inactive_added=True)
+        known_nodes = [n.id for n in nodes]
+
+        self.log.info("Attempting rebalance with %d nodes (expecting failure)",
+                      len(nodes))
+
+        status, content = self.rest.rebalance(known_nodes=known_nodes,
+                                              eject_nodes=[])
+        if not status:
+            error_msg = content.decode('utf-8') if isinstance(content, bytes) \
+                else str(content)
+            if expected_error not in error_msg:
+                raise AssertionError("Expected CE error. Got: %s" % error_msg)
+            self.log.info("Rebalance rejected: %s", error_msg)
+            return error_msg
+
+        # Monitor if rebalance started
+        try:
+            RebalanceUtil(self.cluster).monitor_rebalance()
+            raise AssertionError("Rebalance should have failed but succeeded")
+        except RebalanceFailedException as e:
+            error_msg = str(e)
+            if expected_error not in error_msg:
+                raise AssertionError("Expected CE error. Got: %s" % error_msg)
+            self.log.info("Rebalance failed: %s", error_msg)
+            return error_msg
+
+    def restart_couchbase_server(self, server):
+        """Restart Couchbase Server and wait for ready."""
+        self.log.info("Restarting server %s", server.ip)
+        shell = RemoteMachineShellConnection(server)
+        shell.restart_couchbase()
+        shell.disconnect()
+        self.cluster_util.wait_for_ns_servers_or_assert([server], wait_time=120)
+        self.log.info("Server %s ready", server.ip)
+
+    def rebalance_out_node(self, node_to_remove):
+        """Rebalance out a node from the cluster."""
+        nodes = ClusterUtils.get_nodes(self.cluster.master, inactive_added=True)
+        otp_node = next((n.id for n in nodes if n.ip == node_to_remove.ip), None)
+
+        if not otp_node:
+            raise AssertionError("Node %s not found" % node_to_remove.ip)
+
+        self.log.info("Rebalancing out %s", node_to_remove.ip)
+        result = ClusterUtils.rebalance(
+            self.cluster, wait_for_completion=True, ejected_nodes=[otp_node])
+
+        if not result:
+            raise AssertionError("Rebalance-out failed")
+
+        self.cluster_util.update_cluster_nodes_service_list(self.cluster)
+        self.log.info("Removed node %s", node_to_remove.ip)
+
+    def get_active_node_count(self):
+        """Get count of active nodes in cluster."""
+        return len(ClusterUtils.get_nodes(self.cluster.master))
+
+    def cleanup_pending_nodes(self):
+        """Eject all pending (inactiveAdded) nodes."""
+        cleaned = 0
+        try:
+            nodes = ClusterUtils.get_nodes(self.cluster.master,
+                                           inactive_added=True)
+            pending = [n for n in nodes if n.clusterMembership == "inactiveAdded"]
+            for node in pending:
+                try:
+                    self.rest.eject_node(node.id)
+                    cleaned += 1
+                except Exception as e:
+                    self.log.warning("Failed to eject %s: %s", node.id, e)
+            if cleaned:
+                self.log.info("Cleaned up %d pending nodes", cleaned)
+        except Exception as e:
+            self.log.warning("Cleanup error: %s", e)
+        return cleaned
+
+    def add_node_via_cli(self, node, services, expect_success=True):
+        """
+        Add node via couchbase-cli server-add.
+        Returns (success, error_msg) tuple.
+        """
+        self._eject_node_by_ip(node.ip)
+        self.sleep(8, "Wait after ejection")
+
+        shell = RemoteMachineShellConnection(self.cluster.master)
+        cb_cli = CbCli(shell, username=self.cluster.master.rest_username,
+                       password=self.cluster.master.rest_password)
+
+        try:
+            output = cb_cli.add_node(node, services)
+            output_str = "\n".join(output) if isinstance(output, list) \
+                else str(output)
+            shell.disconnect()
+
+            if "ERROR:" in output_str:
+                raise Exception(output_str)
+
+            # Success - cleanup
+            self._eject_node_by_ip(node.ip)
+            self.sleep(10, "Wait after ejection")
+            return (True, "") if expect_success else (False, "Should be rejected")
+
+        except Exception as e:
+            shell.disconnect()
+            error_msg = str(e)
+            return (True, error_msg) if not expect_success else (False, error_msg)
+
+    def _eject_node_by_ip(self, node_ip):
+        """Eject a pending node by IP."""
+        try:
+            nodes = ClusterUtils.get_nodes(self.cluster.master,
+                                           inactive_added=True)
+            for n in nodes:
+                if n.ip == node_ip:
+                    self.rest.eject_node(n.id)
+                    return True
+        except Exception as e:
+            self.log.warning("Failed to eject %s: %s", node_ip, e)
+        return False
+
+    def add_node_via_rest_and_rebalance_in(self, node, services):
+        """Add node via REST and rebalance in."""
+        self._eject_node_by_ip(node.ip)
+        self.sleep(5, "Wait after ejection")
+
+        self.log.info("Adding %s with services=%s", node.ip, services)
+        status, content = self.rest.add_node(
+            host_name=node.ip,
+            username=node.rest_username,
+            password=node.rest_password,
+            services=services)
+
+        if not status:
+            raise AssertionError("Failed to add node: %s" % content)
+
+        self.cluster_util.update_cluster_nodes_service_list(
+            self.cluster, inactive_added=True)
+        self.verify_node_in_pending_state(node.ip)
+
+        nodes = ClusterUtils.get_nodes(self.cluster.master, inactive_added=True)
+        known_nodes = [n.id for n in nodes]
+
+        status, content = self.rest.rebalance(known_nodes=known_nodes,
+                                              eject_nodes=[])
+        if not status:
+            raise AssertionError("Failed to start rebalance: %s" % content)
+
+        if not RebalanceUtil(self.cluster).monitor_rebalance():
+            raise AssertionError("Rebalance-in failed")
+
+        self.cluster_util.update_cluster_nodes_service_list(self.cluster)