Fix static credential and custom endpoint submission

Move crn validation to common method, make more generic to work with Public and Private cloud endpoints
Fixes #1

Signed-off-by: Daniel Chaffelson <[email protected]>

Author: Chaffelson

src/cdpy/common.py

@@ -85,8 +85,11 @@ def __init__(self, value, field='error_code', default=None, warning=None):
 class StaticCredentials(Credentials):
     """A credential class that simply takes a set of static credentials."""
 
-    def __init__(self, access_key_id, private_key):
-        super(StaticCredentials, self).__init__(access_key_id, private_key, 'static')
+    def __init__(self, access_key_id, private_key, access_token='', method='static'):
+        super(StaticCredentials, self).__init__(
+            access_key_id=access_key_id, private_key=private_key,
+            access_token=access_token, method=method
+        )
 
 
 class CdpcliWrapper(object):
@@ -221,14 +224,15 @@ def _setup_logger(self, log_level, log_format):
 
         self.logger.addHandler(handler)
 
-    def _build_client(self, service, credentials=None):
+    def _build_client(self, service):
         if not self.cdp_credentials:
-            credentials = self._client_creator.context.get_credentials()
+            self.cdp_credentials = self._client_creator.context.get_credentials()
         return self._client_creator.create_client(
             service,
             self.client_endpoint,
             self.tls_verify,
-            credentials)
+            self.cdp_credentials
+        )
 
     @staticmethod
     def _default_throw_error(error: 'CdpError'):
@@ -262,6 +266,12 @@ def _default_throw_warning(warning: 'CdpWarning'):
     def regex_search(pattern, obj):
         return re.search(pattern, obj)
 
+    def validate_crn(self, obj: str):
+        if obj is not None and obj.startswith('crn:'):
+            pass
+        else:
+            self.throw_error(CdpError("Supplied env_crn %s is not a valid CDP crn" % str(obj)))
+
     @staticmethod
     def sleep(seconds):
         sleep(seconds)

src/cdpy/dw.py

@@ -62,8 +62,7 @@ def list_clusters(self, env_crn=None):
         return resp
 
     def gather_clusters(self, env_crn=None):
-        if env_crn is not None and not env_crn.startswith('crn:cdp:environments'):
-            self.sdk.throw_error(CdpError("Supplied env_crn %s is not a valid CDP Environment crn" % str(env_crn)))
+        self.sdk.validate_crn(env_crn)
         clusters = self.list_clusters(env_crn=env_crn)
         out = []
         if clusters:
@@ -77,8 +76,7 @@ def gather_clusters(self, env_crn=None):
 
     def create_cluster(self, env_crn: str, overlay: bool, aws_public_subnets: list = None,
                        aws_private_subnets: list = None, az_subnet: str = None, az_enable_az: bool = None):
-        if env_crn is not None and not env_crn.startswith('crn:cdp:environments'):
-            self.sdk.throw_error(CdpError("Supplied env_crn %s is not a valid CDP Environment crn" % str(env_crn)))
+        self.sdk.validate_crn(env_crn)
         if all(x is not None for x in [aws_private_subnets, aws_private_subnets]):
             aws_options = dict(publicSubnetIds=aws_public_subnets, privateSubnetIds=aws_private_subnets)
         else:

src/cdpy/environments.py

@@ -313,7 +313,7 @@ def set_telemetry(self, name, workload_analytics=None, logs_collection=None):
     def resolve_environment_crn(self, env: Union[str, None]):
         """Ensures a given env string is either the environment crn or None"""
         if isinstance(env, str):
-            if env.startswith('crn:cdp:environments'):
+            if env.startswith('crn:'):
                 return env
             else:
                 env_desc = self.describe_environment(env)