Fix/trailing slash (#54)

* rename method for clarity
* remove trailing slash from cerberus url when initializing the CerberusClient
* split out utils + add unit tests
* fix indentation
Co-authored-by: Kristof Hermans <[email protected]>

CHANGELOG.md

@@ -5,6 +5,10 @@ Date format is Day-Month-Year
 
 ### Added
 
+###### 09-12-2020
+- Version 2.5.2
+  - Bugfix trailing slashes on client initialization
+
 ###### 10-08-2020
 - Version 2.5.0
   - Allows for the use of AWS China Region STS authentication

cerberus/__init__.py

@@ -1,14 +1,15 @@
 # Setting up logging here because this is the root of the module
 import logging
 
-__all__ = ['client', 'user_auth', 'aws_auth', 'util']
+__all__ = ['aws_auth', 'client', 'network_util', 'url_util', 'user_auth']
+
+CLIENT_VERSION = '2.5.2'
 
-CLIENT_VERSION = '2.5.1'
 
 class CerberusClientException(Exception):
     """Wrap third-party exceptions expected by the Cerberus client."""
     pass
 
+
 # This avoids the "No handler found" warnings.
 logging.getLogger(__name__).addHandler(logging.NullHandler())
-

cerberus/aws_auth.py

@@ -20,7 +20,7 @@
 import sys
 
 from . import CerberusClientException, CLIENT_VERSION
-from .util import throw_if_bad_response, post_with_retry
+from .network_util import throw_if_bad_response, post_with_retry
 from collections import OrderedDict
 
 

cerberus/client.py

@@ -20,8 +20,9 @@
 from .aws_auth import AWSAuth
 from .user_auth import UserAuth
 from . import CerberusClientException, CLIENT_VERSION
-from .util import throw_if_bad_response, get_with_retry, post_with_retry, put_with_retry, delete_with_retry, \
+from .network_util import throw_if_bad_response, get_with_retry, post_with_retry, put_with_retry, delete_with_retry, \
     head_with_retry
+from .url_util import ensure_single_trailing_slash, ensure_no_trailing_slash
 
 import ast
 import json
@@ -30,7 +31,6 @@
 import warnings
 import os
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -48,33 +48,28 @@ def __init__(self, cerberus_url, username=None, password=None,
         a botocore.session.Session object, the Cerberus client will sign the request using the session provided
         instead of the default session.
 
-        verbose (default True) controls if the cerberus library will output some debuging statements to the
+        verbose (default True) controls if the cerberus library will output some debugging statements to the
         console (sys.stderr).
         
         """
-        self.cerberus_url = cerberus_url
+        self.cerberus_url = ensure_no_trailing_slash(cerberus_url)
         self.username = username or ""
         self.password = password or ""
         self.region = region
         self.token = token
+        self.aws_session = aws_session
+
         if verbose is None or type(verbose) != bool:
             self.verbose = True
         else:
             self.verbose = verbose
-        self.aws_session = aws_session
+
         if self.token is None:
             self._set_token()
 
         self.HEADERS['X-Cerberus-Token'] = self.token
         self.HEADERS['X-Cerberus-Client'] = 'CerberusPythonClient/' + CLIENT_VERSION
 
-    # noinspection PyMethodMayBeStatic
-    def _add_slash(self, string=None):
-        """ if a string doesn't end in a '/' add one """
-        if not str.endswith(string, '/'):
-            return str.join('', [string, '/'])
-        return str(string)
-
     def _set_token(self):
         """Set the Cerberus token based on auth type"""
         try:
@@ -101,8 +96,7 @@ def get_roles(self):
 
         Roles are permission levels that are granted to IAM or User Groups.  Associating the id for the write role
           would allow that IAM or User Group to write in the safe deposit box."""
-        roles_resp = get_with_retry(self.cerberus_url + '/v1/role',
-                                    headers=self.HEADERS)
+        roles_resp = get_with_retry(self.cerberus_url + '/v1/role', headers=self.HEADERS)
 
         throw_if_bad_response(roles_resp)
         return roles_resp.json()
@@ -154,9 +148,9 @@ def create_sdb(self, name, category_id, owner, description="", user_group_permis
         if iam_principal_permissions is None:
             iam_principal_permissions = []
         if list != type(user_group_permissions):
-            raise(TypeError('Expected list, but got ' + str(type(user_group_permissions))))
+            raise (TypeError('Expected list, but got ' + str(type(user_group_permissions))))
         if list != type(iam_principal_permissions):
-            raise(TypeError('Expected list, but got ' + str(type(iam_principal_permissions))))
+            raise (TypeError('Expected list, but got ' + str(type(iam_principal_permissions))))
         temp_data = {
             "name": name,
             "description": description,
@@ -185,8 +179,7 @@ def delete_sdb(self, sdb_id):
 
     def get_sdbs(self):
         """ Return a list of each SDB the client is authorized to view"""
-        sdb_resp = get_with_retry(self.cerberus_url + '/v2/safe-deposit-box',
-                                  headers=self.HEADERS)
+        sdb_resp = get_with_retry(self.cerberus_url + '/v2/safe-deposit-box', headers=self.HEADERS)
 
         throw_if_bad_response(sdb_resp)
         return sdb_resp.json()
@@ -200,7 +193,6 @@ def get_sdb_path(self, sdb):
         )
 
         throw_if_bad_response(sdb_resp)
-
         return sdb_resp.json()['path']
 
     def get_sdb_keys(self, path):
@@ -211,7 +203,6 @@ def get_sdb_keys(self, path):
         )
 
         throw_if_bad_response(list_resp)
-
         return list_resp.json()['data']['keys']
 
     def get_sdb_id(self, sdb):
@@ -233,7 +224,7 @@ def get_sdb_id_by_path(self, sdb_path):
         json_resp = self.get_sdbs()
 
         # Deal with the supplied path possibly missing an ending slash
-        path = self._add_slash(sdb_path)
+        path = ensure_single_trailing_slash(sdb_path)
 
         for r in json_resp:
             if r['path'] == path:
@@ -333,6 +324,7 @@ def update_sdb(self, sdb_id, owner=None, description=None, user_group_permission
         return sdb_resp.json()
 
     """------ Files ------"""
+
     def delete_file(self, secure_data_path):
         """Delete a file at the given secure data path"""
         secret_resp = delete_with_retry(self.cerberus_url + '/v1/secure-file/' + secure_data_path,
@@ -377,7 +369,7 @@ def _parse_metadata_filename(self, metadata):
         """
         Parse the header metadata to pull out the filename and then store it under the key 'filename'
         """
-        index = metadata['Content-Disposition'].index('=')+1
+        index = metadata['Content-Disposition'].index('=') + 1
         metadata['filename'] = metadata['Content-Disposition'][index:].replace('"', '')
         return metadata
 
@@ -402,8 +394,7 @@ def get_file_data(self, secure_data_path, version=None):
         This only returns the file data, and does not include any of the meta information stored with it.
 
         Keyword arguments:
-
-            secure_data_path (string) -- full path in the secret deposit box that contains the file key
+        secure_data_path (string) -- full path in the secret deposit box that contains the file key
         """
         return self._get_file(secure_data_path, version).content
 
@@ -462,7 +453,7 @@ def list_files(self, secure_data_path, limit=None, offset=None):
 
         # Because of the addition of versionId and the way URLs are constructed, secure_data_path should
         #  always end in a '/'.
-        secure_data_path = self._add_slash(secure_data_path)
+        secure_data_path = ensure_single_trailing_slash(secure_data_path)
         secret_resp = get_with_retry(self.cerberus_url + '/v1/secure-files/' + secure_data_path,
                                      params=payload, headers=self.HEADERS)
         throw_if_bad_response(secret_resp)
@@ -475,8 +466,8 @@ def put_file(self, secure_data_path, filehandle, content_type=None):
         Keyword arguments:
         secure_data_path -- full path in the safety deposit box that contains the file key to store things under
         filehandle -- Pass an opened filehandle to the file you want to upload.
-           Make sure that the file was opened in binary mode, otherwise the size calculations
-           can be off for text files.
+        Make sure that the file was opened in binary mode, otherwise the size calculations
+        can be off for text files.
         content_type -- Optional.  Set the Mime type of the file you're uploading.
         """
 
@@ -498,6 +489,7 @@ def put_file(self, secure_data_path, filehandle, content_type=None):
         return secret_resp
 
     """------ Secrets -----"""
+
     def delete_secret(self, secure_data_path):
         """Delete a secret from the given secure data path"""
         secret_resp = delete_with_retry(self.cerberus_url + '/v1/secret/' + secure_data_path,
@@ -620,7 +612,7 @@ def list_secrets(self, secure_data_path):
 
         # Because of the addition of versionId and the way URLs are constructed, secure_data_path should
         #  always end in a '/'.
-        secure_data_path = self._add_slash(secure_data_path)
+        secure_data_path = ensure_single_trailing_slash(secure_data_path)
         secret_resp = get_with_retry(self.cerberus_url + '/v1/secret/' + secure_data_path + '?list=true',
                                      headers=self.HEADERS)
         throw_if_bad_response(secret_resp)

cerberus/url_util.py

@@ -0,0 +1,26 @@
+"""
+Copyright 2020-present Nike, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+You may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and* limitations under the License.*
+"""
+
+
+def ensure_single_trailing_slash(string):
+    """ if a string doesn't end in a '/' add one """
+    return str.join('', [ensure_no_trailing_slash(string), '/'])
+
+
+def ensure_no_trailing_slash(string):
+    """ if a string ends in a '/' remove it """
+    while str.endswith(string, '/'):
+        string = str(string[:-1])
+    return str(string)

cerberus/user_auth.py

@@ -17,7 +17,7 @@
 
 from . import CerberusClientException, CLIENT_VERSION
 
-from .util import throw_if_bad_response, get_with_retry, post_with_retry
+from .network_util import throw_if_bad_response, get_with_retry, post_with_retry
 
 
 logger = logging.getLogger(__name__)

tests/matcher/__init__.py

@@ -1,6 +1,8 @@
 class AnyDictWithKey(str):
     """ Argument matcher that matches a dictionary (or other data structures) that contains the key"""
+
     def __eq__(self, other):
         return self in other
+
     def __hash__(self):
-        return super(AnyDictWithKey, self).__hash__()
+        return super(AnyDictWithKey, self).__hash__()