feat: rewrite snapshot command output, update README

This commit is contained in:
Igor Ovsyannikov
2026-08-07 20:29:30 +03:00
parent 2c6000d56e
commit 4e632b87df
13 changed files with 295 additions and 147 deletions
+12 -4
View File
@@ -2,7 +2,7 @@
(_loh-up - backup for lohs[1]_) (_loh-up - backup for lohs[1]_)
Lohup - backups done dummy. Lohup - backups done silly.
## Development status ## Development status
@@ -10,11 +10,19 @@ Early alpha. Works on my machines.
## Installation ## Installation
[Restic](https://restic.net) and [uv](https://docs.astral.sh/uv/) for Python required. Or just Python 3.13 and restic. Python 3.14 is required. You can get it using [uv](https://github.com/astral-sh/uv).
Also you will need one of backup backends.
[Restic](https://github.com/restic/restic) and [Rustic](https://github.com/rustic-rs/rustic) are supported.
```bash ```bash
apt update && apt install -y restic apt update && apt install -y restic
# or download rustic binary from https://github.com/rustic-rs/rustic/releases/tag/v0.11.3
# get using uv
uv tool install git+https://github.com/kam1sh/lohup uv tool install git+https://github.com/kam1sh/lohup
# or by pip
pip install
``` ```
## Usage ## Usage
@@ -33,7 +41,7 @@ lohup snapshots --repo cloud
## Core features ## Core features
* Configuration in TOML * Configuration in TOML
* Restic as a backup driver * Restic/Rustic as a backup driver
* Backup hooks, such as create/remove btrfs filesystem snapshot * Backup hooks, such as create/remove btrfs filesystem snapshot
* Environment variables * Environment variables
@@ -58,4 +66,4 @@ uv run pytest --tb=short
uv run pyinstaller -F lohup-cli.spec uv run pyinstaller -F lohup-cli.spec
``` ```
[1] in my native language loh (лох) means "dummy" or "looser" [1] in my native language loh (лох) means "dummy" or "loser". So, yeah, backups are for losers.
+1
View File
@@ -28,6 +28,7 @@ Repository = "https://github.com/kam1sh/lohup"
[dependency-groups] [dependency-groups]
dev = [ dev = [
"isort>=8.0.1",
"pyinstaller>=6.21.0", "pyinstaller>=6.21.0",
"pytest>=9.1.1", "pytest>=9.1.1",
] ]
+11 -6
View File
@@ -1,7 +1,7 @@
import subprocess as procs import subprocess as procs
from lohup import config as tomlconf from lohup import config as tomlconf
from lohup.logger import BasicLogger, LogLevel, LoggerProto from lohup.logger import BasicLogger, LoggerProto, LogLevel
from lohup.restic import Restic from lohup.restic import Restic
from lohup.rustic import Rustic from lohup.rustic import Rustic
@@ -22,9 +22,11 @@ class Lohup:
if self.subsystem not in ("restic", "rustic"): if self.subsystem not in ("restic", "rustic"):
raise KeyError(f"Invalid subsystem: {self.subsystem}") raise KeyError(f"Invalid subsystem: {self.subsystem}")
def invoke_direct(self, repo: str, args: tuple[str, ...]): def invoke_direct(self, repo: str | None, args: tuple[str, ...]):
spec = self.config.repos.get(repo) if repo else self._default_repo spec = self.config.repos.get(repo) if repo else self._default_repo
if spec is None: if spec is None:
if repo is None:
raise KeyError("No repository provided and there is no default one")
raise KeyError(f"Unknown repo: {repo}") raise KeyError(f"Unknown repo: {repo}")
engine = self._engine_for(spec) engine = self._engine_for(spec)
with engine: with engine:
@@ -52,7 +54,7 @@ class Lohup:
procs.check_call(cmd) procs.check_call(cmd)
@property @property
def _default_repo(self): def _default_repo(self) -> tomlconf.Repository | None:
default_list = list(filter(lambda x: x.default, self.config.repos.values())) default_list = list(filter(lambda x: x.default, self.config.repos.values()))
return default_list.pop() if default_list else None return default_list.pop() if default_list else None
@@ -102,13 +104,16 @@ class Lohup:
self._exechooks(self.config.hooks.after_all) self._exechooks(self.config.hooks.after_all)
self.log.info("Finished!") self.log.info("Finished!")
def snapshots(self, repo: str, is_json=False): def snapshots(self, repo: str | None, is_json=False, profile: str | None = None):
spec = self.config.repos.get(repo) if repo else self._default_repo spec = self.config.repos.get(repo) if repo else self._default_repo
if spec is None: if spec is None:
if repo is None:
raise KeyError("No repository provided and there is no default one")
raise KeyError(f"Unknown repo: {repo}") raise KeyError(f"Unknown repo: {repo}")
restic = Restic(spec, log=self.log) engine = self._engine_for(spec)
format = "json" if is_json else "text" format = "json" if is_json else "text"
return restic.snapshots(format=format) with engine:
return engine.snapshots(format=format, profile=profile)
def _invoke_profile(self, restic, profile: tomlconf.Profile): def _invoke_profile(self, restic, profile: tomlconf.Profile):
with restic as engine: with restic as engine:
+34 -49
View File
@@ -1,10 +1,10 @@
import click
import humanize
from datetime import datetime
from typing import no_type_check from typing import no_type_check
import click
import humanize
from lohup import logger, engine
from lohup.app import Lohup from lohup.app import Lohup
from lohup import logger
from lohup.config import ConfigError from lohup.config import ConfigError
@@ -22,10 +22,10 @@ def cli(ctx, config):
@cli.command() @cli.command()
@click.option("--repo", help="Lohup repository name", required=True) @click.option("--repo", help="Lohup repository name")
@click.argument("args", nargs=-1) @click.argument("args", nargs=-1)
@click.pass_obj @click.pass_obj
def restic(obj: Lohup, repo, args: tuple[str, ...]): def restic(obj: Lohup, repo: str | None, args: tuple[str, ...]):
""" """
Pass command to restic Pass command to restic
""" """
@@ -46,59 +46,44 @@ def backup_all(obj: Lohup):
@cli.command() @cli.command()
@click.option("--repo", help="Lohup repository name", required=True) @click.option("--repo", help="Lohup repository name")
@click.option("--raw", "raw_mode", is_flag=True) @click.option("--raw", "raw_mode", is_flag=True)
@click.option("--profile", help="Filter by profile name")
@click.pass_obj @click.pass_obj
@no_type_check @no_type_check
def snapshots(obj: Lohup, repo: str, raw_mode: bool): def snapshots(obj: Lohup, repo: str | None, raw_mode: bool, profile: str | None):
result = obj.snapshots(repo=repo, is_json=not raw_mode) result = obj.snapshots(repo=repo, is_json=not raw_mode, profile=profile)
if raw_mode: if raw_mode:
return click.echo(result, nl=False) return click.echo(result, nl=False)
# time, parent?, tree, paths, hostname, uid, gid groups: dict[str, list[engine.Snapshot]] = result
# tags, version, summary, id, short_id names = sorted(groups)
snapshots = [] for name in names:
for snap in result: raw_items = groups[name]
name = "-".join(snap["tags"]) click.echo(
dt = datetime.fromisoformat(snap["time"]) "Profile {} ({} items):".format(
delta = humanize.naturaltime(datetime.now()) click.style(name, fg="blue", italic=True), len(raw_items)
summary = snap["summary"]
end = datetime.fromisoformat(summary["backup_end"])
changes = summary["files_new"] + summary["files_changed"]
total_files = summary["total_files_processed"]
changes_ratio = changes / total_files if total_files else 0
snapshots.append(
dict(
id=snap["short_id"],
name=name,
when_started=dt,
started_human=delta,
hostname=snap["hostname"],
duration=end - dt,
changes=changes,
changes_ratio=changes_ratio,
size_comp=summary["data_added_packed"],
size_raw=summary["data_added"],
processed=summary["total_bytes_processed"],
) )
) )
snapshots.sort(key=lambda x: x["when_started"]) items = raw_items.copy()
for snap in snapshots: items.sort(key=lambda x: x.time, reverse=True)
name = click.style(snap["name"], fg="blue", italic=True) for snap in items:
click.echo(f"Snapshot {snap['id']} ({name}):") click.echo(f"\tSnapshot {snap.id}:")
started = humanize.naturaltime(snap["when_started"]) started = humanize.naturaltime(snap.time)
text = click.style(started, fg="blue") text = click.style(started, fg="blue")
took = humanize.naturaldelta(snap["duration"]) took = humanize.naturaldelta(snap.duration)
click.echo(f"\tStarted: {text} (took {took})") click.echo(f"\t\tStarted: {text} (took {took})")
ratio = snap["changes_ratio"] * 100 ratio = snap.changes_ratio * 100
ratio = click.style(f"{ratio:.1f}%", fg="blue") ratio = click.style(f"{ratio:.1f}%", fg="blue")
click.echo(f"\tFiles changed (since previous): {snap['changes']} ({ratio})") click.echo(
size = humanize.naturalsize(snap["size_raw"], binary=True) f"\t\tFiles changed (since previous): {snap.total_changes} ({ratio})"
compressed = humanize.naturalsize(snap["size_comp"], binary=True) )
total_bytes = snap["processed"] uncompressed = humanize.naturalsize(snap.size, binary=True)
ratio = snap["size_raw"] / total_bytes if total_bytes else 0 compressed = humanize.naturalsize(snap.size_compressed, binary=True)
total_bytes = snap.bytes_read
ratio = snap.size / total_bytes if total_bytes else 0
ratio = click.style(f"{ratio:.1f}%", fg="blue") ratio = click.style(f"{ratio:.1f}%", fg="blue")
click.echo(f"\tDiff size: {compressed}, unpacked: {size} ({ratio})") click.echo(f"\t\tDiff size: {compressed=!s}, {uncompressed=!s} ({ratio})")
click.echo(f"\tHost: {snap['hostname']}") click.echo(f"\t\tHost: {snap.hostname}")
click.echo() click.echo()
+6 -5
View File
@@ -4,8 +4,7 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from lohup.templater import Templater from lohup.templater import Templater
from lohup.util import Masked, ensure_exists from lohup.util import DeepValidator, Masked, ensure_exists
from lohup.util import DeepValidator
class ConfigError(ValueError): class ConfigError(ValueError):
@@ -209,7 +208,9 @@ class HookSet:
return HookSet(before_all=before_all, after_all=after_all) return HookSet(before_all=before_all, after_all=after_all)
@staticmethod @staticmethod
def hook_for(item: dict, kind: str, templater: ValidatedTemplater, validator: DeepValidator): def hook_for(
item: dict, kind: str, templater: ValidatedTemplater, validator: DeepValidator
):
match kind: match kind:
case "command": case "command":
with validator.inner("command") as inner: with validator.inner("command") as inner:
@@ -319,7 +320,7 @@ class TomlConfig:
if not paths: if not paths:
validator.error(f"profile {name!r}: no paths or command provided") validator.error(f"profile {name!r}: no paths or command provided")
continue continue
resolved: list[Path] = [] resolved: list[str] = []
for i, raw in enumerate(paths): for i, raw in enumerate(paths):
raw = templater.expand_validated(raw, f"paths: {i}", inner) raw = templater.expand_validated(raw, f"paths: {i}", inner)
if not raw: if not raw:
@@ -327,7 +328,7 @@ class TomlConfig:
pth = Path(raw) pth = Path(raw)
if not pth.is_absolute(): if not pth.is_absolute():
pth = settings.backup_base_dir.joinpath(pth) pth = settings.backup_base_dir.joinpath(pth)
resolved.append(pth) resolved.append(str(pth))
profile = PathsProfile( profile = PathsProfile(
name, name,
repo=opts.get("repo"), repo=opts.get("repo"),
+90
View File
@@ -0,0 +1,90 @@
from dataclasses import dataclass
from datetime import datetime
import typing as ty
@dataclass
class Snapshot:
id: str
profile: str
tags: list[str]
hostname: str
time: datetime
end: datetime
raw: dict[str, ty.Any]
parent: str | None
_engine: str
@staticmethod
def from_rustic(item: dict[str, ty.Any], group: dict[str, ty.Any]) -> Snapshot:
profile = ""
for tag in group["tags"]:
if tag.startswith("lohup.profile="):
profile = tag.removeprefix("lohup.profile=")
return Snapshot(
id=item["id"][:8],
profile=profile,
tags=group["tags"],
hostname=group["hostname"],
time=datetime.fromisoformat(item["time"]),
end=datetime.fromisoformat(item["summary"]["backup_end"]),
raw=item,
parent=item.get("parent"),
_engine="rustic"
)
@staticmethod
def from_restic(item: dict[str, ty.Any]) -> Snapshot:
profile = ""
for tag in item["tags"]:
if tag.startswith("lohup.profile="):
profile = tag.removeprefix("lohup.profile=")
return Snapshot(
id=item["short_id"],
profile=profile,
tags=item["tags"],
hostname=item["hostname"],
time=datetime.fromisoformat(item["time"]),
end=datetime.fromisoformat(item["summary"]["backup_end"]),
raw=item,
_engine="restic"
)
@property
def name(self) -> str:
return self.profile or "-".join(self.tags)
@property
def total_changes(self) -> int:
summary = self.raw["summary"]
changes = summary["files_new"] + summary["files_changed"]
return changes
@property
def size(self):
return self.raw["summary"]["data_added"]
@property
def size_compressed(self):
return self.raw["summary"]["data_added_packed"]
@property
def files_read(self):
summary = self.raw["summary"]
return summary["total_files_processed"] + summary["total_dirs_processed"]
@property
def bytes_read(self):
summary = self.raw["summary"]
return summary["total_bytes_processed"]
@property
def changes_ratio(self) -> float:
summary = self.raw["summary"]
total_files: int = summary["total_files_processed"]
ratio = self.total_changes / total_files if total_files else 0.0
return ratio
@property
def duration(self):
return self.end - self.time
+4 -3
View File
@@ -1,7 +1,8 @@
from dataclasses import dataclass, field
import click
import logging
import enum import enum
import logging
from dataclasses import dataclass, field
import click
class LogLevel(enum.IntEnum): class LogLevel(enum.IntEnum):
+13 -7
View File
@@ -1,11 +1,12 @@
from pathlib import Path import json
from dataclasses import dataclass, field
import os import os
import subprocess as procs import subprocess as procs
import json from collections import defaultdict
from dataclasses import dataclass, field
from pathlib import Path
from lohup import config from lohup import config, engine
from lohup.logger import CliLogger, BasicLogger from lohup.logger import BasicLogger, CliLogger
@dataclass @dataclass
@@ -69,7 +70,12 @@ class Restic:
cmd.append("--json") cmd.append("--json")
result = procs.check_output(cmd, env=env, encoding="utf-8") result = procs.check_output(cmd, env=env, encoding="utf-8")
if format == "json": if format == "json":
return json.loads(result) content = json.loads(result)
snapshots = defaultdict(list)
for x in content:
snap = engine.Snapshot.from_restic(x)
snapshots[snap.name].append(snap)
return snapshots
return result return result
def _prepare(self, args: list[str]): def _prepare(self, args: list[str]):
@@ -84,4 +90,4 @@ class Restic:
return self return self
def __exit__(self, type, value, traceback): def __exit__(self, type, value, traceback):
return self return
+29 -9
View File
@@ -1,10 +1,12 @@
from pathlib import Path
from dataclasses import dataclass, field
import subprocess as procs
import json import json
import subprocess as procs
from collections import defaultdict
from dataclasses import dataclass, field
from pathlib import Path
import tomlkit import tomlkit
from lohup import config from lohup import config, engine
from lohup.logger import LoggerProto from lohup.logger import LoggerProto
@@ -23,6 +25,7 @@ class Rustic:
out = {} out = {}
repo_pass = Path(self.repo.repo_key_file.value).read_text().strip() repo_pass = Path(self.repo.repo_key_file.value).read_text().strip()
out["repository"] = {"password": repo_pass} out["repository"] = {"password": repo_pass}
out["global"] = {"group-by": "host,tags", "log-level": "warn"}
match self.repo: match self.repo:
case config.S3Repository(): case config.S3Repository():
out["repository"]["repository"] = "opendal:s3" out["repository"]["repository"] = "opendal:s3"
@@ -44,7 +47,7 @@ class Rustic:
return self.conf_file return self.conf_file
def _cmdline(self): def _cmdline(self):
out = [self.binary, "--log-level=warn", "-P", str(self.conf_dir / "rustic")] out = [self.binary, "-P", str(self.conf_dir / "rustic")]
return out return out
def run(self, args): def run(self, args):
@@ -53,7 +56,13 @@ class Rustic:
procs.check_call(cmd) procs.check_call(cmd)
def backup(self, profile: config.Profile): def backup(self, profile: config.Profile):
args = ["backup", "--tag", profile.name] args = [
"backup",
"--tag",
profile.name,
"--tag",
f"lohup.profile={profile.name}",
]
args.extend(profile.cli_args) args.extend(profile.cli_args)
match profile: match profile:
case config.PathsProfile(): case config.PathsProfile():
@@ -75,14 +84,25 @@ class Rustic:
with procs.Popen(cmd, stdin=procs.PIPE) as rustic: with procs.Popen(cmd, stdin=procs.PIPE) as rustic:
procs.check_call(src_cmd, stdout=rustic.stdin) procs.check_call(src_cmd, stdout=rustic.stdin)
def snapshots(self, format="text"): def snapshots(
self, format="text", profile: str = None
) -> str | dict[str, list[engine.Snapshot]]:
cmd = self._cmdline() cmd = self._cmdline()
cmd.extend(["snapshots", "--compact"]) cmd.extend(["snapshots"])
if format == "json": if format == "json":
cmd.append("--json") cmd.append("--json")
if profile:
cmd.extend(["--filter-tags", f"lohup.profile={profile}"])
result = procs.check_output(cmd, encoding="utf-8") result = procs.check_output(cmd, encoding="utf-8")
if format == "json": if format == "json":
return json.loads(result) content = json.loads(result)
snapshots = defaultdict(list)
for group in content:
key = group["group_key"]
for x in group["snapshots"]:
snap = engine.Snapshot.from_rustic(x, group=key)
snapshots[snap.name].append(snap)
return snapshots
return result return result
def __enter__(self): def __enter__(self):
+6 -2
View File
@@ -16,10 +16,14 @@ class Templater:
def expand(self, text: str, extras: dict[str, str] | None = None) -> str: def expand(self, text: str, extras: dict[str, str] | None = None) -> str:
value = self.safe_expand(text, extras) value = self.safe_expand(text, extras)
if isinstance(value, list): if isinstance(value, list):
raise ExceptionGroup("Errors during templating", [KeyError(x) for x in value]) raise ExceptionGroup(
"Errors during templating", [KeyError(x) for x in value]
)
return value return value
def safe_expand(self, text: str, extras: dict[str, str] | None = None) -> str | list[str]: def safe_expand(
self, text: str, extras: dict[str, str] | None = None
) -> str | list[str]:
if not text or "$" not in text: if not text or "$" not in text:
return text return text
args = self.context.copy() args = self.context.copy()
+9 -5
View File
@@ -1,7 +1,7 @@
from pathlib import Path
from dataclasses import dataclass, field
from contextlib import contextmanager
import typing as ty import typing as ty
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
@dataclass(repr=False) @dataclass(repr=False)
@@ -15,7 +15,6 @@ class Masked:
T = ty.TypeVar("T") T = ty.TypeVar("T")
def ensure_exists(name, expect="file", field=None) -> str | None: def ensure_exists(name, expect="file", field=None) -> str | None:
if not name: if not name:
if field is not None: if field is not None:
@@ -44,8 +43,10 @@ class PrefixedException:
def __str__(self): def __str__(self):
return f"{self.prefix}: {self.value}" return f"{self.prefix}: {self.value}"
AnyFail = PrefixedException | Exception | str AnyFail = PrefixedException | Exception | str
@dataclass @dataclass
class DeepValidator: class DeepValidator:
prefix: str | None = field(default=None) prefix: str | None = field(default=None)
@@ -68,7 +69,9 @@ class DeepValidator:
def assert_type(self, value: T, expect: type) -> T: def assert_type(self, value: T, expect: type) -> T:
if not isinstance(value, expect): if not isinstance(value, expect):
self.error(f"expected {expect.__qualname__}, got {type(value).__qualname__}") self.error(
f"expected {expect.__qualname__}, got {type(value).__qualname__}"
)
return value return value
def lines(self): def lines(self):
@@ -92,6 +95,7 @@ class DeepValidator:
def _prefixed(self, s): def _prefixed(self, s):
return f"{self.prefix}: {s}" if self.prefix else s return f"{self.prefix}: {s}" if self.prefix else s
class ValidationFailed(Exception): class ValidationFailed(Exception):
exceptions: list[PrefixedException] exceptions: list[PrefixedException]
messages: list[str] messages: list[str]
+64 -52
View File
@@ -1,16 +1,59 @@
from pathlib import Path from pathlib import Path
import shutil
import pytest
from lohup.app import Lohup from lohup.app import Lohup
from lohup.config import ConfigError from lohup.config import ConfigError
from lohup.logger import BasicLogger, LogLevel from lohup.logger import BasicLogger, LogLevel
from lohup.util import DeepValidator, ValidationFailed from lohup.util import DeepValidator, ValidationFailed
import pytest
class RepoEnvironment:
def __init__(self, tmp_path: Path):
self.basepath = tmp_path
self.repo = tmp_path / "repo"
self.pwfile = tmp_path / "password.txt"
self.cfg = tmp_path / "lohup.toml"
self.log = BasicLogger(level=LogLevel.DEBUG)
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, vars: dict[str, str]):
kwargs = vars.copy()
kwargs["base"] = str(self.basepath)
kwargs["repo"] = str(self.repo)
kwargs["pwfile"] = str(self.pwfile)
self.cfg.write_text(value.format_map(kwargs))
@pytest.fixture
def repo(tmp_path: Path):
env = RepoEnvironment(tmp_path)
yield env
if env.repo.exists():
shutil.rmtree(env.repo)
toml1 = """ toml1 = """
[settings.globals] [settings.globals]
""" """
def test_error_message(repo):
repo.write_config(toml1, {})
app = Lohup(config_path=str(repo.cfg), logger=repo.log)
msg = "no repositories defined"
with pytest.raises(ConfigError, match="Failed to parse TOML config") as exc:
app.load()
repo.log.error(exc.value)
cause = exc.value.__cause__
assert isinstance(cause, ValidationFailed) and msg in cause.messages
toml2 = """ toml2 = """
[[hooks.before-all]] [[hooks.before-all]]
kind = "???" kind = "???"
@@ -22,6 +65,18 @@ path = "{base}/repo"
repo-key-file = "{pwfile}" repo-key-file = "{pwfile}"
""" """
def test_hook_error(repo):
repo.write_config(toml2, {})
app = Lohup(config_path=str(repo.cfg), logger=repo.log)
msg = "hook: before-all: unsupported kind: ???"
with pytest.raises(ConfigError, match="Failed to parse TOML config") as exc:
app.load()
repo.log.error(exc.value)
cause = exc.value.__cause__
assert isinstance(cause, ValidationFailed) and msg in cause.messages
toml_vars = """ toml_vars = """
[settings.globals] [settings.globals]
foo = "$bar" foo = "$bar"
@@ -34,67 +89,24 @@ repo-key-file = "$unknown"
""" """
def test_invalid_vars(repo):
class RepoEnvironment: repo.write_config(toml_vars, {})
def __init__(self, tmp_path: Path): app = Lohup(config_path=str(repo.cfg), logger=repo.log)
self.basepath = tmp_path
self.log = BasicLogger(level=LogLevel.DEBUG)
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)
path = tmp_path / "lohup.toml"
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)
cause = exc.value.__cause__
assert isinstance(cause, ValidationFailed) and msg in cause.messages
def test_hook_error(tmp_path):
env = RepoEnvironment(tmp_path)
pwfile = env.write_password()
tmp_path.joinpath("repo").mkdir()
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: before-all: unsupported kind: ???"
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) 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 = [ lines = [
"settings: variable 'foo': nested variables in globals are not allowed", "settings: variable 'foo': nested variables in globals are not allowed",
"settings: variable 'bag': invalid type: int, expected string", "settings: variable 'bag': invalid type: int, expected string",
"repo 'local': read repo-key-file: variable 'unknown' is not defined" "repo 'local': read repo-key-file: variable 'unknown' is not defined",
] ]
with pytest.raises(ConfigError, match="Failed to parse TOML config") as exc: with pytest.raises(ConfigError) as exc:
app.load() app.load()
env.log.error(exc.value) repo.log.error(exc.value)
cause = exc.value.__cause__ cause = exc.value.__cause__
assert isinstance(cause, ValidationFailed) assert isinstance(cause, ValidationFailed)
for exc in cause.exceptions: for exc in cause.exceptions:
env.log.error(exc.value) repo.log.error(exc.value)
assert set(lines) == set(cause.messages) assert set(lines) == set(cause.messages)
def test_validator(): def test_validator():
validator = DeepValidator("testing") validator = DeepValidator("testing")
with validator.inner("read config") as inner: with validator.inner("read config") as inner:
Generated
+11
View File
@@ -50,6 +50,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/2c/e1/e6716421ea10d38022b952c159d5161ca1193197fb744506875fbb87ea7b/iniconfig-2.1.0-py3-none-any.whl", hash = "sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760", size = 6050, upload-time = "2025-03-19T20:10:01.071Z" }, { url = "https://files.pythonhosted.org/packages/2c/e1/e6716421ea10d38022b952c159d5161ca1193197fb744506875fbb87ea7b/iniconfig-2.1.0-py3-none-any.whl", hash = "sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760", size = 6050, upload-time = "2025-03-19T20:10:01.071Z" },
] ]
[[package]]
name = "isort"
version = "8.0.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/ef/7c/ec4ab396d31b3b395e2e999c8f46dec78c5e29209fac49d1f4dace04041d/isort-8.0.1.tar.gz", hash = "sha256:171ac4ff559cdc060bcfff550bc8404a486fee0caab245679c2abe7cb253c78d", size = 769592, upload-time = "2026-02-28T10:08:20.685Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3e/95/c7c34aa53c16353c56d0b802fba48d5f5caa2cdee7958acbcb795c830416/isort-8.0.1-py3-none-any.whl", hash = "sha256:28b89bc70f751b559aeca209e6120393d43fbe2490de0559662be7a9787e3d75", size = 89733, upload-time = "2026-02-28T10:08:19.466Z" },
]
[[package]] [[package]]
name = "lohup" name = "lohup"
version = "0.1.0" version = "0.1.0"
@@ -62,6 +71,7 @@ dependencies = [
[package.dev-dependencies] [package.dev-dependencies]
dev = [ dev = [
{ name = "isort" },
{ name = "pyinstaller" }, { name = "pyinstaller" },
{ name = "pytest" }, { name = "pytest" },
] ]
@@ -75,6 +85,7 @@ requires-dist = [
[package.metadata.requires-dev] [package.metadata.requires-dev]
dev = [ dev = [
{ name = "isort", specifier = ">=8.0.1" },
{ name = "pyinstaller", specifier = ">=6.21.0" }, { name = "pyinstaller", specifier = ">=6.21.0" },
{ name = "pytest", specifier = ">=9.1.1" }, { name = "pytest", specifier = ">=9.1.1" },
] ]