tests: add error catch

This commit is contained in:
SWHL committed 2026-10-06 22:24:37 +08:00
1 parent b4000d171b
commit 191bcc7e89
3 files changed
+102 -16

No files matched your search

+32 -7
View File
@@ -1,12 +1,14 @@
# -*- encoding: utf-8 -*-
from collections.abc import Mapping, Sequence
from typing import Any
from omegaconf import OmegaConf
from rapidocr.utils.typings import ModelType, OCRVersion, TaskType
def validate_model_routes(model_routes: Any) -> None:
"""Validate the engine-neutral route schema explicitly."""
if not OmegaConf.is_dict(model_routes):
if not _is_mapping(model_routes):
raise ValueError("Invalid model_routes: expected a mapping")
for path, routes in _iter_model_route_lists(model_routes):
@@ -17,7 +19,8 @@ def _iter_model_route_lists(model_routes: Any):
for version, version_cfg in model_routes.items():
if version == "_defs":
continue
if not OmegaConf.is_dict(version_cfg):
_validate_enum("OCR version", version, OCRVersion)
if not _is_mapping(version_cfg):
raise ValueError(f"Invalid model route version: {version}")
if "aliases" in version_cfg:
_validate_aliases(f"{version}.aliases", version_cfg["aliases"])
@@ -25,18 +28,40 @@ def _iter_model_route_lists(model_routes: Any):
for task, task_cfg in version_cfg.items():
if task == "aliases":
continue
if not OmegaConf.is_dict(task_cfg):
_validate_enum("task", task, TaskType)
if not _is_mapping(task_cfg):
raise ValueError(f"Invalid model route task: {version}.{task}")
if "aliases" in task_cfg:
_validate_aliases(f"{version}.{task}.aliases", task_cfg["aliases"])
for model_type, routes in task_cfg.items():
if model_type != "aliases":
_validate_enum("model type", model_type, ModelType)
yield f"{version}.{task}.{model_type}", routes
def _validate_enum(label: str, value: str, enum_type: Any) -> None:
try:
enum_type(value)
except ValueError as exc:
valid_values = ", ".join(item.value for item in enum_type)
raise ValueError(
f"Invalid {label} {value!r}; expected one of: {valid_values}"
) from exc
def _is_mapping(value: Any) -> bool:
return OmegaConf.is_dict(value) or isinstance(value, Mapping)
def _is_sequence(value: Any) -> bool:
return OmegaConf.is_list(value) or (
isinstance(value, Sequence) and not isinstance(value, (str, bytes))
)
def _validate_aliases(path: str, aliases: Any) -> None:
if not OmegaConf.is_dict(aliases):
if not _is_mapping(aliases):
raise ValueError(f"Invalid aliases: {path}")
for alias, target in aliases.items():
@@ -47,13 +72,13 @@ def _validate_aliases(path: str, aliases: Any) -> None:
def _validate_model_route_list(path: str, routes: Any) -> None:
if not OmegaConf.is_list(routes):
if not _is_sequence(routes):
raise ValueError(f"Invalid model route list: {path}")
seen = set()
seen_languages = set()
for route in routes:
if not OmegaConf.is_dict(route):
if not _is_mapping(route):
raise ValueError(f"Invalid model route entry: {path}")
required = {"lang", "model_key", "supported_langs"}
@@ -72,7 +97,7 @@ def _validate_model_route_list(path: str, routes: Any) -> None:
seen.add(model_key)
supported_langs = route["supported_langs"]
if not OmegaConf.is_list(supported_langs) or not supported_langs:
if not _is_sequence(supported_langs) or not supported_langs:
raise ValueError(f"Empty supported languages: {model_key}")
for lang in supported_langs: