Add custom exception class for CLI (#1919)

* Add custom exception class for CLI

* Fixed TCs for config_manager

---------

Signed-off-by: Yunchu Lee <[email protected]>

Author: yunchu

otx/cli/manager/config_manager.py

@@ -16,6 +16,12 @@
 from otx.api.entities.model_template import ModelTemplate, parse_model_template
 from otx.cli.registry import Registry as OTXRegistry
 from otx.cli.utils.config import configure_dataset, override_parameters
+from otx.cli.utils.errors import (
+    CliException,
+    ConfigValueError,
+    FileNotExistError,
+    NotSupportedError,
+)
 from otx.cli.utils.importing import get_otx_root_path
 from otx.cli.utils.parser import gen_param_help, gen_params_dict_from_args
 from otx.core.data.manager.dataset_manager import DatasetManager
@@ -112,7 +118,7 @@ def data_config_file_path(self) -> Path:
         if "data" in self.args and self.args.data:
             if Path(self.args.data).exists():
                 return Path(self.args.data)
-            raise FileNotFoundError(f"Not found: {self.args.data}")
+            raise FileNotExistError(f"Not found: {self.args.data}")
         return self.workspace_root / "data.yaml"
 
     def check_workspace(self) -> bool:
@@ -140,6 +146,8 @@ def configure_template(self, model: str = None) -> None:
         else:
             task_type = self.task_type
             if not task_type and not model:
+                if not hasattr(self.args, "train_data_roots"):
+                    raise ConfigValueError("Can't find the argument 'train_data_roots'")
                 task_type = self.auto_task_detection(self.args.train_data_roots)
             self.template = self._get_template(task_type, model=model)
         self.task_type = self.template.task_type
@@ -149,7 +157,7 @@ def configure_template(self, model: str = None) -> None:
     def _check_rebuild(self):
         """Checking for Rebuild status."""
         if self.args.task and str(self.template.task_type) != self.args.task.upper():
-            raise NotImplementedError("Task Update is not yet supported.")
+            raise NotSupportedError("Task Update is not yet supported.")
         result = False
         if self.args.model and self.template.name != self.args.model.upper():
             print(f"[*] Rebuild model: {self.template.name} -> {self.args.model.upper()}")
@@ -189,7 +197,7 @@ def _get_train_type(self, ignore_args: bool = False) -> str:
             if hasattr(self.args, "train_type") and self.mode in ("build", "train") and self.args.train_type:
                 self.train_type = self.args.train_type.upper()
                 if self.train_type not in TASK_TYPE_TO_SUB_DIR_NAME:
-                    raise ValueError(f"{self.train_type} is not currently supported by otx.")
+                    raise NotSupportedError(f"{self.train_type} is not currently supported by otx.")
             if self.train_type in TASK_TYPE_TO_SUB_DIR_NAME:
                 return self.train_type
 
@@ -202,7 +210,7 @@ def _get_train_type(self, ignore_args: bool = False) -> str:
     def auto_task_detection(self, data_roots: str) -> str:
         """Detect task type automatically."""
         if not data_roots:
-            raise ValueError("Workspace must already exist or one of {task or model or train-data-roots} must exist.")
+            raise CliException("Workspace must already exist or one of {task or model or train-data-roots} must exist.")
         self.data_format = self.dataset_manager.get_data_format(data_roots)
         return self._get_task_type_from_data_format(self.data_format)
 
@@ -225,7 +233,7 @@ def _get_task_type_from_data_format(self, data_format: str) -> str:
                 self.task_type = task_key
                 print(f"[*] Detected task type: {self.task_type}")
                 return task_key
-        raise ValueError(f"Can't find proper task. we are not support {data_format} format, yet.")
+        raise ConfigValueError(f"Can't find proper task. we are not support {data_format} format, yet.")
 
     def auto_split_data(self, data_roots: str, task: str):
         """Automatically Split train data --> train/val dataset."""
@@ -372,7 +380,7 @@ def _get_template(self, task_type: str, model: Optional[str] = None) -> ModelTem
         if model:
             template_lst = [temp for temp in otx_registry.templates if temp.name.lower() == model.lower()]
             if not template_lst:
-                raise ValueError(
+                raise NotSupportedError(
                     f"[*] {model} is not a type supported by OTX {task_type}."
                     f"\n[*] Please refer to 'otx find --template --task {task_type}'"
                 )
@@ -426,7 +434,7 @@ def build_workspace(self, new_workspace_path: Optional[str] = None) -> None:
 
         model_dir = template_dir.absolute() / train_type_rel_path
         if not model_dir.exists():
-            raise ValueError(f"[*] {self.train_type} is not a type supported by OTX {self.task_type}")
+            raise NotSupportedError(f"[*] {self.train_type} is not a type supported by OTX {self.task_type}")
         train_type_dir = self.workspace_root / train_type_rel_path
         train_type_dir.mkdir(exist_ok=True)
 
@@ -479,7 +487,7 @@ def _copy_config_files(self, target_dir: Path, file_name: str, dest_dir: Path) -
                     config = MPAConfig.fromfile(str(target_dir / file_name))
                     config.dump(str(dest_dir / file_name))
                 except Exception as exc:
-                    raise ImportError(f"{self.task_type} requires mmcv-full to be installed.") from exc
+                    raise CliException(f"{self.task_type} requires mmcv-full to be installed.") from exc
             elif file_name.endswith((".yml", ".yaml")):
                 config = OmegaConf.load(str(target_dir / file_name))
                 (dest_dir / file_name).write_text(OmegaConf.to_yaml(config))

otx/cli/utils/errors.py

@@ -0,0 +1,30 @@
+"""Utils for CLI errors."""
+
+# Copyright (C) 2023 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+#
+
+
+class CliException(Exception):
+    """Custom exception class for CLI."""
+
+
+class ConfigValueError(CliException):
+    """Configuration value is not suitable for CLI."""
+
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class NotSupportedError(CliException):
+    """Not supported error."""
+
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class FileNotExistError(CliException):
+    """Not exist given configuration."""
+
+    def __init__(self, message):
+        super().__init__(message)

tests/fuzzing/cli_fuzzing.py

@@ -4,6 +4,7 @@
 from helper import FuzzingHelper
 
 from otx.cli.tools.cli import main as cli_main
+from otx.cli.utils.errors import CliException
 
 
 @atheris.instrument_func
@@ -21,6 +22,8 @@ def fuzz_otx(input_bytes):
         # argparser will throw SystemExit with code 2 when some required arguments are missing
         if e.code != 2:
             raise
+    except CliException:
+        pass
     # some known exceptions can be catched here
     finally:
         sys.argv = backup_argv

tests/unit/cli/manager/test_config_manager.py

@@ -10,6 +10,12 @@
     set_workspace,
 )
 from otx.cli.registry import Registry
+from otx.cli.utils.errors import (
+    CliException,
+    ConfigValueError,
+    FileNotExistError,
+    NotSupportedError,
+)
 from tests.test_suite.e2e_test_system import e2e_pytest_unit
 
 
@@ -319,7 +325,7 @@ def test_data_config_file_path(self, mocker, tmp_dir_path):
         expected_file_path = tmp_dir_path / "data.yaml"
         args = parser.parse_args(["--data", str(expected_file_path)])
         config_manager.args = args
-        with pytest.raises(FileNotFoundError):
+        with pytest.raises(FileNotExistError):
             config_manager.data_config_file_path
 
         mock_exists.return_value = True
@@ -389,7 +395,7 @@ def test__check_rebuild(self, mocker):
         mock_args.template = mock_template
 
         config_manager = ConfigManager(mock_args)
-        with pytest.raises(NotImplementedError):
+        with pytest.raises(NotSupportedError):
             config_manager._check_rebuild()
 
         config_manager.template.task_type = "DETECTION"
@@ -466,13 +472,13 @@ def test__get_train_type(self, mocker):
     def test_auto_task_detection(self, mocker):
         mock_args = mocker.MagicMock()
         config_manager = ConfigManager(args=mock_args)
-        with pytest.raises(ValueError):
+        with pytest.raises(CliException):
             config_manager.auto_task_detection("")
 
         mock_get_data_format = mocker.patch(
             "otx.cli.manager.config_manager.DatasetManager.get_data_format", return_value="Unexpected"
         )
-        with pytest.raises(ValueError):
+        with pytest.raises(ConfigValueError):
             config_manager.auto_task_detection("data/roots")
 
         mock_get_data_format.return_value = "coco"

tox.ini

@@ -147,7 +147,7 @@ deps =
 use_develop = true
 commands =
     coverage erase
-    - coverage run tests/fuzzing/cli_fuzzing.py {posargs:-dict=tests/fuzzing/assets/cli/operations.dict -artifact_prefix={toxworkdir}/ -print_final_stats=1 -atheris_runs=100000}
+    - coverage run tests/fuzzing/cli_fuzzing.py {posargs:-dict=tests/fuzzing/assets/cli/operations.dict -artifact_prefix={toxworkdir}/ -print_final_stats=1 -atheris_runs=500000}
     coverage report --precision=2
     ; coverage html -d {toxworkdir}/htmlcov