Add coverage pragmas to unsupported data model error messages and remove unused load_yaml_from_path function (#2558)

Author: koxudaxi

src/datamodel_code_generator/__init__.py

@@ -88,12 +88,6 @@ def load_yaml_dict(stream: str | TextIO) -> dict[str, YamlValue]:
     return result
 
 
-def load_yaml_from_path(path: Path, encoding: str) -> YamlValue:
-    """Load YAML content from a file path."""
-    with path.open(encoding=encoding) as f:
-        return load_yaml(f)
-
-
 def load_yaml_dict_from_path(path: Path, encoding: str) -> dict[str, YamlValue]:
     """Load YAML and return as dict from a file path."""
     with path.open(encoding=encoding) as f:

src/datamodel_code_generator/__main__.py

@@ -87,11 +87,11 @@ class Config(BaseModel):
     if PYDANTIC_V2:
         model_config = ConfigDict(arbitrary_types_allowed=True)  # pyright: ignore[reportAssignmentType]
 
-        def get(self, item: str) -> Any:
+        def get(self, item: str) -> Any:  # pragma: no cover
             """Get attribute value by name."""
             return getattr(self, item)
 
-        def __getitem__(self, item: str) -> Any:
+        def __getitem__(self, item: str) -> Any:  # pragma: no cover
             """Get item by key."""
             return self.get(item)
 
@@ -128,7 +128,7 @@ def validate_file(cls, value: Any) -> TextIOBase | None:  # noqa: N805
         if path.is_file():
             return cast("TextIOBase", path.expanduser().resolve().open("rt"))
 
-        msg = f"A file was expected but {value} is not a file."
+        msg = f"A file was expected but {value} is not a file."  # pragma: no cover
         raise Error(msg)  # pragma: no cover
 
     @field_validator(
@@ -151,7 +151,7 @@ def validate_url(cls, value: Any) -> ParseResult | None:  # noqa: N805
             return urlparse(value)
         if value is None:  # pragma: no cover
             return None
-        msg = f"This protocol doesn't support only http/https. --input={value}"
+        msg = f"This protocol doesn't support only http/https. --input={value}"  # pragma: no cover
         raise Error(msg)  # pragma: no cover
 
     # Pydantic 1.5.1 doesn't support each_item=True correctly
@@ -433,7 +433,9 @@ def _get_pyproject_toml_config(source: Path) -> dict[str, Any]:
                 # Convert options from kebap- to snake-case
                 pyproject_config = {k.replace("-", "_"): v for k, v in pyproject_config.items()}
                 # Replace US-american spelling if present (ignore if british spelling is present)
-                if "capitalize_enum_members" in pyproject_config and "capitalise_enum_members" not in pyproject_config:
+                if (
+                    "capitalize_enum_members" in pyproject_config and "capitalise_enum_members" not in pyproject_config
+                ):  # pragma: no cover
                     pyproject_config["capitalise_enum_members"] = pyproject_config.pop("capitalize_enum_members")
                 return pyproject_config
 

src/datamodel_code_generator/model/__init__.py

@@ -133,7 +133,7 @@ def get_data_model_types(
             scalar_model=scalar_class,
             union_model=union_class,
         )
-    msg = f"{data_model_type} is unsupported data model type"
+    msg = f"{data_model_type} is unsupported data model type"  # pragma: no cover
     raise ValueError(msg)  # pragma: no cover
 
 

src/datamodel_code_generator/model/pydantic/base_model.py

@@ -326,7 +326,7 @@ def __init__(  # noqa: PLR0913
             if config_attribute in self.extra_template_data:
                 config_parameters[config_attribute] = self.extra_template_data[config_attribute]
         for data_type in self.all_data_types:
-            if data_type.is_custom_type:
+            if data_type.is_custom_type:  # pragma: no cover
                 config_parameters["arbitrary_types_allowed"] = True
                 break
 

src/datamodel_code_generator/parser/__init__.py

@@ -39,7 +39,7 @@ def get_or_put(
         if default_factory:
             value = self[key] = default_factory(key)
             return value
-        msg = "Not found default and default_factory"
+        msg = "Not found default and default_factory"  # pragma: no cover
         raise ValueError(msg)  # pragma: no cover
 
 

src/datamodel_code_generator/parser/base.py

@@ -574,7 +574,7 @@ def _append_additional_imports(self, additional_imports: list[str] | None) -> No
             additional_imports = []
 
         for additional_import_string in additional_imports:
-            if additional_import_string is None:
+            if additional_import_string is None:  # pragma: no cover
                 continue
             new_import = Import.from_full_path(additional_import_string)
             self.imports.append(new_import)
@@ -730,7 +730,7 @@ def __change_from_import(
                     if imports.use_exact:  # pragma: no cover
                         from_, import_ = exact_import(from_, import_, data_type.reference.short_name)
                     import_ = import_.replace("-", "_")
-                    if (
+                    if (  # pragma: no cover
                         len(model.module_path) > 1
                         and model.module_path[-1].count(".") > 0
                         and not self.treat_dot_as_module
@@ -1033,7 +1033,7 @@ def __collapse_root_models(  # noqa: PLR0912
                             model_field.constraints = ConstraintsBase.merge_constraints(
                                 root_type_field.constraints, model_field.constraints
                             )
-                        if (
+                        if (  # pragma: no cover
                             isinstance(
                                 root_type_field,
                                 pydantic_model.DataModelField,

src/datamodel_code_generator/parser/jsonschema.py

@@ -339,7 +339,7 @@ def __init__(self, **data: Any) -> None:
             """Initialize JsonSchemaObject with extra fields handling."""
             super().__init__(**data)
             self.extras = {k: v for k, v in data.items() if k not in EXCLUDE_FIELD_KEYS}
-            if "const" in data.get(self.__extra_key__, {}):
+            if "const" in data.get(self.__extra_key__, {}):  # pragma: no cover
                 self.extras["const"] = data[self.__extra_key__]["const"]
 
     @cached_property
@@ -1178,7 +1178,7 @@ def parse_item(  # noqa: PLR0911, PLR0912
             )
         if item.ref:
             return self.get_ref_data_type(item.ref)
-        if item.custom_type_path:
+        if item.custom_type_path:  # pragma: no cover
             return self.data_type_manager.get_data_type_from_full_path(item.custom_type_path, is_custom_type=True)
         if item.is_array:
             return self.parse_array_fields(name, item, get_special_path("array", path)).data_type
@@ -1840,7 +1840,7 @@ def _parse_file(
                     try:
                         if definitions := get_model_by_path(raw, split_schema_path):
                             break
-                    except KeyError:
+                    except KeyError:  # pragma: no cover
                         continue
 
                 for key, model in definitions.items():

src/datamodel_code_generator/types.py

@@ -180,29 +180,6 @@ def chain_as_tuple(*iterables: Iterable[T]) -> tuple[T, ...]:
     return tuple(chain(*iterables))
 
 
-@lru_cache
-def _remove_none_from_type(type_: str, split_pattern: Pattern[str], delimiter: str) -> list[str]:
-    """Remove None from a type string and return the remaining types."""
-    types: list[str] = []
-    split_type: str = ""
-    inner_count: int = 0
-    for part in re.split(split_pattern, type_):
-        if part == NONE:
-            continue
-        inner_count += part.count("[") - part.count("]")
-        if split_type:
-            split_type += delimiter
-        if inner_count == 0:
-            if split_type:
-                types.append(f"{split_type}{part}")
-            else:
-                types.append(part)
-            split_type = ""
-            continue
-        split_type += part
-    return types
-
-
 def _remove_none_from_union(type_: str, *, use_union_operator: bool) -> str:  # noqa: PLR0912
     """Remove None from a Union type string, handling nested unions."""
     if use_union_operator:

tests/main/test_main_general.py

@@ -9,6 +9,7 @@
 
 from datamodel_code_generator import (
     DataModelType,
+    Error,
     InputFileType,
     generate,
     snooper_to_methods,
@@ -246,3 +247,28 @@ def test_filename_with_various_control_characters(tmp_path: Path) -> None:
         ), f"System call found for {test_name}"
 
         compile(generated_content, str(output_path), "exec")
+
+
+def test_generate_with_nonexistent_file(tmp_path: Path) -> None:
+    """Test that generating from a nonexistent file raises an error."""
+    nonexistent_file = tmp_path / "nonexistent.json"
+    output_file = tmp_path / "output.py"
+
+    with pytest.raises(Error, match="File not found"):
+        generate(
+            input_=nonexistent_file,
+            output=output_file,
+        )
+
+
+def test_generate_with_invalid_file_format(tmp_path: Path) -> None:
+    """Test that generating from an invalid file format raises an error."""
+    invalid_file = tmp_path / "invalid.txt"
+    invalid_file.write_text("this is not valid json or yaml or anything")
+    output_file = tmp_path / "output.py"
+
+    with pytest.raises(Error, match="Invalid file format"):
+        generate(
+            input_=invalid_file,
+            output=output_file,
+        )