plugin.api.validate: implement ValidationError

- Implement `ValidationError` - Inherit from `ValueError` to preserve backwards compatiblity - Allow collecting multiple errors (AnySchema) - Keep an error stack of parent `ValidationError`s or other exceptions - Format error stack when converting error to string - Raise `ValidationError` instead of `ValueError` - Add error contexts where it makes sense - Add schema names to error instances - Add and update tests
streamlink · back-to · May 8, 2022 · May 7, 2022 · May 7, 2022 · May 7, 2022
commit a59217b91a6049f5f139eb6d5a2a65a42a7db318
diff --git a/src/streamlink/plugin/api/validate/__init__.py b/src/streamlink/plugin/api/validate/__init__.py
@@ -1,3 +1,4 @@
+from streamlink.plugin.api.validate._exception import ValidationError  # noqa: F401
 # noinspection PyPep8Naming,PyShadowingBuiltins
 from streamlink.plugin.api.validate._schemas import (  # noqa: I101, F401
     SchemaContainer,

diff --git a/src/streamlink/plugin/api/validate/_exception.py b/src/streamlink/plugin/api/validate/_exception.py
@@ -0,0 +1,53 @@
+from textwrap import indent
+
+
+class ValidationError(ValueError):
+    def __init__(self, *errors, schema=None, context: "ValidationError" = None):
+        self.schema = schema
+        self.errors = errors
+        self.context = context
+
+    def _get_schema_name(self) -> str:
+        if not self.schema:
+            return ""
+        if type(self.schema) is str:
+            return f"({self.schema})"
+        return f"({self.schema.__name__})"
+
+    def __str__(self):
+        cls = self.__class__
+        ret = []
+        seen = set()
+
+        def append(indentation, error):
+            if error:
+                ret.append(indent(f"{error}", indentation))
+
+        def add(level, error):
+            indentation = "  " * level
+
+            if error in seen:
+                append(indentation, "...")
+                return
+            seen.add(error)
+
+            for err in error.errors:
+                if not isinstance(err, cls):
+                    append(indentation, f"{err}")
+                else:
+                    append(indentation, f"{err.__class__.__name__}{err._get_schema_name()}:")
+                    add(level + 1, err)
+
+            context = error.context
+            if context:
+                if not isinstance(context, cls):
+                    append(indentation, "Context:")
+                    append(f"{indentation}  ", context)
+                else:
+                    append(indentation, f"Context{context._get_schema_name()}:")
+                    add(level + 1, context)
+
+        append("", f"{cls.__name__}{self._get_schema_name()}:")
+        add(1, self)
+
+        return "\n".join(ret)
diff --git a/src/streamlink/plugin/api/validate/_validate.py b/src/streamlink/plugin/api/validate/_validate.py
@@ -6,6 +6,7 @@
 from lxml.etree import Element, iselement
 
 from streamlink.exceptions import PluginError
+from streamlink.plugin.api.validate._exception import ValidationError
 from streamlink.plugin.api.validate._schemas import (
     AllSchema,
     AnySchema,
@@ -27,7 +28,7 @@ class Schema(AllSchema):
     def validate(self, value, name="result", exception=PluginError):
         try:
             return validate(self, value)
-        except ValueError as err:
+        except ValidationError as err:
             raise exception(f"Unable to validate {name}: {err}")
 
 
@@ -37,15 +38,15 @@ def validate(self, value, name="result", exception=PluginError):
 @singledispatch
 def validate(schema, value):
     if schema != value:
-        raise ValueError(f"{value!r} does not equal {schema!r}")
+        raise ValidationError(f"{value!r} does not equal {schema!r}", schema="equality")
 
     return value
 
 
 @validate.register(type)
 def _validate_type(schema, value):
     if not isinstance(value, schema):
-        raise ValueError(f"Type of {value!r} should be '{schema.__name__}', but is '{type(value).__name__}'")
+        raise ValidationError(f"Type of {value!r} should be '{schema.__name__}', but is '{type(value).__name__}'", schema=type)
 
     return value
 
@@ -77,24 +78,32 @@ def _validate_dict(schema, value):
 
         if type(key) in (type, AllSchema, AnySchema, TransformSchema, UnionSchema):
             for subkey, subvalue in value.items():
-                new[validate(key, subkey)] = validate(subschema, subvalue)
+                try:
+                    newkey = validate(key, subkey)
+                except ValidationError as err:
+                    raise ValidationError("Unable to validate key", schema=dict, context=err)
+                try:
+                    newvalue = validate(subschema, subvalue)
+                except ValidationError as err:
+                    raise ValidationError("Unable to validate value", schema=dict, context=err)
+                new[newkey] = newvalue
             break
-        else:
-            if key not in value:
-                raise ValueError(f"Key '{key}' not found in {value!r}")
 
-            try:
-                new[key] = validate(subschema, value[key])
-            except ValueError as err:
-                raise ValueError(f"Unable to validate key '{key}': {err}")
+        if key not in value:
+            raise ValidationError(f"Key '{key}' not found in {value!r}", schema=dict)
+
+        try:
+            new[key] = validate(subschema, value[key])
+        except ValidationError as err:
+            raise ValidationError(f"Unable to validate value of key '{key}'", schema=dict, context=err)
 
     return new
 
 
 @validate.register(abc.Callable)
 def _validate_callable(schema: abc.Callable, value):
     if not schema(value):
-        raise ValueError(f"{schema.__name__}({value!r}) is not true")
+        raise ValidationError(f"{schema.__name__}({value!r}) is not true", schema=abc.Callable)
 
     return value
 
@@ -113,11 +122,10 @@ def _validate_anyschema(schema: AnySchema, value):
     for subschema in schema.schema:
         try:
             return validate(subschema, value)
-        except ValueError as err:
+        except ValidationError as err:
             errors.append(err)
-    else:
-        err = " or ".join(map(str, errors))
-        raise ValueError(err)
+
+    raise ValidationError(*errors, schema=AnySchema)
 
 
 @validate.register(TransformSchema)
@@ -144,21 +152,25 @@ def _validate_getitemschema(schema: GetItemSchema, value):
     except (KeyError, IndexError):
         # only return default value on last item in nested lookup
         if idx < len(item) - 1:
-            raise ValueError(f"Item \"{key}\" was not found in object \"{value}\"")
+            raise ValidationError(f"Item \"{key}\" was not found in object \"{value}\"", schema=GetItemSchema)
         return schema.default
     except (TypeError, AttributeError) as err:
-        raise ValueError(err)
+        raise ValidationError(f"Could not get key \"{key}\" from object \"{value}\"", schema=GetItemSchema, context=err)
 
 
 @validate.register(AttrSchema)
 def _validate_attrschema(schema: AttrSchema, value):
     new = copy(value)
 
-    for key, schema in schema.schema.items():
+    for key, subschema in schema.schema.items():
         if not hasattr(value, key):
-            raise ValueError(f"Attribute \"{key}\" not found on object \"{value}\"")
+            raise ValidationError(f"Attribute \"{key}\" not found on object \"{value}\"", schema=AttrSchema)
+
+        try:
+            value = validate(subschema, getattr(value, key))
+        except ValidationError as err:
+            raise ValidationError(f"Could not validate attribute \"{key}\"", schema=AttrSchema, context=err)
 
-        value = validate(schema, getattr(value, key))
         setattr(new, key, value)
 
     return new
@@ -175,26 +187,26 @@ def _validate_xmlelementschema(schema: XmlElementSchema, value):
     if schema.tag is not None:
         try:
             tag = validate(schema.tag, value.tag)
-        except ValueError as err:
-            raise ValueError(f"Unable to validate XML tag: {err}")
+        except ValidationError as err:
+            raise ValidationError("Unable to validate XML tag", schema=XmlElementSchema, context=err)
 
     if schema.attrib is not None:
         try:
             attrib = validate(schema.attrib, dict(value.attrib))
-        except ValueError as err:
-            raise ValueError(f"Unable to validate XML attributes: {err}")
+        except ValidationError as err:
+            raise ValidationError("Unable to validate XML attributes", schema=XmlElementSchema, context=err)
 
     if schema.text is not None:
         try:
             text = validate(schema.text, value.text)
-        except ValueError as err:
-            raise ValueError(f"Unable to validate XML text: {err}")
+        except ValidationError as err:
+            raise ValidationError("Unable to validate XML text", schema=XmlElementSchema, context=err)
 
     if schema.tail is not None:
         try:
             tail = validate(schema.tail, value.tail)
-        except ValueError as err:
-            raise ValueError(f"Unable to validate XML tail: {err}")
+        except ValidationError as err:
+            raise ValidationError("Unable to validate XML tail", schema=XmlElementSchema, context=err)
 
     new = Element(tag, attrib)
     new.text = text
@@ -214,7 +226,10 @@ def _validate_uniongetschema(schema: UnionGetSchema, value):
 
 @validate.register(UnionSchema)
 def _validate_unionschema(schema: UnionSchema, value):
-    return validate_union(schema.schema, value)
+    try:
+        return validate_union(schema.schema, value)
+    except ValidationError as err:
+        raise ValidationError("Could not validate union", schema=UnionSchema, context=err)
 
 
 # ----
@@ -223,7 +238,7 @@ def _validate_unionschema(schema: UnionSchema, value):
 # noinspection PyUnusedLocal
 @singledispatch
 def validate_union(schema, value):
-    raise ValueError(f"Invalid union type: {type(schema).__name__}")
+    raise ValidationError(f"Invalid union type: {type(schema).__name__}")
 
 
 @validate_union.register(dict)
@@ -236,11 +251,11 @@ def _validate_union_dict(schema, value):
 
         try:
             new[key] = validate(schema, value)
-        except ValueError as err:
+        except ValidationError as err:
             if is_optional:
                 continue
 
-            raise ValueError(f"Unable to validate union '{key}': {err}")
+            raise ValidationError(f"Unable to validate union \"{key}\"", schema=dict, context=err)
 
     return new
 

diff --git a/src/streamlink/plugin/api/validate/_validators.py b/src/streamlink/plugin/api/validate/_validators.py
@@ -3,6 +3,7 @@
 
 from lxml.etree import iselement
 
+from streamlink.plugin.api.validate._exception import ValidationError
 from streamlink.plugin.api.validate._schemas import AllSchema, AnySchema, TransformSchema
 from streamlink.plugin.api.validate._validate import validate
 from streamlink.utils.parse import (
@@ -22,7 +23,7 @@ def validator_length(number: int) -> Callable[[str], bool]:
 
     def min_len(value):
         if not len(value) >= number:
-            raise ValueError(f"Minimum length is {number} but value is {len(value)}")
+            raise ValidationError(f"Minimum length is {number}, but value is {len(value)}", schema="length")
 
         return True
 
@@ -37,7 +38,7 @@ def validator_startswith(string: str) -> Callable[[str], bool]:
     def starts_with(value):
         validate(str, value)
         if not value.startswith(string):
-            raise ValueError(f"'{value}' does not start with '{string}'")
+            raise ValidationError(f"'{value}' does not start with '{string}'", schema="startswith")
 
         return True
 
@@ -52,7 +53,7 @@ def validator_endswith(string: str) -> Callable[[str], bool]:
     def ends_with(value):
         validate(str, value)
         if not value.endswith(string):
-            raise ValueError(f"'{value}' does not end with '{string}'")
+            raise ValidationError(f"'{value}' does not end with '{string}'", schema="endswith")
 
         return True
 
@@ -67,7 +68,7 @@ def validator_contains(string: str) -> Callable[[str], bool]:
     def contains_str(value):
         validate(str, value)
         if string not in value:
-            raise ValueError(f"'{value}' does not contain '{string}'")
+            raise ValidationError(f"'{value}' does not contain '{string}'", schema="contains")
 
         return True
 
@@ -87,16 +88,16 @@ def check_url(value):
         validate(str, value)
         parsed = urlparse(value)
         if not parsed.netloc:
-            raise ValueError(f"'{value}' is not a valid URL")
+            raise ValidationError(f"'{value}' is not a valid URL", schema="url")
 
         for name, schema in attributes.items():
             if not hasattr(parsed, name):
-                raise ValueError(f"Invalid URL attribute '{name}'")
+                raise ValidationError(f"Invalid URL attribute '{name}'", schema="url")
 
             try:
                 validate(schema, getattr(parsed, name))
-            except ValueError as err:
-                raise ValueError(f"Unable to validate URL attribute '{name}': {err}")
+            except ValidationError as err:
+                raise ValidationError(f"Unable to validate URL attribute '{name}'", schema="url", context=err)
 
         return True
 
@@ -185,7 +186,7 @@ def xpath_find(value):
         validate(iselement, value)
         value = value.find(xpath)
         if value is None:
-            raise ValueError(f"XPath '{xpath}' did not return an element")
+            raise ValidationError(f"XPath '{xpath}' did not return an element", schema="xml_find")
 
         return validate(iselement, value)
 
@@ -244,28 +245,28 @@ def validator_parse_json(*args, **kwargs) -> TransformSchema:
     Parse JSON data via the :func:`streamlink.utils.parse.parse_json` utility function.
     """
 
-    return TransformSchema(_parse_json, *args, **kwargs, exception=ValueError, schema=None)
+    return TransformSchema(_parse_json, *args, **kwargs, exception=ValidationError, schema=None)
 
 
 def validator_parse_html(*args, **kwargs) -> TransformSchema:
     """
     Parse HTML data via the :func:`streamlink.utils.parse.parse_html` utility function.
     """
 
-    return TransformSchema(_parse_html, *args, **kwargs, exception=ValueError, schema=None)
+    return TransformSchema(_parse_html, *args, **kwargs, exception=ValidationError, schema=None)
 
 
 def validator_parse_xml(*args, **kwargs) -> TransformSchema:
     """
     Parse XML data via the :func:`streamlink.utils.parse.parse_xml` utility function.
     """
 
-    return TransformSchema(_parse_xml, *args, **kwargs, exception=ValueError, schema=None)
+    return TransformSchema(_parse_xml, *args, **kwargs, exception=ValidationError, schema=None)
 
 
 def validator_parse_qsd(*args, **kwargs) -> TransformSchema:
     """
     Parse a query string via the :func:`streamlink.utils.parse.parse_qsd` utility function.
     """
 
-    return TransformSchema(_parse_qsd, *args, **kwargs, exception=ValueError, schema=None)
+    return TransformSchema(_parse_qsd, *args, **kwargs, exception=ValidationError, schema=None)