Validator major refactory

This commit is contained in:
jbannon 2022-04-03 07:37:28 +00:00
parent a0187cab41
commit 623f495b60
17 changed files with 351 additions and 265 deletions

View file

@ -1,6 +1,9 @@
import pytest
from ytdl_subscribe.utils.formatter_validator import FormatterValidator
from ytdl_subscribe.validators.base.string_formatter_validator import (
StringFormatterValidator,
)
from ytdl_subscribe.validators.exceptions import ValidationException
@pytest.fixture
@ -13,42 +16,21 @@ def error_message_unequal_regex_matches_str():
return "{variable_names} should only contain lowercase letters and underscores with a single open and close bracket."
@pytest.fixture
def error_message_prefix_generator():
def _error_message_prefix_generator(format_string):
return f"Format string '{format_string}' is invalid: "
return _error_message_prefix_generator
@pytest.fixture
def error_message_unequal_brackets_generator(
error_message_prefix_generator, error_message_unequal_brackets_str
):
def _error_message_unequal_brackets_generator(format_string):
return f"{error_message_prefix_generator(format_string)}{error_message_unequal_brackets_str}"
return _error_message_unequal_brackets_generator
@pytest.fixture
def error_message_unequal_regex_matches_generator(
error_message_prefix_generator, error_message_unequal_regex_matches_str
):
def _error_message_unequal_regex_matches_generator(format_string):
return f"{error_message_prefix_generator(format_string)}{error_message_unequal_regex_matches_str}"
return _error_message_unequal_regex_matches_generator
class TestEntryFormatter(object):
def test_parse(self):
format_string = "Here is my {var_one} and {var_two} 💩"
assert FormatterValidator(format_string).parse() == ["var_one", "var_two"]
assert StringFormatterValidator(
name="test_format_variables", format_string=format_string
).format_variables == ["var_one", "var_two"]
def test_parse_no_variables(self):
format_string = "No vars 💩"
assert FormatterValidator(format_string).parse() == []
assert (
StringFormatterValidator(
name="test_format_variables_empty", format_string=format_string
).format_variables
== []
)
@pytest.mark.parametrize(
"format_string",
@ -60,12 +42,14 @@ class TestEntryFormatter(object):
],
)
def test_parse_fail_uneven_brackets(
self, format_string, error_message_unequal_brackets_generator
self, format_string, error_message_unequal_brackets_str
):
expected_error_msg = error_message_unequal_brackets_generator(format_string)
expected_error_msg = (
f"Validation error in fail: {error_message_unequal_brackets_str}"
)
with pytest.raises(ValueError, match=expected_error_msg):
assert FormatterValidator(format_string).parse()
with pytest.raises(ValidationException, match=expected_error_msg):
_ = StringFormatterValidator(name="fail", format_string=format_string)
@pytest.mark.parametrize(
"format_string",
@ -79,11 +63,11 @@ class TestEntryFormatter(object):
],
)
def test_parse_fail_variable(
self, format_string, error_message_unequal_regex_matches_generator
self, format_string, error_message_unequal_regex_matches_str
):
expected_error_msg = error_message_unequal_regex_matches_generator(
format_string
expected_error_msg = (
f"Validation error in fail: {error_message_unequal_regex_matches_str}"
)
with pytest.raises(ValueError, match=expected_error_msg):
assert FormatterValidator(format_string).parse()
with pytest.raises(ValidationException, match=expected_error_msg):
_ = StringFormatterValidator(name="fail", format_string=format_string)

View file

@ -5,7 +5,9 @@ from typing import Optional
from sanitize_filename import sanitize
from ytdl_subscribe.utils.formatter_validator import FormatterValidator
from ytdl_subscribe.validators.base.string_formatter_validator import (
StringFormatterValidator,
)
class Entry:
@ -104,7 +106,9 @@ class Entry:
if overrides:
entry_dict = dict(entry_dict, **overrides)
field_names = FormatterValidator(format_string).parse()
field_names = StringFormatterValidator(
"TODO: UPDATE", format_string
).format_variables
for field_name in field_names:
if field_name not in entry_dict:

View file

@ -22,7 +22,8 @@ class Subscription(object):
options: dict
Dictionary of ytdl options, specific to the source type
ytdl_opts: dict
Dictionary of options passed directly to ytdl. See `youtube_dl.YoutubeDL.YoutubeDL` for options.
Dictionary of options passed directly to ytdl.
See `youtube_dl.YoutubeDL.YoutubeDL` for options.
post_process: dict
Dictionary of ytdl-subscribe post processing options
overrides: dict

View file

@ -1,41 +0,0 @@
import re
from typing import List
class FormatterValidator:
"""
Ensures user-created formatter strings are valid
"""
FIELDS_VALIDATOR = re.compile(r"{([a-z_]+?)}")
def __init__(self, format_string: str):
self.format_string = format_string
self._error_prefix = f"Format string '{self.format_string}' is invalid:"
def parse(self) -> List[str]:
"""
Returns
-------
List of fields to format
"""
open_bracket_count = self.format_string.count("{")
close_bracket_count = self.format_string.count("}")
if open_bracket_count != close_bracket_count:
raise ValueError(
f"{self._error_prefix} Brackets are reserved for {{variable_names}} "
f"and should contain a single open and close bracket."
)
parsed_fields = re.findall(
FormatterValidator.FIELDS_VALIDATOR, self.format_string
)
if len(parsed_fields) != open_bracket_count:
raise ValueError(
f"{self._error_prefix} {{variable_names}} should only contain "
f"lowercase letters and underscores with a single open and close bracket."
)
return sorted(parsed_fields)

View file

@ -1,10 +0,0 @@
from typing import Any
class BaseValidator:
def __init__(self, name: str, value: Any):
self.name = name
self.value = value
def validate(self) -> "BaseValidator":
return self

View file

@ -0,0 +1,11 @@
from typing import Any
from typing import Optional
from typing import Type
from ytdl_subscribe.validators.base.validator import Validator
from ytdl_subscribe.validators.exceptions import ValidationException
class BoolValidator(Validator):
expected_value_type: Type = bool
expected_value_type_name = "boolean"

View file

@ -0,0 +1,95 @@
from typing import Any
from typing import Optional
from typing import Set
from typing import Type
from typing import TypeVar
from ytdl_subscribe.validators.base.validator import Validator
from ytdl_subscribe.validators.exceptions import ValidationException
T = TypeVar("T", bound=Validator)
class DictValidator(Validator):
expected_value_type = dict
expected_value_type_name = "object" # for non-python users
required_fields: Set[str] = set()
optional_fields: Set[str] = set()
allow_extra_fields = False
def __validate_required_fields_are_present(self):
"""
Raises
-------
ValidationException
If the required fields are not present in the dict
"""
for required_key in self.required_fields:
if required_key not in self.value:
error_msg = (
f"'{self.name}' is missing the required field '{required_key}'"
)
raise ValidationException(error_msg)
def __validate_extra_fields(self):
"""
Raises
-------
ValidationException
If allow_extra_fields=False and non-required/options fields are present
"""
if not self.allow_extra_fields:
for object_key in self.object_keys:
if (
object_key not in self.required_fields
and object_key not in self.optional_fields
):
error_msg = (
f"'{self.name}' contains the field '{object_key}' which is not allowed. "
f"Allowed fields: {', '.join(self.allowed_fields)}"
)
raise ValidationException(error_msg)
def __init__(self, name, value):
super().__init__(name, value)
self.__validate_required_fields_are_present()
self.__validate_extra_fields()
def validate_dict_value(
self, dict_value_name: str, validator: Type[T], default: Optional[Any] = None
) -> T:
value = self.get(object_key=dict_value_name, default=default)
if value is None:
raise self._validation_exception(
f"{dict_value_name} is missing when it should be present."
)
return validator(
name=f"{self.name}.{dict_value_name}",
value=self.get(object_key=dict_value_name, default=default),
)
@property
def dict(self) -> dict:
return self._value
@property
def allowed_fields(self):
return sorted(self.required_fields.union(self.optional_fields))
@property
def object_keys(self):
return sorted(list(self.dict.keys()))
@property
def object_items(self):
return self.dict.items()
def get(self, object_key: str, default: Optional[Any] = None) -> Any:
return self.dict.get(object_key, default)
class DictWithExtraFieldsValidator(DictValidator):
allow_extra_fields = True

View file

@ -1,62 +0,0 @@
from typing import Any
from typing import Optional
from typing import Set
from ytdl_subscribe.validators.base.base_validator import BaseValidator
from ytdl_subscribe.validators.exceptions import ValidationException
class ObjectValidator(BaseValidator):
required_fields: Set[str] = []
optional_fields: Set[str] = []
allow_extra_fields = False
@property
def allowed_fields(self):
return sorted(self.required_fields.union(self.optional_fields))
@property
def object_keys(self):
return sorted(list(self.value.keys()))
@property
def object_items(self):
return self.value.items()
def get(self, object_key: str, default: Optional[Any] = None) -> Any:
return self.value.get(object_key, default)
def validate(self) -> "ObjectValidator":
super().validate()
# Ensure the value is an object
if not isinstance(self.value, dict):
error_msg = f"'{self.name}' must be an object"
if self.required_fields:
error_msg = f"{error_msg} with the required field: {', '.join(self.required_fields)}"
raise ValidationException(error_msg)
# Ensure the object contains all the required fields
for required_key in self.required_fields:
if required_key not in self.value:
error_msg = (
f"'{self.name}' is missing the required field '{required_key}'"
)
raise ValidationException(error_msg)
# If extra fields are not allowed, ensure we only have required or optional keys
if not self.allow_extra_fields:
for object_key in self.object_keys:
if (
object_key not in self.required_fields
and object_key not in self.optional_fields
):
error_msg = (
f"'{self.name}' contains the field '{object_key}' which is not allowed. "
f"Allowed fields: {', '.join(self.allowed_fields)}"
)
raise ValidationException(error_msg)
return self

View file

@ -0,0 +1,64 @@
import re
from keyword import iskeyword
from typing import List
from typing import Optional
from ytdl_subscribe.validators.base.string_validator import StringValidator
from ytdl_subscribe.validators.exceptions import ValidationException
class StringFormatterValidator(StringValidator):
"""
Ensures user-created formatter strings are valid
"""
expected_value_type = str
expected_value_type_name = "format string"
FIELDS_VALIDATOR = re.compile(r"{([a-z_]+?)}")
def __validate_and_get_format_variables(self) -> List[str]:
"""
Returns
-------
list[str]
List of format variables in the format string
Raises
-------
ValidationException
If the format string contains invalid variable formatting
"""
open_bracket_count = self.format_string.count("{")
close_bracket_count = self.format_string.count("}")
if open_bracket_count != close_bracket_count:
raise self._validation_exception(
error_message="Brackets are reserved for {variable_names} and should contain a single open and close bracket."
)
format_variables: List[str] = list(
re.findall(StringFormatterValidator.FIELDS_VALIDATOR, self.format_string)
)
if len(format_variables) != open_bracket_count:
raise self._validation_exception(
"{variable_names} should only contain "
"lowercase letters and underscores with a single open and close bracket."
)
for variable in format_variables:
if iskeyword(variable):
raise self._validation_exception(
f"'{variable}' is a Python keyword and cannot be used as a variable."
)
return format_variables
def __init__(self, name, format_string: str):
super().__init__(name=name, value=format_string)
self.format_variables = self.__validate_and_get_format_variables()
@property
def format_string(self) -> str:
return self._value

View file

@ -0,0 +1,15 @@
from typing import Any
from typing import Optional
from typing import Type
from ytdl_subscribe.validators.base.validator import Validator
from ytdl_subscribe.validators.exceptions import ValidationException
class StringValidator(Validator):
expected_value_type: Type = str
expected_value_type_name = "string"
@property
def value(self) -> str:
return self._value

View file

@ -0,0 +1,41 @@
from typing import Any
from typing import Optional
from typing import Type
from ytdl_subscribe.validators.exceptions import ValidationException
class Validator:
# The python type that value should be
expected_value_type: Type = object
# When raising an error, call the type this value instead of its python name
expected_value_type_name: Optional[str] = None
def __validate_value(self):
"""
Returns
-------
Validation exception to raise when the value's type is not the expected type
"""
if not isinstance(self._value, self.expected_value_type):
expected_value_type_name = self.expected_value_type_name or str(
self.expected_value_type
)
raise self._validation_exception(
error_message=f"should be of type {expected_value_type_name}."
)
def __init__(self, name: str, value: Any):
self.name = name
self._value = value
self.__validate_value()
@property
def value(self) -> object:
return self._value
def _validation_exception(self, error_message: str):
prefix = f"Validation error in {self.name}: "
return ValidationException(f"{prefix}{error_message}")

View file

@ -1,32 +1,21 @@
from typing import Any
from typing import Dict
from typing import Optional
import yaml
from ytdl_subscribe.validators.base.object_validator import ObjectValidator
from ytdl_subscribe.validators.config.preset_validator import PresetValidator
from ytdl_subscribe.validators.base.dict_validator import DictValidator
from ytdl_subscribe.validators.base.dict_validator import DictWithExtraFieldsValidator
from ytdl_subscribe.validators.base.string_validator import StringValidator
class ConfigValidator(ObjectValidator):
class ConfigValidator(DictValidator):
required_fields = {"working_directory", "presets"}
def __init__(self, name: str, value: Any):
super().__init__(name, value)
self.working_directory: Optional[str] = None
self.presets: Optional[Dict[str, PresetValidator]] = None
def validate(self) -> "ConfigValidator":
super().validate()
self.working_directory = self.get("working_directory")
self.presets = {}
for preset_name, preset_obj in self.get("presets").items():
self.presets[preset_name] = PresetValidator(
name=f"{self.name}.presets.{preset_name}", value=preset_obj
).validate()
return self
self.working_directory = self.validate_dict_value(
"working_directory", StringValidator
)
self.presets = self.validate_dict_value("presets", DictWithExtraFieldsValidator)
@classmethod
def from_file_path(cls, config_path) -> "ConfigValidator":
@ -34,4 +23,4 @@ class ConfigValidator(ObjectValidator):
with open(config_path, "r", encoding="utf-8") as file:
config_dict = yaml.safe_load(file)
return ConfigValidator(name="config", value=config_dict).validate()
return ConfigValidator(name="config", value=config_dict)

View file

@ -2,7 +2,7 @@ from typing import Any
from typing import Optional
from ytdl_subscribe.utils.enums import SubscriptionSourceName
from ytdl_subscribe.validators.base.object_validator import ObjectValidator
from ytdl_subscribe.validators.base.dict_validator import DictValidator
from ytdl_subscribe.validators.config.sources.base_source_validator import (
BaseSourceValidator,
)
@ -15,9 +15,14 @@ from ytdl_subscribe.validators.config.sources.youtube_source_validator import (
from ytdl_subscribe.validators.exceptions import ValidationException
class PresetValidator(ObjectValidator):
class PresetValidator(DictValidator):
required_fields = {"post_process"}
optional_fields = {"ytdl_options", "output_path", *SubscriptionSourceName.all()}
optional_fields = {
"ytdl_options",
"output_path",
"overrides",
*SubscriptionSourceName.all(),
}
allow_extra_fields = False
def __init__(self, name: str, value: Any):
@ -25,8 +30,6 @@ class PresetValidator(ObjectValidator):
self.subscription_source: Optional[BaseSourceValidator] = None
self.subscription_source_name: Optional[str] = None
def _validate_subscription_source(self):
# Make sure we only have a single subscription source
for object_key, object_value in self.object_items:
if object_key in SubscriptionSourceName.all() and self.subscription_source:
raise ValidationException(
@ -35,22 +38,19 @@ class PresetValidator(ObjectValidator):
if object_key == SubscriptionSourceName.SOUNDCLOUD:
self.subscription_source_name = SubscriptionSourceName.SOUNDCLOUD
self.subscription_source = SoundcloudSourceValidator(
name=f"{self.name}{object_key}", value=object_value
).validate()
self.subscription_source = self.validate_dict_value(
dict_value_name=object_key,
validator=SoundcloudSourceValidator,
)
elif object_key == SubscriptionSourceName.YOUTUBE:
self.subscription_source_name = SubscriptionSourceName.YOUTUBE
self.subscription_source = YoutubeSourceValidator(
name=f"{self.name}{object_key}", value=object_value
).validate()
self.subscription_source = self.validate_dict_value(
dict_value_name=object_key,
validator=YoutubeSourceValidator,
)
# If subscription source was not set, error
if not self.subscription_source:
raise ValidationException(
f"'{self.name} must have one of the following sources: {SubscriptionSourceName.pretty_all()}"
)
def validate(self) -> "PresetValidator":
super().validate()
self._validate_subscription_source()
return self

View file

@ -2,24 +2,29 @@ from typing import Any
from typing import Optional
from typing import Set
from ytdl_subscribe.validators.base.object_validator import ObjectValidator
from ytdl_subscribe.validators.base.dict_validator import DictValidator
from ytdl_subscribe.validators.base.string_validator import StringValidator
from ytdl_subscribe.validators.exceptions import ValidationException
class BaseSourceValidator(ObjectValidator):
class BaseSourceValidator(DictValidator):
# All media sources must define a download strategy
required_fields = {"download_strategy"}
download_strategies: Set[str] = {}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.download_strategy: Optional[str] = None
def _validate_and_get_download_strategy(self) -> str:
download_strategy = self.validate_dict_value(
dict_value_name="download_strategy",
validator=StringValidator,
)
def validate(self) -> "BaseSourceValidator":
super().validate()
if self.value["download_strategy"] not in self.download_strategies:
raise ValidationException(
f"'download_strategy' must be one of the following: {', '.join(self.download_strategies)}"
if download_strategy.value not in self.download_strategies:
raise self._validation_exception(
f"download_strategy must be one of the following: {', '.join(self.download_strategies)}"
)
return self
return download_strategy.value
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.download_strategy = self._validate_and_get_download_strategy()

View file

@ -1,30 +1,27 @@
from typing import Any
from typing import Optional
from ytdl_subscribe.validators.base.bool_validator import BoolValidator
from ytdl_subscribe.validators.base.string_validator import StringValidator
from ytdl_subscribe.validators.config.sources.base_source_validator import (
BaseSourceValidator,
)
from ytdl_subscribe.validators.exceptions import ValidationException
class SoundcloudSourceValidator(BaseSourceValidator):
optional_fields = {"username", "skip_premiere_tracks"}
download_strategies = {"albums_and_singles"}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.username: Optional[str] = None
self.skip_premiere_tracks: Optional[bool] = None
def validate(self) -> "SoundcloudSourceValidator":
super().validate()
self.skip_premiere_tracks = self.get("skip_premiere_tracks", True)
def __validate_and_set_download_strategy_fields(self):
if self.download_strategy == "albums_and_singles":
self.username = self.value.get("username")
if not self.username or not isinstance(self.username, str):
raise ValidationException(
"Soundcloud download_strategy 'albums_and_singles' requires the field 'username'"
)
self.username = self.validate_dict_value(
dict_value_name="username",
validator=StringValidator,
).value
return self
def __init__(self, name: str, value: dict):
super().__init__(name=name, value=value)
self.skip_premiere_tracks = self.validate_dict_value(
"skip_premiere_tracks", BoolValidator, default=True
).value
self.username: Optional[str] = None
self.__validate_and_set_download_strategy_fields()

View file

@ -1,10 +1,9 @@
from typing import Any
from typing import Optional
from ytdl_subscribe.validators.base.string_validator import StringValidator
from ytdl_subscribe.validators.config.sources.base_source_validator import (
BaseSourceValidator,
)
from ytdl_subscribe.validators.exceptions import ValidationException
class YoutubeSourceValidator(BaseSourceValidator):
@ -13,16 +12,4 @@ class YoutubeSourceValidator(BaseSourceValidator):
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.playlist_id: Optional[str] = None
def validate(self) -> "YoutubeSourceValidator":
super().validate()
if self.download_strategy == "playlist":
self.playlist_id = self.value.get("playlist_id")
if not self.playlist_id or not isinstance(self.playlist_id, str):
raise ValidationException(
"Youtube download_strategy 'playlist' requires the field 'playlist_id' to be a string"
)
return self
self.playlist_id = self.validate_dict_value("playlist_id", StringValidator)

View file

@ -1,7 +1,5 @@
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
import yaml
from mergedeep import mergedeep
@ -10,64 +8,72 @@ from ytdl_subscribe.subscriptions.soundcloud import SoundcloudSubscription
from ytdl_subscribe.subscriptions.subscription import Subscription
from ytdl_subscribe.subscriptions.youtube import YoutubeSubscription
from ytdl_subscribe.utils.enums import SubscriptionSourceName
from ytdl_subscribe.validators.base.dict_validator import DictValidator
from ytdl_subscribe.validators.base.dict_validator import DictWithExtraFieldsValidator
from ytdl_subscribe.validators.base.string_validator import StringValidator
from ytdl_subscribe.validators.config.config_validator import ConfigValidator
from ytdl_subscribe.validators.config.preset_validator import PresetValidator
from ytdl_subscribe.validators.exceptions import ValidationException
class SubscriptionValidator(PresetValidator):
class SubscriptionValidator(DictValidator):
"""
A Subscription is a preset but overrides it with specific values
"""
required_fields = {"preset"}
optional_fields = {"overrides"}.union(
PresetValidator.required_fields, PresetValidator.optional_fields
optional_fields = PresetValidator.required_fields.union(
PresetValidator.optional_fields
)
def __init__(self, config: ConfigValidator, name: str, value: Any):
super().__init__(name, value)
self.config = config
self.overrides: Optional[Dict] = None
self.overrides = self.validate_dict_value(
dict_value_name="overrides",
validator=DictWithExtraFieldsValidator,
default={},
)
def validate(self) -> "SubscriptionValidator":
# If a preset is present, update the
preset_name = self.get("preset")
available_presets = sorted(list(self.config.presets.keys()))
preset_name = self.validate_dict_value(
dict_value_name="preset",
validator=StringValidator,
).value
available_presets = self.config.presets.object_keys
if preset_name not in available_presets:
raise ValidationException(
f"'{self.name}.preset' does not exist in the provided config. "
raise self._validation_exception(
f"'preset '{preset_name}' does not exist in the provided config. "
f"Available presets: {', '.join(available_presets)}"
)
self.overrides = self.get("overrides", {})
# A little hacky, we will override the preset with the contents of this subscription, then validate it
self.value = mergedeep.merge(
self.config.presets[preset_name].value,
self.value,
preset_dict = mergedeep.merge(
self.config.presets.dict[preset_name],
self.dict,
strategy=mergedeep.Strategy.REPLACE,
)
del preset_dict["preset"]
_ = super().validate()
return self
self.preset = PresetValidator(
name=self.name,
value=preset_dict,
)
def to_subscription(self) -> Subscription:
subscription_class = None
if self.subscription_source_name == SubscriptionSourceName.SOUNDCLOUD:
if self.preset.subscription_source_name == SubscriptionSourceName.SOUNDCLOUD:
subscription_class = SoundcloudSubscription
elif self.subscription_source_name == SubscriptionSourceName.YOUTUBE:
elif self.preset.subscription_source_name == SubscriptionSourceName.YOUTUBE:
subscription_class = YoutubeSubscription
else:
raise ValueError("subscription source class not found")
return subscription_class(
name=self.name,
options=self.get(self.subscription_source_name),
ytdl_opts=self.get("ytdl_options"),
post_process=self.get("post_process"),
overrides=self.get("overrides"),
output_path=self.get("output_path"),
options=self.preset.get(self.preset.subscription_source_name),
ytdl_opts=self.preset.get("ytdl_options"),
post_process=self.preset.get("post_process"),
overrides=self.preset.get("overrides"),
output_path=self.preset.get("output_path"),
)
@classmethod
@ -83,7 +89,7 @@ class SubscriptionValidator(PresetValidator):
subscriptions.append(
SubscriptionValidator(
config=config, name=subscription_key, value=subscription_object
).validate()
)
)
return subscriptions