Refactoring and improved error handling

cfbs/analyze.py

@@ -8,7 +8,7 @@
     version_as_comparable_list,
 )
 from cfbs.utils import (
-    FetchError,
+    CFBSNetworkError,
     cfbs_dir,
     deduplicate_list,
     fetch_url,
@@ -17,7 +17,7 @@
     immediate_subdirectories,
     mkdir,
     read_json,
-    GenericExitError,
+    CFBSExitError,
 )
 
 
@@ -115,11 +115,11 @@ def mpf_vcf_dicts(offline=False):
 
         cfbs_ri_dir = os.path.join(cfbs_dir(), RI_SUBDIRS)
         if not os.path.exists(cfbs_ri_dir):
-            raise GenericExitError(ERROR_MESSAGE)
+            raise CFBSExitError(ERROR_MESSAGE)
 
         ri_versions = immediate_subdirectories(cfbs_ri_dir)
         if len(ri_versions) == 0:
-            raise GenericExitError(ERROR_MESSAGE)
+            raise CFBSExitError(ERROR_MESSAGE)
 
         ri_latest_version = max(ri_versions)
         mpf_vcf_path = os.path.join(
@@ -136,8 +136,8 @@ def mpf_vcf_dicts(offline=False):
 
         try:
             latest_release_data = get_json(LATEST_RELEASE_API_URL)
-        except FetchError:
-            raise GenericExitError(
+        except CFBSNetworkError:
+            raise CFBSExitError(
                 "Downloading CFEngine release information failed - check your Wi-Fi / network settings."
             )
 
@@ -157,8 +157,8 @@ def mpf_vcf_dicts(offline=False):
             archive_checksums_path = ri_version_path + "/checksums.txt"
             try:
                 fetch_url(ri_checksums_url, archive_checksums_path)
-            except FetchError as e:
-                raise GenericExitError(str(e))
+            except CFBSNetworkError as e:
+                raise CFBSExitError(str(e))
 
             with open(archive_checksums_path) as file:
                 lines = [line.rstrip() for line in file]
@@ -630,7 +630,7 @@ def analyze_policyset(
                     + "\n  ".join(possible_policyset_relpaths)
                     + "\n"
                 )
-            raise GenericExitError(
+            raise CFBSExitError(
                 "There doesn't seem to be a valid policy set in the supplied path.\n       Usage: cfbs analyze path/to/policy-set\n"
                 + extra_error_text
             )

cfbs/args.py

@@ -3,7 +3,7 @@
 from typing import List, Union
 
 from cfbs import commands
-from cfbs.utils import cache, GenericExitError
+from cfbs.utils import cache, CFBSExitError
 
 
 class ArgsTypesNamespace(argparse.Namespace):
@@ -54,13 +54,13 @@ def get_manual():
             with open(file_path, "r", encoding="utf-8") as man_file:
                 man = man_file.read()
                 if not man:
-                    raise GenericExitError("Manual file is empty")
+                    raise CFBSExitError("Manual file is empty")
                 else:
                     return man
         except OSError:
-            raise GenericExitError("Error reading manual file " + file_path)
+            raise CFBSExitError("Error reading manual file " + file_path)
     else:
-        raise GenericExitError("Manual file does not exist")
+        raise CFBSExitError("Manual file does not exist")
 
 
 @cache

cfbs/build.py

@@ -26,7 +26,7 @@
     sh,
     strip_left,
     touch,
-    GenericExitError,
+    CFBSExitError,
     write_json,
 )
 from cfbs.pretty import pretty, pretty_file
@@ -92,17 +92,17 @@ def _perform_replace_step(n, a, b, filename):
         with open(filename, "r") as f:
             content = f.read()
     except FileNotFoundError:
-        raise GenericExitError("No such file '%s' in replace build step" % (filename,))
+        raise CFBSExitError("No such file '%s' in replace build step" % (filename,))
     except:
-        raise GenericExitError(
+        raise CFBSExitError(
             "Could not open/read '%s' in replace build step" % (filename,)
         )
     new_content = previous_content = content
     for i in range(0, n):
         previous_content = new_content
         new_content = previous_content.replace(a, b, 1)
         if new_content == previous_content:
-            raise GenericExitError(
+            raise CFBSExitError(
                 "replace build step could only replace '%s' in '%s' %s times, not %s times (required)"
                 % (a, filename, i, n)
             )
@@ -114,12 +114,12 @@ def _perform_replace_step(n, a, b, filename):
             if new_content == previous_content:
                 break
     if a in new_content:
-        raise GenericExitError("too many occurences of '%s' in '%s'" % (a, filename))
+        raise CFBSExitError("too many occurences of '%s' in '%s'" % (a, filename))
     try:
         with open(filename, "w") as f:
             f.write(new_content)
     except:
-        raise GenericExitError("Failed to write to '%s'" % (filename,))
+        raise CFBSExitError("Failed to write to '%s'" % (filename,))
 
 
 def _perform_build_step(module, i, step, max_length):
@@ -160,7 +160,7 @@ def _perform_build_step(module, i, step, max_length):
             dst = ""
         print("%s json '%s' 'masterfiles/%s'" % (prefix, src, dst))
         if not os.path.isfile(os.path.join(source, src)):
-            raise GenericExitError("'%s' is not a file" % src)
+            raise CFBSExitError("'%s' is not a file" % src)
         src, dst = os.path.join(source, src), os.path.join(destination, dst)
         extras, original = read_json(src), read_json(dst)
         if not extras:
@@ -230,7 +230,7 @@ def _perform_build_step(module, i, step, max_length):
         extras = _generate_augment(name, extras)
         log.debug("Generated augment: %s", pretty(extras))
         if not extras:
-            raise GenericExitError(
+            raise CFBSExitError(
                 "Input data '%s' is incomplete: Skipping build step."
                 % os.path.basename(src)
             )
@@ -253,7 +253,7 @@ def _perform_build_step(module, i, step, max_length):
                 cf_files = find("out/masterfiles/" + file, extension=".cf")
                 files += (strip_left(f, "out/masterfiles/") for f in cf_files)
             else:
-                raise GenericExitError(
+                raise CFBSExitError(
                     "Unsupported filetype '%s' for build step '%s': "
                     % (file, operation)
                     + "Expected directory (*/) of policy file (*.cf)"
@@ -308,33 +308,33 @@ def _perform_build_step(module, i, step, max_length):
 
 def perform_build(config) -> int:
     if not config.get("build"):
-        raise GenericExitError("No 'build' key found in the configuration")
+        raise CFBSExitError("No 'build' key found in the configuration")
 
     # mini-validation
     for module in config.get("build", []):
         for step in module["steps"]:
             operation, args = split_build_step(step)
 
             if step.split() != [operation] + args:
-                raise GenericExitError(
+                raise CFBSExitError(
                     "Incorrect whitespace in the `%s` build step - singular spaces are required"
                     % step
                 )
 
             if operation not in AVAILABLE_BUILD_STEPS:
-                raise GenericExitError("Unknown build step operation: %s" % operation)
+                raise CFBSExitError("Unknown build step operation: %s" % operation)
 
             expected = AVAILABLE_BUILD_STEPS[operation]
             actual = len(args)
             if not step_has_valid_arg_count(args, expected):
                 if type(expected) is int:
-                    raise GenericExitError(
+                    raise CFBSExitError(
                         "The `%s` build step expects %d arguments, %d were given"
                         % (step, expected, actual)
                     )
                 else:
                     expected = int(expected[0:-1])
-                    raise GenericExitError(
+                    raise CFBSExitError(
                         "The `%s` build step expects %d or more arguments, %d were given"
                         % (step, expected, actual)
                     )

cfbs/cfbs_config.py

@@ -7,7 +7,8 @@
 
 from cfbs.result import Result
 from cfbs.utils import (
-    GenericExitError,
+    CFBSExitError,
+    CFBSUserError,
     read_file,
     write_json,
     load_bundlenames,
@@ -100,9 +101,9 @@ def add_with_dependencies(
             module_str = module
             module = (remote_config or self).get_module_for_build(module, str_added_by)
             if not module:
-                raise GenericExitError("Module '%s' not found" % module_str)
+                raise CFBSExitError("Module '%s' not found" % module_str)
         if not module:
-            raise GenericExitError("Module '%s' not found" % str(module))
+            raise CFBSExitError("Module '%s' not found" % str(module))
         assert "name" in module
         name = module["name"]
         assert "steps" in module
@@ -151,7 +152,7 @@ def _add_using_url(
         if len(to_add) == 0:
             modules = list(provides.values())
             if not any(modules):
-                raise GenericExitError("no modules available, nothing to do")
+                raise CFBSExitError("no modules available, nothing to do")
             print("Found %d modules in '%s':" % (len(modules), url))
             for m in modules:
                 deps = m.get("dependencies", [])
@@ -169,7 +170,7 @@ def _add_using_url(
         else:
             missing = [k for k in to_add if k not in provides]
             if missing:
-                raise GenericExitError("Missing modules: " + ", ".join(missing))
+                raise CFBSExitError("Missing modules: " + ", ".join(missing))
             modules = [provides[k] for k in to_add]
 
         for i, module in enumerate(modules, start=1):
@@ -388,7 +389,7 @@ def add_command(
         checksum=None,
     ) -> Result:
         if not to_add:
-            raise GenericExitError("Must specify at least one module to add")
+            raise CFBSUserError("Must specify at least one module to add")
 
         modules_in_build_key = self.get("build", [])
         assert type(modules_in_build_key) is list
@@ -404,7 +405,7 @@ def add_command(
         else:
             # for this `if` to be valid, module names containing `://` should be illegal
             if "://" in to_add[0]:
-                raise GenericExitError(
+                raise CFBSUserError(
                     "URI scheme not supported. The supported URI schemes are: "
                     + ", ".join(SUPPORTED_URI_SCHEMES)
                 )
@@ -459,7 +460,7 @@ def input_command(self, module_name, input_data):
         def _check_keys(keys, input_data):
             for key in keys:
                 if key not in input_data:
-                    raise GenericExitError(
+                    raise CFBSExitError(
                         "Expected attribute '%s' in input definition: %s"
                         % (key, pretty(input_data))
                     )
@@ -478,7 +479,7 @@ def _input_elements(subtype):
             for element in subtype:
                 _check_keys(["type", "label", "question", "key"], element)
                 if element["type"] != "string":
-                    raise GenericExitError(
+                    raise CFBSExitError(
                         "Subtype of type '%s' not supported for type list"
                         % element["type"]
                     )
@@ -505,7 +506,7 @@ def _input_list(input_data):
             elif isinstance(subtype, dict):
                 _check_keys(["type", "label", "question"], subtype)
                 if subtype["type"] != "string":
-                    raise GenericExitError(
+                    raise CFBSExitError(
                         "Subtype of type '%s' not supported for type list"
                         % subtype["type"]
                     )
@@ -518,7 +519,7 @@ def _input_list(input_data):
                 ).lower() in ("yes", "y"):
                     result.append(_input_string(subtype))
                 return result
-            raise GenericExitError(
+            raise CFBSExitError(
                 "Expected the value of attribute 'subtype' to be a JSON list or object, not: %s"
                 % pretty(input_data["subtype"])
             )
@@ -532,8 +533,6 @@ def _input_list(input_data):
             elif definition["type"] == "list":
                 definition["response"] = _input_list(definition)
             else:
-                raise GenericExitError(
-                    "Unsupported input type '%s'" % definition["type"]
-                )
+                raise CFBSExitError("Unsupported input type '%s'" % definition["type"])
 
         return None

cfbs/cfbs_json.py

@@ -4,7 +4,7 @@
 
 from cfbs.index import Index
 from cfbs.pretty import pretty, TOP_LEVEL_KEYS, MODULE_KEYS
-from cfbs.utils import read_json, GenericExitError
+from cfbs.utils import read_json, CFBSExitError
 
 
 def _construct_provided_module(name, data, url, commit, added_by="cfbs add"):
@@ -16,7 +16,7 @@ def _construct_provided_module(name, data, url, commit, added_by="cfbs add"):
     module = OrderedDict()
     module["name"] = name
     if "description" not in data:
-        raise GenericExitError(
+        raise CFBSExitError(
             "missing required key 'description' in module definition: %s" % pretty(data)
         )
     module["description"] = data["description"]
@@ -31,7 +31,7 @@ def _construct_provided_module(name, data, url, commit, added_by="cfbs add"):
     if "input" in data:
         module["input"] = data["input"]
     if "steps" not in data:
-        raise GenericExitError(
+        raise CFBSExitError(
             "missing required key 'steps' in module definition: %s" % pretty(data)
         )
     module["steps"] = data["steps"]
@@ -154,7 +154,7 @@ def get_provides(self, added_by="cfbs add"):
         modules = OrderedDict()
         assert self._data is not None
         if "provides" not in self._data:
-            raise GenericExitError(
+            raise CFBSExitError(
                 "missing required key 'provides' in module definition: %s"
                 % pretty(self._data)
             )