if isinstance(source_value, int) and not isinstance(source_value, bool):
if hasattr(self, "_min") and (source_value < self._min):
raise SchemaException(
- f"The value {source_value} is lower than the allowed minimum {self._min}.", object_path
+ f"value {source_value} is lower than the minimum {self._min}.", object_path
)
if hasattr(self, "_max") and (source_value > self._max):
raise SchemaException(
- f"The value {source_value} is higher than the allowed maximum {self._max}", object_path
+ f"value {source_value} is higher than the maximum {self._max}", object_path
)
self._value = source_value
else:
raise SchemaException(
- f"Unexpected input type for integer - {type(source_value)}."
- " Cause might be invalid format or invalid type.",
+ f"expected integer, got '{type(source_value)}'",
object_path,
)
raise SchemaException(f"'{source_value}' does not match '{self._re.pattern}' pattern", object_path)
else:
raise SchemaException(
- f"Unexpected input type for string pattern - {type(source_value)}."
- " Cause might be invalid format or invalid type.",
+ f"expected string, got '{type(source_value)}'",
object_path,
)
raise SchemaException(f"{type(self._value)} Failed to convert: {self}", object_path)
elif isinstance(source_value, int):
raise SchemaException(
- "We do not accept number without units."
- f" Please convert the value to string an add a unit - {list(type(self)._units.keys())}",
+ f"number without units, please convert to string and add unit - {list(type(self)._units.keys())}",
object_path,
)
else:
raise SchemaException(
- f"Unexpected input type for Unit type - {type(source_value)}."
- " Cause might be invalid format or invalid type.",
+ f"expected number with units in a string, got '{type(source_value)}'.",
object_path,
)
class SubprocessControllerException(KresManagerException):
pass
+
class SubprocessControllerTimeoutException(KresManagerException):
pass
return self._tree_path
def msg(self):
- return f"field {self.where()}: " + super().__str__()
+ return f"[{self.where()}] " + super().__str__()
def recursive_msg(self, indentation_level: int = 0) -> str:
INDENT = indentation_level * "\t"
from knot_resolver_manager.compat.dataclasses import dataclass
from knot_resolver_manager.constants import kres_gc_executable, kresd_cache_dir, kresd_config_file, kresd_executable
from knot_resolver_manager.datamodel.config_schema import KresConfig
-from knot_resolver_manager.kresd_controller.interface import KresID, SubprocessType
from knot_resolver_manager.exceptions import SubprocessControllerException, SubprocessControllerTimeoutException
+from knot_resolver_manager.kresd_controller.interface import KresID, SubprocessType
logger = logging.getLogger(__name__)
if result_state == "timeout":
raise SubprocessControllerTimeoutException(f"systemd job '{job_path}' did not finish in {timeout_sec} seconds")
if result_state != "done":
- raise SubprocessControllerException(f"systemd job '{job_path}' completed with state '{result_state}' instead of expected 'done'")
+ raise SubprocessControllerException(
+ f"systemd job '{job_path}' completed with state '{result_state}' instead of expected 'done'"
+ )
@_wrap_dbus_errors
_wait_for_job_completion(systemd, job)
except SubprocessControllerTimeoutException:
logger.error(
- f"Failed to start transient '{name}'."
- "The start operation did not finish within the expected timeframe"
+ f"Failed to start transient '{name}'." "The start operation did not finish within the expected timeframe"
)
raise
except SubprocessControllerException as e:
except SchemaException as e:
errs.append(e)
- raise SchemaException("failed to parse union type, all variants failed", object_path, child_exceptions=errs)
+ raise SchemaException("could not parse any of the possible variants", object_path, child_exceptions=errs)
# after this, there is no place for a None object
elif obj is None:
# List[T]
elif is_list(cls):
+ if isinstance(obj, str):
+ raise SchemaException("expected list, got string", object_path)
return _validated_list(cls, obj, object_path)
# Tuple[A,B,C,D,...]