replace assert statements with exceptions

Author: geritwagner

colrev/env/language_service.py

@@ -40,7 +40,8 @@ def __init__(self) -> None:
     # pylint: disable=too-many-return-statements
     # pylint: disable=too-many-branches
     def _determine_alphabet(self, str_to_check: str) -> str:
-        assert len(str_to_check) != 0
+        if len(str_to_check) == 0:
+            raise ValueError("str_to_check must not be empty")
 
         str_to_check = re.sub(r"[\s\d\.\:]*", "", str_to_check)
 
@@ -116,7 +117,10 @@ def compute_language_confidence_values(self, *, text: str) -> list:
 
     def validate_iso_639_3_language_codes(self, *, lang_code_list: list) -> None:
         """Validates whether a list of language codes complies with the ISO 639-3 standard."""
-        assert isinstance(lang_code_list, list)
+        if not isinstance(lang_code_list, list):
+            raise TypeError(
+                f"Expected lang_code_list to be a list, got {type(lang_code_list).__name__}"
+            )
 
         invalid_language_codes = [x for x in lang_code_list if 3 != len(x)]
         if invalid_language_codes:

colrev/env/tei_parser.py

@@ -59,7 +59,8 @@ def __init__(
         - tei_path: read TEI from file.
         """
         # pylint: disable=consider-using-with
-        assert pdf_path is not None or tei_path is not None
+        if pdf_path is None and tei_path is None:
+            raise ValueError("Either pdf_path or tei_path must be provided")
         if pdf_path is not None:
             pdf_path = Path(pdf_path)
             if pdf_path.is_symlink():  # pragma: no cover

colrev/git_repo.py

@@ -58,11 +58,8 @@ def has_record_changes(self, *, change_type: str = "all") -> bool:
 
     def has_changes(self, relative_path: Path, *, change_type: str = "all") -> bool:
         """Check whether the relative path (or the git repository) has changes."""
-        assert change_type in [
-            "all",
-            "staged",
-            "unstaged",
-        ], "Invalid change_type specified"
+        if change_type not in ["all", "staged", "unstaged"]:
+            raise ValueError(f"Invalid change_type specified: {change_type!r}")
         try:
             bool(self.repo.head.commit)
         except ValueError:
@@ -179,11 +176,10 @@ def file_in_history(self, filepath: Path) -> bool:
     def get_commit_message(self, *, commit_nr: int) -> str:
         """Get the commit message for commit #."""
         master = self.repo.head.reference
-        assert commit_nr == 0  # extension : implement other cases
-        if commit_nr == 0:
-            cmsg = master.commit.message
-            return cmsg
-        return ""
+        if commit_nr != 0:
+            raise NotImplementedError("Only commit_nr == 0 is currently supported")
+        cmsg = master.commit.message
+        return cmsg
 
     def add_setting_changes(self) -> None:
         """Add changes in settings to git."""

colrev/loader/loader.py

@@ -29,7 +29,10 @@ def __init__(
         """Initialize the instance."""
         self.filename = filename
         self.unique_id_field = unique_id_field
-        assert id_labeler is not None or unique_id_field != ""
+        if id_labeler is None and unique_id_field == "":
+            raise ValueError(
+                "Either id_labeler must be provided or unique_id_field must be set"
+            )
         self.id_labeler = id_labeler
         self.entrytype_setter = entrytype_setter
         self.field_mapper = field_mapper
@@ -48,23 +51,22 @@ def _set_ids(self, records_list: list) -> None:
         else:
             self.id_labeler(records_list)  # type: ignore
 
-        assert all(
-            Fields.ID in record_dict for record_dict in records_list
-        ), "ID not set in all records"
+        if not all(Fields.ID in record_dict for record_dict in records_list):
+            raise ValueError("ID not set in all records")
         unique_ids = {record_dict[Fields.ID] for record_dict in records_list}
         non_unique_ids = [
             id
             for id in unique_ids
             if sum(1 for r in records_list if r[Fields.ID] == id) > 1
         ]
-        assert not non_unique_ids, f"ID is not unique in records: {non_unique_ids}"
+        if non_unique_ids:
+            raise ValueError(f"ID is not unique in records: {non_unique_ids}")
 
     def _set_entrytypes(self, records_dict: dict) -> None:
         for r_dict_val in records_dict.values():
             self.entrytype_setter(r_dict_val)
-        assert all(
-            Fields.ENTRYTYPE in r for r in records_dict.values()
-        ), "ENTRYTYPE not set in all records"
+        if not all(Fields.ENTRYTYPE in r for r in records_dict.values()):
+            raise ValueError("ENTRYTYPE not set in all records")
 
         for r in records_dict.values():
             if r[Fields.ENTRYTYPE] in ENTRYTYPES.get_all():

colrev/ops/merge.py

@@ -98,7 +98,8 @@ def main(self, *, branch: str) -> None:
             remote.fetch()
 
         branches = git_repo.heads
-        assert branch in [b.name for b in branches]
+        if branch not in [b.name for b in branches]:
+            raise ValueError(f"Branch does not exist: {branch!r}")
 
         git_branch = [b for b in branches if b.name == branch][0]
         merging_branch_author = git_branch.commit.author
@@ -114,7 +115,10 @@ def main(self, *, branch: str) -> None:
         unmerged_blobs = git_repo.index.unmerged_blobs()
 
         # Note : only two-way merges supported for now.
-        assert all(len(v) == 3 for k, v in unmerged_blobs.items())
+        if not all(len(v) == 3 for _, v in unmerged_blobs.items()):
+            raise RuntimeError(
+                "Only two-way merges with three blob stages are supported"
+            )
 
         # Ensure the path uses forward slashes, which is compatible with Git's path handling
         if self.review_manager.paths.RECORDS_FILE_GIT in unmerged_blobs:

colrev/ops/search.py

@@ -54,7 +54,12 @@ def _get_search_sources(
                 if s.search_results_path.name in selected_filenames
             ]
 
-        assert len(sources_selected) != 0
+        if len(sources_selected) == 0:
+            raise colrev_exceptions.ParameterError(
+                parameter="select",
+                value=selection_str,
+                options=[str(s.search_results_path) for s in self.sources],
+            )
         return sources_selected
 
     def _remove_forthcoming(

colrev/packages/export_man_prep/src/prep_man_export.py

@@ -81,7 +81,10 @@ def __init__(
         self.logger = logger or logging.getLogger(__name__)
         if "pdf_handling_mode" not in settings:
             settings["pdf_handling_mode"] = "symlink"
-        assert settings["pdf_handling_mode"] in ["symlink", "copy_first_page"]
+        if settings["pdf_handling_mode"] not in ["symlink", "copy_first_page"]:
+            raise ValueError(
+                "pdf_handling_mode must be one of ['symlink', 'copy_first_page']"
+            )
 
         self.settings = self.settings_class(**settings)
 

colrev/packages/importer/src/importer.py

@@ -239,7 +239,10 @@ def _import_records(
             continue
         original_id = duplicate_id_set[0]
         other_id = duplicate_id_set[1]
-        assert other_id.startswith("other_")
+        if not other_id.startswith("other_"):
+            raise ValueError(
+                f"Expected other_id to start with 'other_', got {other_id!r}"
+            )
         print(f"{original_id} <-- {other_id}")
 
         # merge

colrev/packages/open_citations_forward_search/src/open_citations_forward_search.py

@@ -82,7 +82,10 @@ def _validate_source(self) -> None:
 
         self.logger.debug(f"Validate SearchSource {source.search_results_path}")
 
-        assert source.search_type == SearchType.FORWARD_SEARCH
+        if source.search_type != SearchType.FORWARD_SEARCH:
+            raise ValueError(
+                f"Expected search_type {SearchType.FORWARD_SEARCH}, got {source.search_type}"
+            )
 
         if "scope" not in source.search_parameters:
             raise colrev_exceptions.InvalidQueryException(

colrev/packages/pdf_backward_search/src/pdf_backward_search.py

@@ -101,7 +101,10 @@ def _validate_source(self) -> None:
         source = self.search_source
         self.logger.debug(f"Validate SearchSource {source.search_results_path}")
 
-        assert source.search_type == SearchType.BACKWARD_SEARCH
+        if source.search_type != SearchType.BACKWARD_SEARCH:
+            raise ValueError(
+                f"Expected search_type {SearchType.BACKWARD_SEARCH}, got {source.search_type}"
+            )
 
         if "scope" not in source.search_parameters:
             raise colrev_exceptions.InvalidQueryException(
@@ -576,8 +579,14 @@ def add_endpoint(
         if "min_intext_citations" not in params_dict:
             cls._get_params_from_ui(params=params_dict, path=path)
         else:
-            assert params_dict["min_intext_citations"].isdigit()
-            assert params_dict["min_ref_freq"].isdigit()
+            if not params_dict["min_intext_citations"].isdigit():
+                raise ValueError(
+                    f"min_intext_citations must be numeric, got {params_dict['min_intext_citations']!r}"
+                )
+            if not params_dict["min_ref_freq"].isdigit():
+                raise ValueError(
+                    f"min_ref_freq must be numeric, got {params_dict['min_ref_freq']!r}"
+                )
 
         search_source = cls.get_default_source()
         if "min_intext_citations" in params_dict: