refactor: use pyupgrade with --py37-plus

Author: artemrys

.pre-commit-config.yaml

@@ -3,7 +3,7 @@ repos:
     rev: v2.29.0
     hooks:
     -   id: pyupgrade
-        args: [--py3-plus]
+        args: [--py37-plus]
 -   repo: https://github.com/psf/black
     rev: 21.9b0
     hooks:

examples/main.py

@@ -35,7 +35,7 @@
 
 def setup_environment():
     print("Setup solnlib demo environment...")
-    print("Copying {} to {}".format(source_dir, target_dir))
+    print(f"Copying {source_dir} to {target_dir}")
     shutil.copytree(source_dir, target_dir)
     shutil.copytree(solnlib_lib_dir, solnlib_lib_target_dir)
     os.system(splunk_bin + " start")

examples/test_splunkenv.py

@@ -44,4 +44,4 @@ def test_splunkenv():
     assert port
 
     uri = splunkenv.get_splunkd_uri()
-    assert uri == "{scheme}://{host}:{port}".format(scheme=scheme, host=host, port=port)
+    assert uri == f"{scheme}://{host}:{port}"

solnlib/conf_manager.py

@@ -82,7 +82,7 @@ def __init__(
         host=None,
         port=None,
         realm=None,
-        **context
+        **context,
     ):
         self._name = name
         self._conf = conf
@@ -112,7 +112,7 @@ def _cred_mgr(self):
                 scheme=self._scheme,
                 host=self._host,
                 port=self._port,
-                **self._context
+                **self._context,
             )
 
         return self._cred_manager
@@ -222,12 +222,12 @@ def get(self, stanza_name, only_current_app=False):
                 raise
 
             raise ConfStanzaNotExistException(
-                "Stanza: {} does not exist in {}.conf".format(stanza_name, self._name)
+                f"Stanza: {stanza_name} does not exist in {self._name}.conf"
             )
 
         if len(stanza_mgrs) == 0:
             raise ConfStanzaNotExistException(
-                "Stanza: {} does not exist in {}.conf".format(stanza_name, self._name)
+                f"Stanza: {stanza_name} does not exist in {self._name}.conf"
             )
 
         stanza = self._decrypt_stanza(stanza_mgrs[0].name, stanza_mgrs[0].content)
@@ -257,7 +257,7 @@ def get_all(self, only_current_app=False):
         """
 
         if only_current_app:
-            stanza_mgrs = self._conf.list(search="eai:acl.app={}".format(self._app))
+            stanza_mgrs = self._conf.list(search=f"eai:acl.app={self._app}")
         else:
             stanza_mgrs = self._conf.list()
         res = {}
@@ -337,7 +337,7 @@ def delete(self, stanza_name):
                 "Delete stanza: %s error: %s.", stanza_name, traceback.format_exc()
             )
             raise ConfStanzaNotExistException(
-                "Stanza: {} does not exist in {}.conf".format(stanza_name, self._name)
+                f"Stanza: {stanza_name} does not exist in {self._name}.conf"
             )
 
     @retry(exceptions=[binding.HTTPError])
@@ -405,7 +405,7 @@ def __init__(
         host=None,
         port=None,
         realm=None,
-        **context
+        **context,
     ):
         self._session_key = session_key
         self._app = app
@@ -421,7 +421,7 @@ def __init__(
             scheme=self._scheme,
             host=self._host,
             port=self._port,
-            **self._context
+            **self._context,
         )
         self._confs = None
         self._realm = realm
@@ -462,7 +462,7 @@ def get_conf(self, name, refresh=False):
             self._host,
             self._port,
             self._realm,
-            **self._context
+            **self._context,
         )
 
     @retry(exceptions=[binding.HTTPError])
@@ -489,5 +489,5 @@ def create_conf(self, name):
             self._host,
             self._port,
             self._realm,
-            **self._context
+            **self._context,
         )

solnlib/credentials.py

@@ -98,7 +98,7 @@ def __init__(
         scheme=None,
         host=None,
         port=None,
-        **context
+        **context,
     ):
         self._realm = realm
         self.service = rest_client.SplunkRestClient(
@@ -108,7 +108,7 @@ def __init__(
             scheme=scheme,
             host=host,
             port=port,
-            **context
+            **context,
         )
         self._storage_passwords = self.service.storage_passwords
 
@@ -139,7 +139,7 @@ def get_password(self, user):
                 return password["clear_password"]
 
         raise CredentialNotExistException(
-            "Failed to get password of realm={}, user={}.".format(self._realm, user)
+            f"Failed to get password of realm={self._realm}, user={user}."
         )
 
     @retry(exceptions=[binding.HTTPError])
@@ -244,7 +244,7 @@ def delete_password(self, user):
     def _get_all_passwords_in_realm(self):
         if self._realm:
             all_passwords = self._storage_passwords.list(
-                count=-1, search="realm={}".format(self._realm)
+                count=-1, search=f"realm={self._realm}"
             )
         else:
             all_passwords = self._storage_passwords.list(count=-1, search="")
@@ -255,7 +255,7 @@ def _get_all_passwords(self):
         all_passwords = self._storage_passwords.list(count=-1)
 
         results = {}
-        ptn = re.compile(r"(.+){cred_sep}(\d+)".format(cred_sep=self.SEP))
+        ptn = re.compile(fr"(.+){self.SEP}(\d+)")
         for password in all_passwords:
             match = ptn.match(password.name)
             if match:

solnlib/log.py

@@ -146,9 +146,9 @@ def _reset_root_logger(cls):
     @classmethod
     def _get_log_file(cls, name):
         if cls._default_namespace:
-            name = "{}_{}.log".format(cls._default_namespace, name)
+            name = f"{cls._default_namespace}_{name}.log"
         else:
-            name = "{}.log".format(name)
+            name = f"{name}.log"
 
         if cls._default_directory:
             directory = cls._default_directory

solnlib/net_utils.py

@@ -152,7 +152,7 @@ def wrapper(*args, **kwargs):
                 if arg in optional and optional[arg] == value:
                     continue
                 if not func(value):
-                    raise ValueError("Illegal argument: {}={}".format(arg, value))
+                    raise ValueError(f"Illegal argument: {arg}={value}")
             return f(*args, **kwargs)
 
         return wrapper

solnlib/splunk_rest_client.py

@@ -47,7 +47,7 @@ def _get_proxy_info(context):
     if context.get("proxy_username") and context.get("proxy_password"):
         username = quote(context["proxy_username"], safe="")
         password = quote(context["proxy_password"], safe="")
-        user_pass = "{user}:{password}@".format(user=username, password=password)
+        user_pass = f"{username}:{password}@"
 
     proxy = "http://{user_pass}{host}:{port}".format(
         user_pass=user_pass, host=context["proxy_hostname"], port=context["proxy_port"]
@@ -148,7 +148,7 @@ def request(url, message, **kwargs):
                 verify=verify,
                 proxies=proxies,
                 cert=cert,
-                **kwargs
+                **kwargs,
             )
         except Exception as e:
             logging.error(
@@ -208,7 +208,7 @@ def __init__(
         scheme=None,
         host=None,
         port=None,
-        **context
+        **context,
     ):
         # Only do splunkd URI discovery in SPLUNK env (SPLUNK_HOME is set)
         if not all([scheme, host, port]) and os.environ.get("SPLUNK_HOME"):

solnlib/splunkenv.py

@@ -140,7 +140,7 @@ def make_splunkhome_path(parts):
     # Check that we haven't escaped from intended parent directories.
     if os.path.relpath(fullpath, basepath)[0:2] == "..":
         raise ValueError(
-            'Illegal escape from parent directory "{}": {}'.format(basepath, fullpath)
+            f'Illegal escape from parent directory "{basepath}": {fullpath}'
         )
     return fullpath
 
@@ -207,7 +207,7 @@ def get_splunkd_uri():
         return os.environ["SPLUNKD_URI"]
 
     scheme, host, port = get_splunkd_access_info()
-    return "{scheme}://{host}:{port}".format(scheme=scheme, host=host, port=port)
+    return f"{scheme}://{host}:{port}"
 
 
 def get_conf_key_value(conf_name, stanza, key):

solnlib/user_access.py

@@ -291,7 +291,7 @@ def __init__(
         scheme=None,
         host=None,
         port=None,
-        **context
+        **context,
     ):
         collection_name = "{app}_{collection_name}".format(
             app=app, collection_name=collection_name
@@ -613,7 +613,7 @@ def __init__(
         scheme=None,
         host=None,
         port=None,
-        **context
+        **context,
     ):
         self._app = app
 
@@ -722,7 +722,7 @@ def check_user_access(
     scheme=None,
     host=None,
     port=None,
-    **context
+    **context,
 ):
     """User access checker.
 
@@ -778,7 +778,7 @@ def check_user_access(
         scheme=scheme,
         host=host,
         port=port,
-        **context
+        **context,
     ):
         raise UserAccessException(
             "Permission denied, %s does not have the capability: %s."
@@ -868,7 +868,7 @@ def get_user_capabilities(
     _rest_client = rest_client.SplunkRestClient(
         session_key, "-", scheme=scheme, host=host, port=port, **context
     )
-    url = "/services/authentication/users/{username}".format(username=username)
+    url = f"/services/authentication/users/{username}"
     try:
         response = _rest_client.get(url, output_mode="json").body.read()
     except binding.HTTPError as e:
@@ -947,7 +947,7 @@ def get_user_roles(session_key, username, scheme=None, host=None, port=None, **c
     _rest_client = rest_client.SplunkRestClient(
         session_key, "-", scheme=scheme, host=host, port=port, **context
     )
-    url = "/services/authentication/users/{username}".format(username=username)
+    url = f"/services/authentication/users/{username}"
     try:
         response = _rest_client.get(url, output_mode="json").body.read()
     except binding.HTTPError as e: