feat: rewrite validation, rename built-in vars, bump python version

This commit is contained in:
Igor Ovsyannikov
2026-08-04 21:50:37 +03:00
parent 992829d163
commit 2c6000d56e
12 changed files with 428 additions and 295 deletions
+1 -1
View File
@@ -1 +1 @@
3.13
3.14
+4 -4
View File
@@ -5,7 +5,7 @@ tmp-dir = "/tmp/rustic"
subsystem-name = "rustic"
[settings.globalvars]
# also built-in
# also built-in ara available (BASEDIR=backup-base-dir, BUILDDIR=tmp-dir)
BTRVOL = "snap1"
CONF_BASE = "/home/osmium/backup"
@@ -40,14 +40,14 @@ command = "du -sh $CONF_BASE/local-repo"
[[hooks.after-all]]
kind = "btrfs"
action = "delete"
subvolume = "$BDIR"
subvolume = "$BASEDIR"
[profiles.documents]
paths = ["$BDIR/Documents"]
paths = ["Documents"]
[profiles.code]
paths = ["$BDIR/code"]
paths = ["code"]
exclude-paths = [
".cache",
"venv",
+4 -4
View File
@@ -1,7 +1,7 @@
[project]
name = "lohup"
version = "0.1.0"
description = "Backups done dummy"
description = "Backups done silly"
readme = "README.md"
maintainers = [
{name = "Igor Ovsyannikov"}
@@ -16,7 +16,7 @@ classifiers = [
"Topic :: System :: Archiving :: Backup",
"Topic :: Utilities",
]
requires-python = ">=3.13"
requires-python = ">=3.14"
dependencies = [
"click>=8.2.1",
"humanize>=4.13.0",
@@ -28,8 +28,8 @@ Repository = "https://github.com/kam1sh/lohup"
[dependency-groups]
dev = [
"pyinstaller>=6.15.0",
"pytest>=8.4.1",
"pyinstaller>=6.21.0",
"pytest>=9.1.1",
]
[project.scripts]
+21 -10
View File
@@ -1,20 +1,23 @@
import subprocess as procs
from lohup import config
from lohup import config as tomlconf
from lohup.logger import BasicLogger, LogLevel, LoggerProto
from lohup.restic import Restic
from lohup.rustic import Rustic
class Lohup:
def __init__(self, config_path: str | None, logger: LoggerProto = None):
subsystem: str | None
_config: tomlconf.TomlConfig | None
def __init__(self, config_path: str | None, logger: LoggerProto | None = None):
self._config_path = config_path or "lohup.toml"
self.config = None
self._config = None
self.subsystem = None
self.log = logger or BasicLogger(level=LogLevel.INFO)
def load(self):
self.config = config.TomlConfig.from_file(self._config_path, logger=self.log)
self._config = tomlconf.TomlConfig.from_file(self._config_path)
self.subsystem = self.config.settings.subsystem
if self.subsystem not in ("restic", "rustic"):
raise KeyError(f"Invalid subsystem: {self.subsystem}")
@@ -30,15 +33,17 @@ class Lohup:
def _exechooks(self, hooks: list):
for h in hooks:
match h:
case config.BtrfsHook():
case tomlconf.BtrfsHook():
self._btrfs(h)
case config.CommandHook():
case tomlconf.CommandHook():
procs.check_call(h.command.split())
@staticmethod
def _btrfs(spec: config.BtrfsHook):
def _btrfs(spec: tomlconf.BtrfsHook):
cmd = ["btrfs"]
if spec.action == "snapshot":
if not spec.snapshot:
raise tomlconf.ConfigError("hook: btrfs: snapshot not specified")
cmd += ["subvolume", "snapshot", spec.subvolume, spec.snapshot]
elif spec.action == "delete":
cmd += ["subvolume", "delete", spec.subvolume]
@@ -51,7 +56,7 @@ class Lohup:
default_list = list(filter(lambda x: x.default, self.config.repos.values()))
return default_list.pop() if default_list else None
def _repo_for(self, profile: config.Profile):
def _repo_for(self, profile: tomlconf.Profile):
default = self._default_repo
if profile.repo is not None:
if repo := self.config.repos.get(profile.repo):
@@ -62,7 +67,7 @@ class Lohup:
raise KeyError(f"No repo attached to profile: {profile.name!r}")
return default
def _engine_for(self, repo: config.Repository):
def _engine_for(self, repo: tomlconf.Repository):
match self.subsystem:
case "rustic":
return Rustic(
@@ -105,7 +110,7 @@ class Lohup:
format = "json" if is_json else "text"
return restic.snapshots(format=format)
def _invoke_profile(self, restic, profile: config.Profile):
def _invoke_profile(self, restic, profile: tomlconf.Profile):
with restic as engine:
engine.backup(profile)
self.log.info("Backup created successfully.")
@@ -115,3 +120,9 @@ class Lohup:
if not result:
raise KeyError(f"Unknown backup profile: {name!r}")
return result
@property
def config(self) -> tomlconf.TomlConfig:
if self._config is None:
raise ValueError("Configuration is not available yet.")
return self._config
+2
View File
@@ -1,6 +1,7 @@
import click
import humanize
from datetime import datetime
from typing import no_type_check
from lohup.app import Lohup
from lohup import logger
@@ -48,6 +49,7 @@ def backup_all(obj: Lohup):
@click.option("--repo", help="Lohup repository name", required=True)
@click.option("--raw", "raw_mode", is_flag=True)
@click.pass_obj
@no_type_check
def snapshots(obj: Lohup, repo: str, raw_mode: bool):
result = obj.snapshots(repo=repo, is_json=not raw_mode)
if raw_mode:
+200 -149
View File
@@ -1,17 +1,35 @@
import platform
import tomllib
from pathlib import Path
from dataclasses import dataclass
from pathlib import Path
from lohup.util import catch_errors, ensure_exists, CatcherError, Masked
from lohup.expander import VarExpander
from lohup.logger import LogLevel
from lohup.templater import Templater
from lohup.util import Masked, ensure_exists
from lohup.util import DeepValidator
class ConfigError(ValueError):
pass
class ValidatedTemplater(Templater):
def read(self, item: dict, key: str, validator: DeepValidator, **kwargs) -> str:
return self.expand_validated(
item.get(key, ""), prefix=f"read {key}", validator=validator, **kwargs
)
def expand_validated(
self, line: str, prefix: str, validator: DeepValidator, **kwargs
) -> str:
with validator.inner(prefix) as inner:
value = self.safe_expand(line, **kwargs)
if isinstance(value, str):
return value
for line in value:
inner.error(line)
return ""
@dataclass
class Settings:
backup_base_dir: Path
@@ -20,30 +38,43 @@ class Settings:
subsystem: str
@staticmethod
def load(conf: dict):
with catch_errors() as catch:
basedir = conf.get("backup-base-dir")
basepath = Path(basedir) if basedir else Path.cwd()
tmpdir = conf.get("tmp-dir")
tmp_path = Path("/tmp/lohup")
if not tmpdir:
if platform.system() == "Windows":
tmp_path = Path.home().joinpath("AppData", "Local", "Temp", "lohup")
else:
tmp_path = Path(tmpdir)
settings = Settings(
backup_base_dir=basepath,
globalvars=conf.get("globalvars") or {},
build_dir=tmp_path,
subsystem=conf.get("subsystem-name", "restic"),
)
settings.globalvars["BDIR"] = str(basepath)
settings.globalvars["BUILDDIR"] = str(settings.build_dir)
for key, value in settings.globalvars.items():
if "$" in value:
catch.error(
f"variable {key!r}: nested references in globals are not allowed"
def load(conf: dict, validator: DeepValidator):
basepath = Path.cwd()
if pth := conf.get("backup-base-dir"):
if "$" in pth:
validator.error(
f"backup-base-dir: got ({pth!r}): variables are not allowed"
)
pth = Path(pth)
if not pth.is_absolute:
validator.error(
f"backup-base-dir: got ({pth!r}): only absolute paths are supported"
)
basepath = pth
tmp_path = Path("/tmp/lohup")
if tmpdir := conf.get("tmp-dir"):
if "$" in tmpdir:
validator.error(f"tmp-dir: got {tmpdir!r}: variables are not allowed")
tmp_path = Path(tmpdir)
elif platform.system() == "Windows":
tmp_path = Path.home().joinpath("AppData", "Local", "Temp", "lohup")
settings = Settings(
backup_base_dir=basepath,
globalvars=conf.get("globals") or {},
build_dir=tmp_path,
subsystem=conf.get("subsystem-name", "restic"),
)
for key, value in settings.globalvars.items():
with validator.inner(f"variable {key!r}") as inner:
if not isinstance(value, str):
inner.error(
f"invalid type: {type(value).__qualname__}, expected string"
)
continue
if "$" in value:
inner.error("nested variables in globals are not allowed")
settings.globalvars["BASEDIR"] = str(basepath)
settings.globalvars["BUILDDIR"] = str(settings.build_dir)
return settings
@@ -55,18 +86,24 @@ class LocalRepository:
default: bool
@staticmethod
def load(name: str, conf: dict, expander: VarExpander):
with catch_errors() as catcher:
repo = LocalRepository(
name=name,
path=expander.expand(conf.get("path")),
repo_key_file=Masked(expander.expand(conf.get("repo-key-file", ""))),
default=conf.get("default", False),
)
if not repo.path:
catcher.error("field 'path' not set")
if msg := ensure_exists(repo.repo_key_file, field="repo-key-file"):
catcher.error(msg)
def load(
name: str, conf: dict, templater: ValidatedTemplater, validator: DeepValidator
):
path = conf.get("path")
if not path:
validator.error("path: field is required")
key_file = Masked("")
if value := templater.read(conf, "repo-key-file", validator):
if msg := ensure_exists(value, field="repo-key-file"):
validator.error(msg)
else:
key_file.value = value
repo = LocalRepository(
name=name,
path=templater.read(conf, "path", validator),
repo_key_file=key_file,
default=conf.get("default", False),
)
return repo
@@ -83,31 +120,30 @@ class S3Repository:
default: bool
@staticmethod
def load(name: str, conf: dict, expander: VarExpander):
def load(
name: str, conf: dict, templater: ValidatedTemplater, validator: DeepValidator
):
repo = S3Repository(
name=name,
endpoint=expander.expand(conf.get("endpoint", "https://s3.amazonaws.com")),
endpoint=templater.read(conf, "endpoint", validator),
region=conf.get("region", ""),
access_key_file=Masked(expander.expand(conf.get("access-key-file", ""))),
secret_key_file=Masked(expander.expand(conf.get("secret-key-file", ""))),
repo_key_file=Masked(expander.expand(conf.get("repo-key-file", ""))),
bucket=expander.expand(conf.get("bucket", "")),
path=expander.expand(conf.get("path", "/")),
access_key_file=Masked(templater.read(conf, "access-key-file", validator)),
secret_key_file=Masked(templater.read(conf, "secret-key-file", validator)),
repo_key_file=Masked(templater.read(conf, "repo-key-file", validator)),
bucket=templater.read(conf, "bucket", validator),
path=templater.read(conf, "path", validator) or "/",
default=conf.get("default", False),
)
with catch_errors() as catcher:
if not repo.endpoint:
catcher.error("field 'endpoint': not set")
if not repo.bucket:
catcher.error("field 'bucket': not set")
if value := repo.access_key_file.value:
if msg := ensure_exists(value, field="access-key-file"):
catcher.error(msg)
if value := repo.secret_key_file.value:
if msg := ensure_exists(value, field="secret-key-file"):
catcher.error(msg)
if msg := ensure_exists(repo.repo_key_file, field="repo-key-file"):
catcher.error(msg)
if not repo.bucket:
validator.error("bucket: field is required")
if value := repo.access_key_file.value:
if msg := ensure_exists(value, field="access-key-file"):
validator.error(msg)
if value := repo.secret_key_file.value:
if msg := ensure_exists(value, field="secret-key-file"):
validator.error(msg)
if msg := ensure_exists(repo.repo_key_file, field="repo-key-file"):
validator.error(msg)
return repo
@@ -119,8 +155,12 @@ class CommandHook:
command: str
hook_kind: str
def load(conf: dict, kind: str, expander: VarExpander):
return CommandHook(command=expander.expand(conf.get("command")), hook_kind=kind)
def load(
conf: dict, kind: str, templater: ValidatedTemplater, validator: DeepValidator
):
return CommandHook(
command=templater.read(conf, "command", validator), hook_kind=kind
)
@dataclass
@@ -131,11 +171,13 @@ class BtrfsHook:
hook_kind: str
@staticmethod
def load(conf: dict, kind: str, expander: VarExpander):
def load(
conf: dict, kind: str, templater: ValidatedTemplater, validator: DeepValidator
):
return BtrfsHook(
action=conf.get("action"),
subvolume=expander.expand(conf.get("subvolume")),
snapshot=expander.expand(conf.get("snapshot")),
action=conf.get("action", ""),
subvolume=templater.read(conf, "subvolume", validator),
snapshot=templater.read(conf, "snapshot", validator),
hook_kind=kind,
)
@@ -149,47 +191,42 @@ class HookSet:
after_all: list[Hook]
@staticmethod
def load(conf: dict, expander: VarExpander):
with catch_errors() as catcher:
before_all = []
after_all = []
for hook in conf.get("before-all", []):
match kind := hook.get("kind"):
case "command":
catcher.catch(
lambda: before_all.append(
CommandHook.load(hook, kind=kind, expander=expander)
),
prefix="before-all command:",
)
case "btrfs":
catcher.catch(
lambda: before_all.append(
BtrfsHook.load(hook, kind=kind, expander=expander)
),
prefix="before-all btrfs:",
)
case _:
catcher.error(f"Unsupported before-all: {kind}")
for hook in conf.get("after-all", []):
match kind := hook.get("kind"):
case "command":
catcher.catch(
lambda: after_all.append(
CommandHook.load(hook, kind=kind, expander=expander)
),
prefix="after-all command:",
)
case "btrfs":
catcher.catch(
lambda: after_all.append(
BtrfsHook.load(hook, kind=kind, expander=expander)
),
prefix="after-all btrfs:",
)
case _:
catcher.error(f"after-all: unknown kind {kind!r}")
return HookSet(before_all=before_all, after_all=after_all)
def load(conf: dict, templater: ValidatedTemplater, validator: DeepValidator):
before_all = []
after_all = []
for hook in conf.get("before-all", []):
with validator.inner("before-all") as inner:
if value := HookSet.hook_for(
hook, hook.get("kind"), templater=templater, validator=inner
):
before_all.append(value)
for hook in conf.get("after-all", []):
with validator.inner("after-all") as inner:
if value := HookSet.hook_for(
hook, hook.get("kind"), templater=templater, validator=inner
):
after_all.append(value)
return HookSet(before_all=before_all, after_all=after_all)
@staticmethod
def hook_for(item: dict, kind: str, templater: ValidatedTemplater, validator: DeepValidator):
match kind:
case "command":
with validator.inner("command") as inner:
return CommandHook.load(
item, kind=kind, templater=templater, validator=inner
)
case "btrfs":
with validator.inner("btrfs") as inner:
return BtrfsHook.load(
item, kind=kind, templater=templater, validator=inner
)
case _:
validator.error(f"unsupported kind: {kind}")
@staticmethod
def empty():
return HookSet(before_all=[], after_all=[])
@dataclass
@@ -217,82 +254,96 @@ class TomlConfig:
settings: Settings
repos: dict[str, Repository]
hooks: HookSet
expander: VarExpander
templater: ValidatedTemplater
profiles: dict[str, Profile]
@staticmethod
def from_file(name: str, logger):
def from_file(name: str):
path = Path(name)
try:
with catch_errors(failfast=logger.accepts(LogLevel.VERBOSE)) as catcher:
return TomlConfig._ffile_impl(name, path, catcher)
except CatcherError as e:
raise ConfigError("Failed to parse TOML config") from e
validator = DeepValidator()
result = TomlConfig._ffile_impl(name, path, validator)
if exc := validator.check():
raise ConfigError("Failed to parse TOML config") from exc
return result
@staticmethod
def _ffile_impl(name: str, path: Path, catcher):
def _ffile_impl(name: str, path: Path, validator: DeepValidator):
if not path.exists():
catcher.error(f"Config {name!r} does not exist")
validator.error(f"Config {name!r} does not exist")
return
with path.open("rb") as fp:
conf: dict = tomllib.load(fp)
settings = Settings.load({})
settings = Settings.load({}, validator=validator)
if value := conf.get("settings"):
settings = catcher.catch(lambda: Settings.load(value), prefix="settings:")
if not settings:
return
expander = VarExpander.from_conf(settings)
with validator.inner("settings") as inner:
settings = Settings.load(value, validator=inner)
if not settings:
return
templater = ValidatedTemplater(settings.globalvars)
raw_repos = conf.get("repos", {})
repos = {}
for name, opts in conf.get("repos", {}).items():
error_prefix = f"repo {name!r}:"
match kind := opts.get("kind"):
case "s3":
repos[name] = catcher.catch(
lambda: S3Repository.load(name, opts, expander=expander),
prefix=error_prefix,
)
case "local":
repos[name] = catcher.catch(
lambda: LocalRepository.load(name, opts, expander=expander),
prefix=error_prefix,
)
case None:
catcher.error(f"{error_prefix} repository type not set")
case _:
catcher.error(f"{error_prefix} unsupported kind: {kind}")
if not repos:
catcher.error("no repositories defined")
hooks = None
for name, opts in raw_repos.items():
error_prefix = f"repo {name!r}"
with validator.inner(error_prefix) as inner:
match kind := opts.get("kind"):
case "s3":
repos[name] = S3Repository.load(
name, opts, templater=templater, validator=inner
)
case "local":
repos[name] = LocalRepository.load(
name, opts, templater=templater, validator=inner
)
case None:
inner.error("repository type not set")
case _:
inner.error(f"unsupported kind: {kind!r}")
if not raw_repos:
validator.error("no repositories defined")
hooks = HookSet.empty()
if hook_conf := conf.get("hooks"):
hooks = catcher.catch(
lambda: HookSet.load(hook_conf, expander=expander), prefix="hook:"
)
with validator.inner("hook") as inner:
hooks = HookSet.load(hook_conf, templater=templater, validator=inner)
profiles = {}
for name, opts in conf.get("profiles", {}).items():
args = opts.get("cli-args", [])
if cmd := opts.get("command"):
profiles[name] = CommandProfile(
name, repo=opts.get("repo"), command=cmd, cli_args=args
name,
repo=opts.get("repo"),
command=cmd,
cli_args=args,
)
else:
paths = opts.get("paths")
if not paths:
catcher.error(f"profile {name!r}: no paths or command provided")
validator.error(f"profile {name!r}: no paths or command provided")
continue
profiles[name] = PathsProfile(
resolved: list[Path] = []
for i, raw in enumerate(paths):
raw = templater.expand_validated(raw, f"paths: {i}", inner)
if not raw:
continue
pth = Path(raw)
if not pth.is_absolute():
pth = settings.backup_base_dir.joinpath(pth)
resolved.append(pth)
profile = PathsProfile(
name,
repo=opts.get("repo"),
paths=[expander.expand(x) for x in paths],
paths=resolved,
exclude_paths=[
expander.expand(x) for x in opts.get("exclude-paths", [])
templater.expand_validated(x, "exclude-paths", validator)
for x in opts.get("exclude-paths", [])
],
cli_args=args,
)
profiles[name] = profile
toml = TomlConfig(
settings=settings,
repos=repos,
hooks=hooks,
expander=expander,
templater=templater,
profiles=profiles,
)
return toml
-28
View File
@@ -1,28 +0,0 @@
from dataclasses import dataclass
import re
@dataclass
class VarExpander:
globalvars: dict[str, str]
_PATTERN = re.compile(r"(\$([\w_]+)[^\w_/$]*)+")
def expand(self, text: str | None, extras: dict[str, str] | None = None):
if not text or "$" not in text:
return text
all_vars = self.globalvars.copy()
if extras:
all_vars.update(extras)
match = self._PATTERN.findall(text)
if not match:
return
for var, name in match:
value = all_vars.get(name)
if not value:
raise KeyError(f"variable {var} is not defined")
text = text.replace(var, value)
return text
@staticmethod
def from_conf(settings):
return VarExpander(globalvars=settings.globalvars)
+2 -1
View File
@@ -35,12 +35,13 @@ class Rustic:
opts["bucket"] = self.repo.bucket
opts["root"] = self.repo.path
opts["region"] = self.repo.region
out["repository"]["options"] = opts
out["repository"]["options"] = opts # ty: ignore[invalid-assignment]
case config.LocalRepository():
out["repository"]["repository"] = self.repo.path
self.conf_dir.mkdir(exist_ok=True)
with self.conf_file.open("w") as f:
tomlkit.dump(out, f)
return self.conf_file
def _cmdline(self):
out = [self.binary, "--log-level=warn", "-P", str(self.conf_dir / "rustic")]
+36
View File
@@ -0,0 +1,36 @@
from dataclasses import dataclass
from string import Template
@dataclass
class Templater:
context: dict[str, str]
def require(self, text, extras: dict[str, str] | None = None) -> str:
if text is None:
raise ValueError("not provided")
if not isinstance(text, str):
raise ValueError(f"expected string, got {type(text)}")
return self.expand(text, extras=extras)
def expand(self, text: str, extras: dict[str, str] | None = None) -> str:
value = self.safe_expand(text, extras)
if isinstance(value, list):
raise ExceptionGroup("Errors during templating", [KeyError(x) for x in value])
return value
def safe_expand(self, text: str, extras: dict[str, str] | None = None) -> str | list[str]:
if not text or "$" not in text:
return text
args = self.context.copy()
if extras:
args.update(extras)
errors: list[str] = []
tpl = Template(text)
names = tpl.get_identifiers()
for name in names:
if name not in args:
errors.append(f"variable {name!r} is not defined")
if errors:
return errors
return tpl.substitute(args)
+79 -66
View File
@@ -1,6 +1,7 @@
from pathlib import Path
from dataclasses import dataclass, field
from contextlib import contextmanager
import typing as ty
@dataclass(repr=False)
@@ -11,74 +12,9 @@ class Masked:
return "[masked]"
@contextmanager
def catch_errors(sep="\n", prefix=None, failfast=False):
catcher = ErrorCatcher(prefix=prefix, failfast=failfast)
try:
yield catcher
except:
raise
catcher.verify(sep=sep)
T = ty.TypeVar("T")
@dataclass
class ErrorCatcher:
outer_prefix: str | None = field(default=None)
prefix: str | None = field(default=None)
errorlist: list = field(default_factory=list)
failfast: bool = field(default=False)
def catch(self, func, prefix=None):
try:
return func()
except CatcherError as e:
if e.catcher.prefix:
e.catcher.outer_prefix = prefix
else:
e.catcher.prefix = prefix
self.errorlist.append(e)
except Exception as e:
if self.failfast:
raise
self.errorlist.append(e)
def error(self, msg):
self.errorlist.append(msg)
def verify(self, sep="\n"):
result = self.lines()
if result:
raise CatcherError(
sep.join(self._prefixed(x) for x in result), catcher=self
)
def lines(self):
out = []
extras = []
for err in self.errorlist:
match err:
case str():
out.append(err)
case CatcherError():
extras.extend(err.catcher.lines())
case _:
extras.append(str(err))
out.extend(extras)
return [self._prefixed(x) for x in out]
def _prefixed(self, s):
if self.outer_prefix:
return f"{self.outer_prefix} {self.prefix} {s}"
return f"{self.prefix} {s}" if self.prefix else s
class CatcherError(ValueError):
catcher: ErrorCatcher
def __init__(self, message, catcher=None):
self.catcher = catcher
super().__init__(message)
def ensure_exists(name, expect="file", field=None) -> str | None:
if not name:
@@ -98,3 +34,80 @@ def ensure_exists(name, expect="file", field=None) -> str | None:
if msg is not None and field is not None:
return f"field {field!r}: {msg}"
return msg
@dataclass
class PrefixedException:
prefix: str
value: Exception
def __str__(self):
return f"{self.prefix}: {self.value}"
AnyFail = PrefixedException | Exception | str
@dataclass
class DeepValidator:
prefix: str | None = field(default=None)
errors: list[AnyFail] = field(default_factory=list)
@contextmanager
def inner(self, prefix: str):
value = DeepValidator(prefix)
try:
yield value
except ExceptionGroup as group:
self.errors.extend(group.exceptions)
except Exception as e:
self.error(e)
if fail := value.check():
self.errors.extend(fail.items)
def error(self, msg: AnyFail):
self.errors.append(msg)
def assert_type(self, value: T, expect: type) -> T:
if not isinstance(value, expect):
self.error(f"expected {expect.__qualname__}, got {type(value).__qualname__}")
return value
def lines(self):
return [self._prefixed(x) for x in self.errors]
def check(self) -> ValidationFailed | None:
lines: list[str | PrefixedException] = []
for error in self.errors:
result: str | PrefixedException
match error:
case str():
result = self._prefixed(error)
case PrefixedException(text, exc):
result = PrefixedException(self._prefixed(text), exc)
case Exception():
result = PrefixedException(self.prefix or "", error)
lines.append(result)
if lines:
return ValidationFailed.from_lines("Validation error", lines)
def _prefixed(self, s):
return f"{self.prefix}: {s}" if self.prefix else s
class ValidationFailed(Exception):
exceptions: list[PrefixedException]
messages: list[str]
@classmethod
def from_lines(cls, text, items: list[str | PrefixedException]):
messages = [str(x) for x in items]
excs = [x for x in items if isinstance(x, PrefixedException)]
result = cls(text)
result.messages = messages
result.exceptions = excs
for line in items:
if isinstance(line, str):
result.add_note(line)
return result
@property
def items(self) -> list[str | PrefixedException]:
return self.exceptions + self.messages
+57 -9
View File
@@ -1,6 +1,9 @@
from pathlib import Path
from lohup.app import Lohup
from lohup.config import ConfigError
from lohup.logger import BasicLogger, LogLevel
from lohup.util import DeepValidator, ValidationFailed
import pytest
@@ -10,7 +13,7 @@ toml1 = """
toml2 = """
[[hooks.before-all]]
kind = "unknown"
kind = "???"
foo = "bar"
[repos.local]
@@ -19,29 +22,45 @@ path = "{base}/repo"
repo-key-file = "{pwfile}"
"""
toml_vars = """
[settings.globals]
foo = "$bar"
baz = "1"
bag = 2
[repos.local]
kind = "local"
path = "$foo"
repo-key-file = "$unknown"
"""
class RepoEnvironment:
def __init__(self, tmp_path):
def __init__(self, tmp_path: Path):
self.basepath = tmp_path
self.log = BasicLogger(level=LogLevel.DEBUG)
def write_password(self, value: str = "1"):
def write_password(self, value: str = "1") -> Path:
pth = self.basepath / "password.txt"
pth.write_text("1")
return pth
def write_config(self, value: str) -> Path:
pth = self.basepath / "lohup.toml"
pth.write_text(value)
return pth
def test_error_message(tmp_path):
env = RepoEnvironment(tmp_path)
pwfile = env.write_password()
path = tmp_path / "lohup.toml"
path.write_text(toml1.format(base=tmp_path, pwfile=pwfile))
app = Lohup(config_path=path, logger=env.log)
path.write_text(toml1)
app = Lohup(config_path=str(path), logger=env.log)
msg = "no repositories defined"
with pytest.raises(ConfigError, match="Failed to parse TOML config") as exc:
app.load()
env.log.error(exc.value)
assert str(exc.value.__cause__) == msg
cause = exc.value.__cause__
assert isinstance(cause, ValidationFailed) and msg in cause.messages
def test_hook_error(tmp_path):
@@ -51,8 +70,37 @@ def test_hook_error(tmp_path):
path = tmp_path / "lohup.toml"
path.write_text(toml2.format(base=tmp_path, pwfile=pwfile))
app = Lohup(config_path=path, logger=env.log)
msg = "hook: Unsupported before-all: unknown"
msg = "hook: before-all: unsupported kind: ???"
with pytest.raises(ConfigError, match="Failed to parse TOML config") as exc:
app.load()
env.log.error(exc.value)
assert str(exc.value.__cause__) == msg
cause = exc.value.__cause__
assert isinstance(cause, ValidationFailed) and msg in cause.messages
def test_invalid_vars(tmp_path):
env = RepoEnvironment(tmp_path)
path = env.write_config(toml_vars)
app = Lohup(config_path=str(path), logger=env.log)
lines = [
"settings: variable 'foo': nested variables in globals are not allowed",
"settings: variable 'bag': invalid type: int, expected string",
"repo 'local': read repo-key-file: variable 'unknown' is not defined"
]
with pytest.raises(ConfigError, match="Failed to parse TOML config") as exc:
app.load()
env.log.error(exc.value)
cause = exc.value.__cause__
assert isinstance(cause, ValidationFailed)
for exc in cause.exceptions:
env.log.error(exc.value)
assert set(lines) == set(cause.messages)
def test_validator():
validator = DeepValidator("testing")
with validator.inner("read config") as inner:
with inner.inner("key 'foo'") as children:
children.error("expected string")
res = validator.check()
assert isinstance(res, ValidationFailed)
assert len(res.messages) == 1, "expected error during validation"
assert res.messages[0] == "testing: read config: key 'foo': expected string"
Generated
+22 -23
View File
@@ -1,6 +1,6 @@
version = 1
revision = 3
requires-python = ">=3.13"
requires-python = ">=3.14"
[[package]]
name = "altgraph"
@@ -75,8 +75,8 @@ requires-dist = [
[package.metadata.requires-dev]
dev = [
{ name = "pyinstaller", specifier = ">=6.15.0" },
{ name = "pytest", specifier = ">=8.4.1" },
{ name = "pyinstaller", specifier = ">=6.21.0" },
{ name = "pytest", specifier = ">=9.1.1" },
]
[[package]]
@@ -129,7 +129,7 @@ wheels = [
[[package]]
name = "pyinstaller"
version = "6.15.0"
version = "6.21.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "altgraph" },
@@ -140,37 +140,36 @@ dependencies = [
{ name = "pywin32-ctypes", marker = "sys_platform == 'win32'" },
{ name = "setuptools" },
]
sdist = { url = "https://files.pythonhosted.org/packages/64/17/b2bb4de22650adbeef401fa82a1b43028976547a8728602e4d29735b455e/pyinstaller-6.15.0.tar.gz", hash = "sha256:a48fc4644ee4aa2aa2a35e7b51f496f8fbd7eecf6a2150646bbf1613ad07bc2d", size = 4331521, upload-time = "2025-08-03T18:33:35.709Z" }
sdist = { url = "https://files.pythonhosted.org/packages/d5/4d/ec706c3fcf39e26888c35b39615ff4d5865d184069666c47492cff1fbe50/pyinstaller-6.21.0.tar.gz", hash = "sha256:bb9fab705983e393a2d1cac77d6972513057ad800215fd861dc15ff5272e98fd", size = 4061519, upload-time = "2026-06-13T14:15:06.25Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/24/dd/d5c8a127446adda954f68ea7fac22772f7ab8656ad4b06df396d82574ca9/pyinstaller-6.15.0-py3-none-macosx_10_13_universal2.whl", hash = "sha256:9f00c71c40148cd1e61695b2c6f1e086693d3bcf9bfa22ab513aa4254c3b966f", size = 1016981, upload-time = "2025-08-03T18:31:52.034Z" },
{ url = "https://files.pythonhosted.org/packages/2d/2a/7b50593b419db43e48d9bdeebaac0ff92a5fe035f3c30f87ca3e1650d7e2/pyinstaller-6.15.0-py3-none-manylinux2014_aarch64.whl", hash = "sha256:cbcc8eb77320c60722030ac875883b564e00768fe3ff1721c7ba3ad0e0a277e9", size = 726337, upload-time = "2025-08-03T18:31:57.592Z" },
{ url = "https://files.pythonhosted.org/packages/77/83/7f498fba0154c57eb5fc93eb9680a2dbadb9f780a3389fb85b8d79683378/pyinstaller-6.15.0-py3-none-manylinux2014_i686.whl", hash = "sha256:c33e6302bc53db2df1104ed5566bd980b3e0ee7f18416a6e3caa908c12a54542", size = 737539, upload-time = "2025-08-03T18:32:02.221Z" },
{ url = "https://files.pythonhosted.org/packages/09/d6/e4477feab7c8379fb49e7ec95c82d0a69ad88f6ccc247f76bef3cb0e3432/pyinstaller-6.15.0-py3-none-manylinux2014_ppc64le.whl", hash = "sha256:eb902d0fed3bb1f8b7190dc4df5c11f3b59505767e0d56d1ed782b853938bbf3", size = 735426, upload-time = "2025-08-03T18:32:06.485Z" },
{ url = "https://files.pythonhosted.org/packages/32/7e/ff25648276f15e2e77fc563d36d8cfcd917e077bf2a172420df3588601b4/pyinstaller-6.15.0-py3-none-manylinux2014_s390x.whl", hash = "sha256:b4df862adae7cf1f08eff53c43ace283822447f7f528f72e4f94749062712f15", size = 732210, upload-time = "2025-08-03T18:32:21.667Z" },
{ url = "https://files.pythonhosted.org/packages/db/3d/267a7dddd0647de95d260780050ccd8228ab29d2b9edea54ed1f56800967/pyinstaller-6.15.0-py3-none-manylinux2014_x86_64.whl", hash = "sha256:b9ebf16ed0f99016ae8ae5746dee4cb244848a12941539e62ce2eea1df5a3f95", size = 732194, upload-time = "2025-08-03T18:32:29.536Z" },
{ url = "https://files.pythonhosted.org/packages/4d/61/962b2eb79ef225233e2d6e04600e998935328011dfb2fa775b1dd16b943a/pyinstaller-6.15.0-py3-none-musllinux_1_1_aarch64.whl", hash = "sha256:22193489e6a22435417103f61e7950363bba600ef36ec3ab1487303668c81092", size = 731256, upload-time = "2025-08-03T18:32:36.069Z" },
{ url = "https://files.pythonhosted.org/packages/67/5e/4e20e1c0e5791b09b69bef3ac921fd0cd25551b56879324ad999b92fa045/pyinstaller-6.15.0-py3-none-musllinux_1_1_x86_64.whl", hash = "sha256:18f743069849dbaee3e10900385f35795a5743eabab55e99dcc42f204e40a0db", size = 731148, upload-time = "2025-08-03T18:32:41.269Z" },
{ url = "https://files.pythonhosted.org/packages/88/31/28956c534991f289e2f981c715730b6241e75dc6295737a8cbd050a0cc8c/pyinstaller-6.15.0-py3-none-win32.whl", hash = "sha256:60da8f1b5071766b45c0f607d8bc3d7e59ba2c3b262d08f2e4066ba65f3544a2", size = 1312297, upload-time = "2025-08-03T18:32:50.572Z" },
{ url = "https://files.pythonhosted.org/packages/09/ab/6a45186c7f8e34c422faecd72580116a67d068158c57faa2d2f6d01faa7f/pyinstaller-6.15.0-py3-none-win_amd64.whl", hash = "sha256:cbea297e16eeda30b41c300d6ec2fd2abea4dbd8d8a32650eeec36431c94fcd9", size = 1373091, upload-time = "2025-08-03T18:32:58.133Z" },
{ url = "https://files.pythonhosted.org/packages/5b/86/72159af032b9db36f2470a3b085f79277ec1c38e7e48f8c5dc1ed16dc4e1/pyinstaller-6.15.0-py3-none-win_arm64.whl", hash = "sha256:f43c035621742cf2d19b84308c60e4e44e72c94786d176b8f6adcde351b5bd98", size = 1314305, upload-time = "2025-08-03T18:33:05.557Z" },
{ url = "https://files.pythonhosted.org/packages/0c/4a/53cf98bf66daed012dc9cd78c8203f19a675d696f2fc12afcf8c5049a0e0/pyinstaller-6.21.0-py3-none-macosx_10_13_universal2.whl", hash = "sha256:327d132389f37912609e01be62810cf96b5aa95b613903e4b8692e0d12fb0eda", size = 1052350, upload-time = "2026-06-13T14:13:55.88Z" },
{ url = "https://files.pythonhosted.org/packages/30/83/b591295c352ef464c50b4c6ffff1c4f771d875c9e833f578d1b9f564f6b3/pyinstaller-6.21.0-py3-none-manylinux2014_aarch64.whl", hash = "sha256:7071d4b094d5b40deeef5fa3d3b98a1b846087f7562b49209663d5f9281fe251", size = 748477, upload-time = "2026-06-13T14:14:00.327Z" },
{ url = "https://files.pythonhosted.org/packages/3d/8f/88fff4e403873b1e22286911350e75ff00db014aa08e57045da9d4328993/pyinstaller-6.21.0-py3-none-manylinux2014_i686.whl", hash = "sha256:6b6374d652107dd4a2eeece903ff82bb4045bb5e1006c5a158a6dcdbefe84bf2", size = 760877, upload-time = "2026-06-13T14:14:04.836Z" },
{ url = "https://files.pythonhosted.org/packages/8a/13/f0e48fbdfd1d05d948157121cea8b1b823dcb89efe6934b71fdd8bdb3f0f/pyinstaller-6.21.0-py3-none-manylinux2014_ppc64le.whl", hash = "sha256:4e3108b3f02384560da70e39b8bf22b0ad597d02bd68a40d76ea91c1cfa00cad", size = 759194, upload-time = "2026-06-13T14:14:10.61Z" },
{ url = "https://files.pythonhosted.org/packages/dd/d5/ea7878cf9924ed30d946d8288777424e6d069d94f5bde56b4d0890069664/pyinstaller-6.21.0-py3-none-manylinux2014_s390x.whl", hash = "sha256:697532279f535ad572bda613db4f821540e235c7854ca6da4d3bf0373f4415ee", size = 754979, upload-time = "2026-06-13T14:14:15.226Z" },
{ url = "https://files.pythonhosted.org/packages/9f/09/51b8905714b733bac66dbc041a7821372d70d888d273ae474c4037d4202d/pyinstaller-6.21.0-py3-none-manylinux2014_x86_64.whl", hash = "sha256:605169523a6b5ace39f13dfbff21add9f2bc43df99c7daf9394fefb2c45e8b6f", size = 754812, upload-time = "2026-06-13T14:14:20.264Z" },
{ url = "https://files.pythonhosted.org/packages/4b/43/d77779439d8c6c2e27a77bcfbd1d5cc0f568ebb611bb472b11af81b5f177/pyinstaller-6.21.0-py3-none-musllinux_1_1_aarch64.whl", hash = "sha256:5fa56746c1e76f93634d018502301378a2d0c382553d37d8c3c34ff436c12dd1", size = 753887, upload-time = "2026-06-13T14:14:25.268Z" },
{ url = "https://files.pythonhosted.org/packages/51/8f/c22df1f6837784ac349057ba693f08e7b1ca7a0e06f9c33c63bc6280007b/pyinstaller-6.21.0-py3-none-musllinux_1_1_x86_64.whl", hash = "sha256:42395ec76df8e8120c36b13339d9db8cab83e316a12839ee303cc00fc941bb74", size = 753779, upload-time = "2026-06-13T14:14:29.445Z" },
{ url = "https://files.pythonhosted.org/packages/c9/76/1ce8a27ce62ba8cf3a87c9ce6d575610f4e55d7cb0123e7512fc3f4b921a/pyinstaller-6.21.0-py3-none-win32.whl", hash = "sha256:c6b28d30d8fd99ce162ff3aab5013ed44dbfb747566b1f01b9bed7964d7c14e9", size = 1336462, upload-time = "2026-06-13T14:14:35.785Z" },
{ url = "https://files.pythonhosted.org/packages/c1/fa/ca1d7e5257dd8566a9dfc0dfb02f8a8075eeb53d4b2d3c579f1276759042/pyinstaller-6.21.0-py3-none-win_amd64.whl", hash = "sha256:7fae06c494ce0ebfe6bd3055c0e409def884f63af2e3705d06bd431ad9237fc7", size = 1397487, upload-time = "2026-06-13T14:14:42.328Z" },
{ url = "https://files.pythonhosted.org/packages/dc/75/21b51523ce8d96629b71311775a0a65f5f5a872124ab0de33e5c848f8bff/pyinstaller-6.21.0-py3-none-win_arm64.whl", hash = "sha256:f13c95c9c03fb567217135919f93815c305813126780b0ed6e0123cb8acaf025", size = 1346094, upload-time = "2026-06-13T14:14:48.914Z" },
]
[[package]]
name = "pyinstaller-hooks-contrib"
version = "2025.8"
version = "2026.6"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "packaging" },
{ name = "setuptools" },
]
sdist = { url = "https://files.pythonhosted.org/packages/71/d6/e5b378b7d4add8c879295c531309b0320e9c07a70458665d091760ffdc87/pyinstaller_hooks_contrib-2025.8.tar.gz", hash = "sha256:3402ad41dfe9b5110af134422e37fc5d421ba342c6cb980bd67cb30b7415641c", size = 164214, upload-time = "2025-07-27T16:37:31.943Z" }
sdist = { url = "https://files.pythonhosted.org/packages/94/5b/c9fe0db5e83ee1c39b2258fa21d23b15e1a60786b6c5990ee5074ead8bb6/pyinstaller_hooks_contrib-2026.6.tar.gz", hash = "sha256:bef5002c32f4f50bd55b005da12cff64eca8783e7eaf86a06a62410164bab725", size = 173354, upload-time = "2026-06-08T22:37:16.152Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/48/34/1d973d0dae849683e53fbcda84443ce016f315e6f4dc7605ede4f56a28c3/pyinstaller_hooks_contrib-2025.8-py3-none-any.whl", hash = "sha256:8d0b8cfa0cb689a619294ae200497374234bd4e3994b3ace2a4442274c899064", size = 442346, upload-time = "2025-07-27T16:37:30.268Z" },
{ url = "https://files.pythonhosted.org/packages/e7/31/f2d7343d8ed5f7c4678377886f6ce533e6eaaa131b252ce950114c2a7efa/pyinstaller_hooks_contrib-2026.6-py3-none-any.whl", hash = "sha256:fd13b8ac126b35361175edacd41a0d97080b75dd5f4b594ecefefff969509dd3", size = 457159, upload-time = "2026-06-08T22:37:14.722Z" },
]
[[package]]
name = "pytest"
version = "8.4.1"
version = "9.1.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
@@ -179,9 +178,9 @@ dependencies = [
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/08/ba/45911d754e8eba3d5a841a5ce61a65a685ff1798421ac054f85aa8747dfb/pytest-8.4.1.tar.gz", hash = "sha256:7c67fd69174877359ed9371ec3af8a3d2b04741818c51e5e99cc1742251fa93c", size = 1517714, upload-time = "2025-06-18T05:48:06.109Z" }
sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/29/16/c8a903f4c4dffe7a12843191437d7cd8e32751d5de349d45d3fe69544e87/pytest-8.4.1-py3-none-any.whl", hash = "sha256:539c70ba6fcead8e78eebbf1115e8b589e7565830d7d006a8723f19ac8a0afb7", size = 365474, upload-time = "2025-06-18T05:48:03.955Z" },
{ url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" },
]
[[package]]