MB-52177 Relax 'cluster_init_check' for 'ssl-manage'

Change-Id: I3ee747d449361b7c295514722e327de4071f5819
Reviewed-on: https://review.couchbase.org/c/couchbase-cli/+/174956
Well-Formed: Restriction Checker
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: James Lee <james.lee@couchbase.com>

Author: timofey-barmin

cbmgr.py

@@ -3428,9 +3428,6 @@ def __init__(self):
 
     @rest_initialiser(version_check=True)
     def execute(self, opts):
-        if not (opts.cluster_ca or opts.cluster_cert or opts.delete_ca or opts.upload_cert):
-            check_cluster_initialized(self.rest)
-
         if opts.regenerate is not None:
             try:
                 open(opts.regenerate, 'a', encoding="utf-8").close()
@@ -3481,16 +3478,20 @@ def val_or_unknown(ca, key):
 
         elif opts.load_ca:
             nodes_data, errors = self.rest.pools('nodes')
-            _exit_if_errors(errors)
-            loaded_none = True
-            for node in nodes_data['nodes']:
-                hostname = f'http://{node["hostname"]}'
+            if errors and errors[0] == 'unknown pool':
+                hostnames = [self.rest.hostname]
+            else:
+                _exit_if_errors(errors)
+                scheme = 'http'
                 if opts.ssl:
-                    hostname = f'https://{node["hostname"]}'
+                    scheme = 'https'
+                hostnames = [f'{scheme}://{n["hostname"]}' for n in nodes_data['nodes']]
+            loaded_none = True
+            for hostname in hostnames:
                 _, errors = self.rest.load_cluster_ca(hostname)
                 if not errors:
                     loaded_none = False
-                    print(f'{node["hostname"]}: Successfully load CA from inbox/CA')
+                    print(f'{hostname}: Successfully load CA from inbox/CA')
                 # If a CA is not loaded ns_server returns a error, this error is handled and the next node is tried
                 elif "Couldn't load CA certificate" in errors[0]:
                     break

test/test_cli.py

@@ -1559,6 +1559,15 @@ def test_cluster_ca_load(self):
         self.assertIn('localhost', self.str_output)
         self.assertIn('127.0.0.1', self.str_output)
 
+    def test_cluster_ca_load_not_init(self):
+        self.server_args['/pools/nodes'] = 'unknown pool'
+        self.server_args['override-status'] = 400
+        self.server_args['init'] = False
+        self.no_error_run(self.command + ['--cluster-ca-load'], self.server_args)
+        self.assertIn('POST:/node/controller/loadTrustedCAs', self.server.trace)
+        self.assertIn('127.0.0.1', self.str_output)
+        self.assertIn('Successfully load CA from inbox/CA', self.str_output)
+
     def test_cluster_ca_delete(self):
         self.no_error_run(self.command + ['--cluster-ca-delete', '0'], self.server_args)
         self.assertIn('DELETE:/pools/default/trustedCAs/0', self.server.trace)
@@ -1612,18 +1621,47 @@ def test_node_cert_info(self):
         self.assertIn(f'GET:/pools/default/certificate/node/{host}:{port}', self.server.trace)
         self.assertIn('127.0.0.1', self.str_output)
 
+    def test_node_cert_info_not_init(self):
+        self.server_args['init'] = False
+        certificate = {'warnings': [{'message': 'Out-of-the-box certificates are self-signed. To further secure your '
+                                                'system, you must create new X.509 certificates signed by a trusted '
+                                                'CA.'}],
+                       'subject': f'CN=Couchbase Server Node ({host})',
+                       'expires': '2049-12-31T23:59:59.000Z',
+                       'type': 'generated',
+                       'pem': '-----BEGIN CERTIFICATE-----\nCert String\n-----END CERTIFICATE-----\n',
+                       'privateKeyPassphrase': {}}
+        self.server_args[f'/pools/default/certificate/node/{host}:{port}'] = certificate
+        self.no_error_run(self.command + ['--node-cert-info'], self.server_args)
+        self.assertIn(f'GET:/pools/default/certificate/node/{host}:{port}', self.server.trace)
+        self.assertIn('127.0.0.1', self.str_output)
+
     def test_regenerate_cert(self):
         self.server_args['/controller/regenerateCertificate'] = 'This is a cert'
         self.no_error_run(self.command + ['--regenerate-cert', 'node1.pem'], self.server_args)
         os.remove('node1.pem')
         self.assertIn('POST:/controller/regenerateCertificate', self.server.trace)
         self.assertIn('Certificate regenerate and copied to `node1.pem`', self.str_output)
 
+    def test_regenerate_cert_not_init(self):
+        self.server_args['init'] = False
+        self.server_args['/controller/regenerateCertificate'] = 'This is a cert'
+        self.no_error_run(self.command + ['--regenerate-cert', 'node1.pem'], self.server_args)
+        os.remove('node1.pem')
+        self.assertIn('POST:/controller/regenerateCertificate', self.server.trace)
+        self.assertIn('Certificate regenerate and copied to `node1.pem`', self.str_output)
+
     def test_set_node_certificate(self):
         self.no_error_run(self.command + ['--set-node-certificate'], self.server_args)
         self.assertIn('POST:/node/controller/reloadCertificate', self.server.trace)
         self.assertIn('Node certificate set', self.str_output)
 
+    def test_set_node_certificate_not_init(self):
+        self.server_args['init'] = False
+        self.no_error_run(self.command + ['--set-node-certificate'], self.server_args)
+        self.assertIn('POST:/node/controller/reloadCertificate', self.server.trace)
+        self.assertIn('Node certificate set', self.str_output)
+
     def test_set_node_certificate_with_pkey_settings(self):
         pkey_settings_file = tempfile.NamedTemporaryFile(delete=False)
         pkey_settings_file.write(b'{"type":"plain","password":"asdf"}')
@@ -1642,12 +1680,28 @@ def test_set_client_auth(self):
         self.assertIn('POST:/settings/clientCertAuth', self.server.trace)
         self.assertIn('SSL client auth updated', self.str_output)
 
+    def test_set_client_auth_not_init(self):
+        self.server_args['init'] = False
+        client_json = tempfile.NamedTemporaryFile(delete=False)
+        client_json.write(b'{"name":"json"}')
+        client_json.close()
+        self.no_error_run(self.command + ['--set-client-auth', client_json.name], self.server_args)
+        self.assertIn('POST:/settings/clientCertAuth', self.server.trace)
+        self.assertIn('SSL client auth updated', self.str_output)
+
     def test_client_auth(self):
         self.server_args['/settings/clientCertAuth'] = {'prefixes': [], 'state': 'disable'}
         self.no_error_run(self.command + ['--client-auth'], self.server_args)
         self.assertIn('GET:/settings/clientCertAuth', self.server.trace)
         self.assertIn('prefixes', self.str_output)
 
+    def test_client_auth_not_init(self):
+        self.server_args['init'] = False
+        self.server_args['/settings/clientCertAuth'] = {'prefixes': [], 'state': 'disable'}
+        self.no_error_run(self.command + ['--client-auth'], self.server_args)
+        self.assertIn('GET:/settings/clientCertAuth', self.server.trace)
+        self.assertIn('prefixes', self.str_output)
+
 
 class TestUserManage(CommandTest):
     def setUp(self):