hippogrief/pobsync
1
0
Fork 0
Code Issues 21 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

refactor: replace legacy CLI with Django command surface

Browse Source 
Retire the old YAML and cron oriented pobsync CLI commands and expose a
SQL-first Django-backed command surface instead. Add schedule and
retention management commands, move shared defaults/parsing out of legacy
commands, remove obsolete command modules, and update documentation and
tests for the new workflow.
This commit is contained in:
Peter van Arkel
2026-05-19 05:14:29 +02:00
parent 6d9ddc4457
commit e564262c72
22 changed files with 351 additions and 2043 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
README.md
View File

@@ -1,123 +1,25 @@
# pobsync
`pobsync` is a pull-based backup tool that runs on a central backup server and pulls data from remote servers via rsync over SSH.
`pobsync` is a pull-based backup service. It runs on a central backup server and pulls data from remote machines via rsync over SSH.
Key points:
The refactor direction is SQL-first:
- All backup data lives on the backup server.
- Snapshots are rsync-based and use hardlinking (--link-dest) for space efficiency.
- Designed for scheduled runs (cron) and manual runs.
- Minimal external dependencies (currently only PyYAML).
- Django is the management layer and source of truth.
- SQLite is the default database; MariaDB is optional.
- Backups still use the existing rsync snapshot engine internally.
- Scheduling is handled by a Django/Docker scheduler process, not host cron.
- Legacy YAML import/export exists only for migration and inspection.
## Requirements
On the backup server:
On the backup server or in the container:
- Python 3
- Python 3.11+
- rsync
- ssh
- SSH key-based access from the backup server to remotes
## Canonical installation (no venv, repo used only for deployment)
This project uses a simple and explicit deployment model:
- The git clone is only used as a deployment input (and later for updates).
- Runtime code is deployed into /opt/pobsync/lib.
- The canonical entrypoint is /opt/pobsync/bin/pobsync.
### Install
```git clone https://code.hosting.hippogrief.nl/hippogrief/pobsync.git
cd pobsync
sudo ./scripts/deploy --prefix /opt/pobsync
pobsync install --backup-root /mnt/backups/pobsync (install default configurations)
pobsync doctor (check if the installation was done correctly)
```
### Update
```
cd /path/to/pobsync
git pull
sudo ./scripts/deploy --prefix /opt/pobsync
sudo /opt/pobsync/bin/pobsync doctor
```
## Configuration
Global configuration is stored at:
- /opt/pobsync/config/global.yaml
Per-host configuration files are stored at:
- /opt/pobsync/config/hosts/<host>.yaml
## Some useful commands to get you started
Create a new host configuration:
`pobsync init-host <host>`
List configured remotes:
`pobsync list-remotes`
Inspect the effective configuration for a host:
`pobsync show-config <host>`
## Running backups
Run a scheduled backup for a host:
`pobsync run-scheduled <host>`
Optionally apply retention pruning after the run:
`pobsync run-scheduled <host> --prune`
## Scheduling (cron)
Create a cron schedule (writes into /etc/cron.d/pobsync by default):
`pobsync schedule create <host> --daily 02:15 --prune`
List existing schedules:
`pobsync schedule list`
Remove a schedule:
`pobsync schedule remove <host>`
Cron output is redirected to:
- /var/log/pobsync/<host>.cron.log
## Development (optional)
For development purposes you can still use an editable install, this is why pyproject.toml still exists. On systems with an externally managed Python installation, create a virtualenv first.
```
python3 -m venv .venv
. .venv/bin/activate
python3 -m pip install -e .
pobsync --help
```
For production use, always use the canonical entrypoint:
/opt/pobsync/bin/pobsync
## Django backend (early refactor layer)
The Django backend is becoming the management layer and source of truth for pobsync. Structured SQL fields store backup, SSH, rsync, retention, schedule, run, and snapshot state; legacy JSON/YAML remains only as an import/export compatibility path while the engine is being refactored.
### Local SQLite development
## Local Development
```
python3 -m venv .venv
@@ -133,40 +35,69 @@ The admin is available at:
- http://127.0.0.1:8000/admin/
Import existing YAML configs into the database:
## SQL-First Setup
Create global config:
```
pobsync configure-global --backup-root /mnt/backups/pobsync
```
Create a host config:
```
pobsync configure-host <host> --address <host-or-ip>
```
Run a backup:
```
pobsync backup <host> --prune
```
Create or update a schedule:
```
pobsync schedule <host> --cron "15 2 * * *" --prune
```
Run the scheduler:
```
pobsync scheduler --loop --interval 60
```
Plan or apply retention manually:
```
pobsync retention <host>
pobsync retention <host> --apply --yes --max-delete 10
```
The `pobsync` executable is a thin wrapper around Django management commands. Direct Django access is also available:
```
pobsync django check
python3 manage.py run_pobsync_backup <host> --prune
```
## Migration Helpers
Import existing legacy YAML configs:
```
python3 manage.py import_pobsync_configs --prefix /opt/pobsync
```
Create SQL-backed configuration directly:
```
python3 manage.py configure_pobsync_global --backup-root /mnt/backups/pobsync
python3 manage.py configure_pobsync_host <host> --address <host-or-ip>
```
Run a backup through Django while still using the existing pobsync engine:
```
python3 manage.py run_pobsync_backup <host> --prefix /opt/pobsync --prune
```
The Django backup command reads backup and retention config from SQL directly. Runtime YAML export is kept as a compatibility tool for older CLI flows during the transition.
Export database configs to runtime YAML for legacy CLI compatibility:
Export SQL config to legacy runtime YAML for inspection or one-off compatibility:
```
python3 manage.py export_pobsync_configs --prefix /opt/pobsync
```
Run due schedules from the database:
These commands are migration helpers, not the normal operating model.
```
python3 manage.py run_pobsync_scheduler --loop --interval 60
```
### Docker with SQLite
## Docker With SQLite
```
docker compose up --build web
@@ -176,15 +107,15 @@ This starts Django on:
- http://127.0.0.1:8000/admin/
The container persists `/opt/pobsync` and the SQLite database in Docker volumes.
Run the Django scheduler alongside the web admin:
Run the scheduler alongside the web admin:
```
docker compose up --build web scheduler
```
### Docker with MariaDB
The container persists `/opt/pobsync` and the SQLite database in Docker volumes.
## Docker With MariaDB
```
docker compose --profile mariadb up --build web-mariadb
@@ -196,15 +127,22 @@ With the scheduler:
docker compose --profile mariadb up --build web-mariadb scheduler-mariadb
```
The MariaDB profile is optional. SQLite remains the default because it is enough for a single backup server and keeps deployment simple.
SQLite remains the default because it is enough for a single backup server and keeps deployment simple.
### Refactor direction
## Current Architecture
Recommended next steps:
The public command surface is Django-first. The old YAML/cron CLI has been retired from the `pobsync` entrypoint.
- Remove remaining legacy YAML-first commands after SQL-first setup covers all workflows.
- Record more engine-side run details into `BackupRun` and `SnapshotRecord`.
- Treat SQL as the source of truth and export YAML only as a compatibility layer for the current engine.
- Run schedules from Django/Docker instead of writing host cron files.
- Add a snapshot discovery command that syncs existing snapshot metadata into `SnapshotRecord`.
- Add tests around retention, scheduling, and config merge before deeper internal reshaping.
The remaining internal engine code still contains reusable backup primitives:
- snapshot naming and metadata
- rsync command construction and execution
- retention planning and pruning
- host locking
Next refactor targets:
- Record discovered snapshots into `SnapshotRecord`.
- Move more snapshot lifecycle details into typed domain objects.
- Replace remaining dictionary-shaped config at engine boundaries.
- Remove legacy YAML import/export once production migration no longer needs it.

493
src/pobsync/cli.py
View File

@@ -1,462 +1,59 @@
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
import os
import sys
from typing import Sequence
from .commands.doctor import run_doctor
from .commands.init_host import run_init_host
from .commands.install import run_install
from .commands.list_remotes import run_list_remotes
from .commands.retention_apply import run_retention_apply
from .commands.retention_plan import run_retention_plan
from .commands.run_scheduled import run_scheduled
from .commands.schedule_create import run_schedule_create
from .commands.schedule_list import run_schedule_list
from .commands.schedule_remove import run_schedule_remove
from .commands.show_config import dump_yaml, run_show_config
from .commands.snapshots_list import run_snapshots_list
from .commands.snapshots_show import run_snapshots_show
from .errors import LockError, PobsyncError
from .schedule import CRON_FILE_DEFAULT
from .util import to_json_safe
from django.core.management import execute_from_command_line
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(prog="pobsync")
p.add_argument("--prefix", default="/opt/pobsync", help="Pobsync home directory (default: /opt/pobsync)")
p.add_argument("--json", action="store_true", help="Machine-readable JSON output")
sub = p.add_subparsers(dest="command", required=True)
# install
ip = sub.add_parser("install", help="Bootstrap /opt/pobsync layout and create global config")
ip.add_argument("--backup-root", help="Backup root directory (e.g. /srv/backups)")
ip.add_argument("--retention", default="daily=14,weekly=8,monthly=12,yearly=0", help="Default retention for init-host")
ip.add_argument("--force", action="store_true", help="Overwrite existing global config")
ip.add_argument("--dry-run", action="store_true", help="Show actions, do not write")
ip.set_defaults(_handler=cmd_install)
# init-host
hp = sub.add_parser("init-host", help="Create a host config YAML under config/hosts")
hp.add_argument("host", help="Host name (used as filename)")
hp.add_argument("--address", help="Hostname or IP of the remote")
hp.add_argument("--ssh-user", default=None)
hp.add_argument("--ssh-port", type=int, default=None)
hp.add_argument("--retention", default=None, help="Override retention for this host (daily=...,weekly=...)")
hp.add_argument("--exclude-add", action="append", default=[], help="Additional excludes (repeatable)")
hp.add_argument("--exclude-replace", action="append", default=None, help="Replace excludes list (repeatable)")
hp.add_argument("--include", action="append", default=[], help="Include patterns (repeatable)")
hp.add_argument("--force", action="store_true")
hp.add_argument("--dry-run", action="store_true")
hp.set_defaults(_handler=cmd_init_host)
# doctor
dp = sub.add_parser("doctor", help="Validate installation and configuration")
dp.add_argument("host", nargs="?", default=None, help="Optional host to validate")
dp.add_argument("--connect", action="store_true", help="Try SSH connectivity check (phase 2)")
dp.add_argument("--rsync-dry-run", action="store_true", help="Try rsync dry run (phase 2)")
dp.set_defaults(_handler=cmd_doctor)
# list remotes
lp = sub.add_parser("list-remotes", help="List configured remotes (host configs)")
lp.set_defaults(_handler=cmd_list_remotes)
# show config
sp = sub.add_parser("show-config", help="Show host configuration (raw or effective)")
sp.add_argument("host", help="Host to show")
sp.add_argument("--effective", action="store_true", help="Show merged effective config")
sp.set_defaults(_handler=cmd_show_config)
# run scheduled
rp = sub.add_parser("run-scheduled", help="Run a scheduled backup for a host")
rp.add_argument("host", help="Host to back up")
rp.add_argument("--dry-run", action="store_true", help="Run rsync --dry-run without creating directories")
rp.add_argument("--prune", action="store_true", help="Apply retention after a successful run (default: false)")
rp.add_argument("--prune-max-delete", type=int, default=10, help="Refuse to prune more than N snapshots (default: 10)")
rp.add_argument("--prune-protect-bases", action="store_true", help="When pruning, also keep base snapshots referenced in meta")
rp.set_defaults(_handler=cmd_run_scheduled)
# snapshots
sn = sub.add_parser("snapshots", help="Inspect snapshots (list/show)")
sn_sub = sn.add_subparsers(dest="snapshots_cmd", required=True)
sn_list = sn_sub.add_parser("list", help="List snapshots for a host")
sn_list.add_argument("host", help="Host name")
sn_list.add_argument("--kind", default="all", help="scheduled|manual|incomplete|all (default: all)")
sn_list.add_argument("--limit", type=int, default=20, help="Max results (default: 20)")
sn_list.add_argument("--include-incomplete", action="store_true", help="Include .incomplete when --kind=all")
sn_list.set_defaults(_handler=cmd_snapshots_list)
sn_show = sn_sub.add_parser("show", help="Show snapshot metadata")
sn_show.add_argument("host", help="Host name")
sn_show.add_argument("--kind", required=True, help="scheduled|manual|incomplete")
sn_show.add_argument("dirname", help="Snapshot directory name")
sn_show.add_argument("--tail", type=int, default=None, help="Show last N lines of rsync.log")
sn_show.set_defaults(_handler=cmd_snapshots_show)
# retention
rt = sub.add_parser("retention", help="Retention management")
rt_sub = rt.add_subparsers(dest="retention_cmd", required=True)
rt_plan = rt_sub.add_parser("plan", help="Show retention prune plan (dry-run)")
rt_plan.add_argument("host", help="Host name")
rt_plan.add_argument("--kind", default="scheduled", help="scheduled|manual|all (default: scheduled)")
rt_plan.add_argument("--protect-bases", action="store_true", help="Also keep base snapshots referenced in meta (default: false)")
rt_plan.set_defaults(_handler=cmd_retention_plan)
rt_apply = rt_sub.add_parser("apply", help="Apply retention plan (DESTRUCTIVE)")
rt_apply.add_argument("host", help="Host name")
rt_apply.add_argument("--kind", default="scheduled", help="scheduled|manual|all (default: scheduled)")
rt_apply.add_argument("--protect-bases", action="store_true", help="Also keep base snapshots referenced in meta (default: false)")
rt_apply.add_argument("--max-delete", type=int, default=10, help="Refuse to delete more than N snapshots (default: 10)")
rt_apply.add_argument("--yes", action="store_true", help="Confirm deletion")
rt_apply.set_defaults(_handler=cmd_retention_apply)
# schedule
sch = sub.add_parser("schedule", help="Manage cron schedules in /etc/cron.d/pobsync")
sch_sub = sch.add_subparsers(dest="schedule_cmd", required=True)
sch_create = sch_sub.add_parser("create", help="Create or update a schedule for a host")
sch_create.add_argument("host", help="Host name")
mode = sch_create.add_mutually_exclusive_group(required=True)
mode.add_argument("--cron", default=None, help='Raw cron expression (5 fields), e.g. "15 2 * * *"')
mode.add_argument("--daily", default=None, help="Daily at HH:MM")
mode.add_argument("--hourly", type=int, default=None, help="Hourly at minute N (0..59)")
mode.add_argument("--weekly", action="store_true", help="Weekly schedule (requires --dow and --time)")
mode.add_argument("--monthly", action="store_true", help="Monthly schedule (requires --day and --time)")
sch_create.add_argument("--dow", default=None, help="For --weekly: mon,tue,wed,thu,fri,sat,sun")
sch_create.add_argument("--day", type=int, default=None, help="For --monthly: day of month (1..31)")
sch_create.add_argument("--time", default=None, help="For --weekly/--monthly: HH:MM")
sch_create.add_argument("--user", default="root", help="Cron user field (default: root)")
sch_create.add_argument("--cron-file", default=CRON_FILE_DEFAULT, help="Cron file path (default: /etc/cron.d/pobsync)")
sch_create.add_argument("--prune", action="store_true", help="Run retention prune after successful backup")
sch_create.add_argument("--prune-max-delete", type=int, default=10, help="Prune guardrail (default: 10)")
sch_create.add_argument("--prune-protect-bases", action="store_true", help="Prune with base protection (default: false)")
sch_create.add_argument("--dry-run", action="store_true", help="Show actions, do not write")
sch_create.set_defaults(_handler=cmd_schedule_create)
sch_list = sch_sub.add_parser("list", help="List schedules from /etc/cron.d/pobsync")
sch_list.add_argument("--host", default=None, help="Filter by host")
sch_list.add_argument("--cron-file", default=CRON_FILE_DEFAULT, help="Cron file path (default: /etc/cron.d/pobsync)")
sch_list.set_defaults(_handler=cmd_schedule_list)
sch_remove = sch_sub.add_parser("remove", help="Remove schedule block for a host")
sch_remove.add_argument("host", help="Host name")
sch_remove.add_argument("--cron-file", default=CRON_FILE_DEFAULT, help="Cron file path (default: /etc/cron.d/pobsync)")
sch_remove.add_argument("--dry-run", action="store_true", help="Show actions, do not write")
sch_remove.set_defaults(_handler=cmd_schedule_remove)
return p
COMMAND_ALIASES = {
"configure-global": "configure_pobsync_global",
"configure-host": "configure_pobsync_host",
"schedule": "configure_pobsync_schedule",
"backup": "run_pobsync_backup",
"retention": "run_pobsync_retention",
"scheduler": "run_pobsync_scheduler",
}
def parse_retention(s: str) -> dict[str, int]:
out: dict[str, int] = {}
parts = [p.strip() for p in s.split(",") if p.strip()]
for part in parts:
if "=" not in part:
raise ValueError(f"Invalid retention component: {part!r}")
k, v = part.split("=", 1)
k = k.strip()
v = v.strip()
if k not in {"daily", "weekly", "monthly", "yearly"}:
raise ValueError(f"Invalid retention key: {k!r}")
n = int(v)
if n < 0:
raise ValueError(f"Retention must be >= 0 for {k}")
out[k] = n
for k in ("daily", "weekly", "monthly", "yearly"):
out.setdefault(k, 0)
return out
def _usage() -> str:
commands = "\n".join(f" {name}" for name in sorted(COMMAND_ALIASES))
return f"""pobsync is now backed by Django management commands.
Usage:
pobsync <command> [options]
pobsync django <management-command> [options]
Commands:
{commands}
"""
def _print(result: dict[str, Any], as_json: bool) -> None:
if as_json:
print(json.dumps(to_json_safe(result), indent=2, sort_keys=False))
return
def main(argv: Sequence[str] | None = None) -> int:
args = list(sys.argv[1:] if argv is None else argv)
if not args or args[0] in {"-h", "--help", "help"}:
print(_usage())
return 0
if result.get("ok") is True:
print("OK")
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "pobsync_server.settings")
command = args[0]
if command == "django":
django_args = ["pobsync", *args[1:]]
else:
print("FAILED")
if "actions" in result:
for a in result["actions"]:
print(f"- {a}")
if "results" in result:
for r in result["results"]:
label = "OK" if r.get("ok") else "FAIL"
name = r.get("check", "check")
msg = r.get("message") or r.get("error") or ""
extra = ""
if "path" in r:
extra = f" ({r['path']})"
elif "host" in r:
extra = f" ({r['host']})"
line = f"- {label} {name}{extra}"
if msg:
line += f" {msg}"
print(line)
if "hosts" in result:
for h in result["hosts"]:
print(h)
if "snapshot" in result:
print(f"- snapshot {result['snapshot']}")
if "base" in result and result["base"]:
print(f"- base {result['base']}")
if "snapshots" in result:
for s in result["snapshots"]:
kind = s.get("kind", "?")
dirname = s.get("dirname", "?")
status = s.get("status") or "unknown"
started_at = s.get("started_at") or ""
dur = s.get("duration_seconds")
dur_s = f"{dur}s" if isinstance(dur, int) else ""
extra = " ".join(x for x in [started_at, dur_s] if x)
if extra:
extra = " " + extra
print(f"- {kind} {dirname} {status}{extra}")
if "keep" in result and "delete" in result:
keep = result.get("keep") or []
delete = result.get("delete") or []
reasons = result.get("reasons") or {}
total = len(keep) + len(delete)
print(f"- total {total}")
print(f"- keep {len(keep)}")
print(f"- delete {len(delete)}")
if result.get("protect_bases") is True:
print("- protect_bases true")
if keep:
print("- keep:")
for d in keep:
rs = reasons.get(d) or []
rs_s = f" ({', '.join(rs)})" if rs else ""
print(f" - {d}{rs_s}")
if delete:
print("- delete:")
for item in delete:
dirname = item.get("dirname", "?")
dt = item.get("dt") or ""
status = item.get("status") or "unknown"
kind = item.get("kind", "?")
extra = " ".join(x for x in [kind, status, dt] if x)
if extra:
extra = " " + extra
print(f" - {dirname}{extra}")
if "schedules" in result:
for s in result["schedules"]:
host = s.get("host", "?")
cron = s.get("cron") or "unknown"
user = s.get("user") or "unknown"
prune = bool(s.get("prune", False))
prune_max = s.get("prune_max_delete", None)
protect = bool(s.get("prune_protect_bases", False))
extra = ""
if prune:
extra = " prune"
if isinstance(prune_max, int):
extra += f" max_delete={prune_max}"
if protect:
extra += " protect_bases"
print(f"- {host} {cron} {user}{extra}")
def cmd_install(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
retention = parse_retention(args.retention)
result = run_install(
prefix=prefix,
backup_root=args.backup_root,
retention=retention,
dry_run=bool(args.dry_run),
force=bool(args.force),
)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_init_host(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_init_host(
prefix=prefix,
host=args.host,
address=args.address,
retention=args.retention,
ssh_user=args.ssh_user,
ssh_port=args.ssh_port,
excludes_add=list(args.exclude_add),
excludes_replace=args.exclude_replace,
includes=list(args.include),
dry_run=bool(args.dry_run),
force=bool(args.force),
)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_doctor(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_doctor(prefix=prefix, host=args.host, connect=bool(args.connect), rsync_dry_run=bool(args.rsync_dry_run))
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_list_remotes(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_list_remotes(prefix=prefix)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_show_config(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_show_config(prefix=prefix, host=args.host, effective=bool(args.effective))
if args.json:
_print(result, as_json=True)
else:
print(dump_yaml(result["config"]).rstrip())
return 0 if result.get("ok") else 1
def cmd_run_scheduled(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_scheduled(
prefix=prefix,
host=args.host,
dry_run=bool(args.dry_run),
prune=bool(args.prune),
prune_max_delete=int(args.prune_max_delete),
prune_protect_bases=bool(args.prune_protect_bases),
)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 2
def cmd_snapshots_list(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_snapshots_list(
prefix=prefix,
host=args.host,
kind=args.kind,
limit=int(args.limit),
include_incomplete=bool(args.include_incomplete),
)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_snapshots_show(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_snapshots_show(prefix=prefix, host=args.host, kind=args.kind, dirname=args.dirname, tail=args.tail)
if args.json:
_print(result, as_json=True)
else:
print(dump_yaml(result.get("meta", {})).rstrip())
if result.get("log_path"):
print(f"\n# rsync.log: {result['log_path']}")
if result.get("log_tail"):
print("\n# rsync.log (tail)")
for line in result["log_tail"]:
print(line)
return 0 if result.get("ok") else 1
def cmd_retention_plan(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_retention_plan(prefix=prefix, host=args.host, kind=args.kind, protect_bases=bool(args.protect_bases))
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_retention_apply(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_retention_apply(
prefix=prefix,
host=args.host,
kind=args.kind,
protect_bases=bool(args.protect_bases),
yes=bool(args.yes),
max_delete=int(args.max_delete),
)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_schedule_create(args: argparse.Namespace) -> int:
prefix = Path(args.prefix)
result = run_schedule_create(
host=args.host,
prefix=prefix,
cron_file=Path(args.cron_file),
cron_expr=args.cron,
daily=args.daily,
hourly=args.hourly,
weekly=bool(args.weekly),
dow=args.dow,
time=args.time,
monthly=bool(args.monthly),
day=args.day,
user=args.user,
prune=bool(args.prune),
prune_max_delete=int(args.prune_max_delete),
prune_protect_bases=bool(args.prune_protect_bases),
dry_run=bool(args.dry_run),
)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_schedule_list(args: argparse.Namespace) -> int:
result = run_schedule_list(cron_file=Path(args.cron_file), host=args.host)
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def cmd_schedule_remove(args: argparse.Namespace) -> int:
result = run_schedule_remove(host=args.host, cron_file=Path(args.cron_file), dry_run=bool(args.dry_run))
_print(result, as_json=bool(args.json))
return 0 if result.get("ok") else 1
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
mapped = COMMAND_ALIASES.get(command)
if mapped is None:
print(f"Unknown pobsync command: {command}", file=sys.stderr)
print(_usage(), file=sys.stderr)
return 2
django_args = ["pobsync", mapped, *args[1:]]
try:
handler = getattr(args, "_handler")
return int(handler(args))
except PobsyncError as e:
if args.json:
_print({"ok": False, "error": str(e), "type": type(e).__name__}, as_json=True)
else:
print(f"ERROR: {e}")
if isinstance(e, LockError):
return 10
execute_from_command_line(django_args)
except SystemExit as exc:
code = exc.code
if isinstance(code, int):
return code
return 1
except KeyboardInterrupt:
if args.json:
_print({"ok": False, "error": "interrupted"}, as_json=True)
else:
print("ERROR: interrupted")
return 130
return 0

216
src/pobsync/commands/doctor.py
View File

@@ -1,216 +0,0 @@
from __future__ import annotations
import os
import shutil
import subprocess
from pathlib import Path
from typing import Any
from ..config.load import load_global_config, load_host_config
from ..paths import PobsyncPaths
from ..util import is_absolute_non_root
CRON_FILE_DEFAULT = Path("/etc/cron.d/pobsync")
LOG_DIR_DEFAULT = Path("/var/log/pobsync")
def _check_binary(name: str) -> tuple[bool, str]:
p = shutil.which(name)
if not p:
return False, f"missing binary: {name}"
return True, f"ok: {name} -> {p}"
def _check_writable_dir(path: Path) -> tuple[bool, str]:
try:
path.mkdir(parents=True, exist_ok=True)
except OSError as e:
return False, f"cannot create dir {path}: {e}"
try:
test = path / ".pobsync_write_test"
test.write_text("test", encoding="utf-8")
test.unlink(missing_ok=True)
except OSError as e:
return False, f"not writable: {path}: {e}"
return True, f"ok: writable {path}"
def _run(cmd: list[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(
cmd,
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
def _check_cron_service() -> tuple[bool, str]:
"""
Best-effort check: verify cron service is active on systemd hosts.
If systemctl is missing, we don't fail doctor phase 1.
"""
systemctl = shutil.which("systemctl")
if not systemctl:
return True, "ok: systemctl not found; cannot verify cron service status"
for svc in ("cron", "crond"):
cp = _run([systemctl, "is-active", svc])
if cp.returncode == 0 and cp.stdout.strip() == "active":
return True, f"ok: cron service active ({svc})"
return False, "cron service not active (tried: cron, crond)"
def _check_cron_file_permissions(path: Path) -> tuple[bool, str]:
"""
/etc/cron.d files must not be writable by group/other.
Owner should be root.
Mode can be 0600 or 0644 (both ok as long as not group/other-writable).
"""
if not path.exists():
return True, f"ok: cron file not present ({path}); schedule may not be configured yet"
try:
st = path.stat()
except OSError as e:
return False, f"cannot stat cron file {path}: {e}"
if not path.is_file():
return False, f"cron file is not a regular file: {path}"
problems: list[str] = []
# root owner
if st.st_uid != 0:
problems.append("owner is not root")
# must not be group/other writable
if (st.st_mode & 0o022) != 0:
problems.append("writable by group/other")
if problems:
mode_octal = oct(st.st_mode & 0o777)
return False, f"unsafe cron file permissions/ownership for {path} (mode={mode_octal}): {', '.join(problems)}"
mode_octal = oct(st.st_mode & 0o777)
return True, f"ok: cron file permissions/ownership OK ({path}, mode={mode_octal})"
def _check_pobsync_executable(prefix: Path) -> tuple[bool, str]:
exe = prefix / "bin" / "pobsync"
if not exe.exists():
return False, f"missing executable: {exe}"
if not os.access(str(exe), os.X_OK):
return False, f"not executable: {exe}"
return True, f"ok: executable {exe}"
def run_doctor(prefix: Path, host: str | None, connect: bool, rsync_dry_run: bool) -> dict[str, Any]:
# Phase 1 doctor does not perform network checks yet (connect/rsync_dry_run acknowledged).
paths = PobsyncPaths(home=prefix)
results: list[dict[str, Any]] = []
ok = True
# Check required layout
for d in (paths.config_dir, paths.hosts_dir, paths.state_dir, paths.locks_dir, paths.logs_dir):
exists = d.exists()
results.append({"check": "path_exists", "path": str(d), "ok": exists})
if not exists:
ok = False
# Load and validate global config
global_cfg: dict[str, Any] | None = None
if paths.global_config_path.exists():
try:
global_cfg = load_global_config(paths.global_config_path)
results.append({"check": "global_config", "path": str(paths.global_config_path), "ok": True})
except Exception as e:
ok = False
results.append({"check": "global_config", "path": str(paths.global_config_path), "ok": False, "error": str(e)})
else:
ok = False
results.append({"check": "global_config", "path": str(paths.global_config_path), "ok": False, "error": "missing"})
# Basic binaries
b1, m1 = _check_binary("rsync")
results.append({"check": "binary", "name": "rsync", "ok": b1, "message": m1})
ok = ok and b1
b2, m2 = _check_binary("ssh")
results.append({"check": "binary", "name": "ssh", "ok": b2, "message": m2})
ok = ok and b2
# backup_root checks
if global_cfg is not None:
backup_root = global_cfg.get("backup_root")
if isinstance(backup_root, str) and is_absolute_non_root(backup_root):
br = Path(backup_root)
w_ok, w_msg = _check_writable_dir(br)
results.append({"check": "backup_root", "path": str(br), "ok": w_ok, "message": w_msg})
ok = ok and w_ok
else:
ok = False
results.append({"check": "backup_root", "ok": False, "error": "invalid backup_root"})
else:
results.append({"check": "backup_root", "ok": False, "error": "global config not loaded"})
# ---- Scheduling checks (Step 1) ----
c_ok, c_msg = _check_cron_service()
results.append({"check": "schedule_cron_service", "ok": c_ok, "message": c_msg})
ok = ok and c_ok
f_ok, f_msg = _check_cron_file_permissions(CRON_FILE_DEFAULT)
results.append({"check": "schedule_cron_file", "path": str(CRON_FILE_DEFAULT), "ok": f_ok, "message": f_msg})
ok = ok and f_ok
# We treat missing log dir as a warning rather than hard-fail in phase 1:
# cron redirection may fail, but backups can still run.
if LOG_DIR_DEFAULT.exists():
l_ok, l_msg = _check_writable_dir(LOG_DIR_DEFAULT)
results.append({"check": "schedule_log_dir", "path": str(LOG_DIR_DEFAULT), "ok": l_ok, "message": l_msg})
ok = ok and l_ok
else:
results.append(
{
"check": "schedule_log_dir",
"path": str(LOG_DIR_DEFAULT),
"ok": True,
"message": f"ok: log dir does not exist ({LOG_DIR_DEFAULT}); cron redirection may fail (backlog: create in install)",
}
)
e_ok, e_msg = _check_pobsync_executable(prefix)
results.append({"check": "schedule_pobsync_executable", "path": str(prefix / "bin" / "pobsync"), "ok": e_ok, "message": e_msg})
ok = ok and e_ok
# host checks
if host is not None:
host_path = paths.hosts_dir / f"{host}.yaml"
if not host_path.exists():
ok = False
results.append({"check": "host_config", "host": host, "ok": False, "error": f"missing {host_path}"})
else:
try:
_ = load_host_config(host_path)
results.append({"check": "host_config", "host": host, "ok": True, "path": str(host_path)})
except Exception as e:
ok = False
results.append({"check": "host_config", "host": host, "ok": False, "path": str(host_path), "error": str(e)})
# Phase 1: report that connect/rsync_dry_run are not implemented yet
if connect:
results.append({"check": "connect", "ok": False, "error": "not implemented in phase 1"})
ok = False
if rsync_dry_run:
results.append({"check": "rsync_dry_run", "ok": False, "error": "not implemented in phase 1"})
ok = False
if not ok:
# Do not raise; return structured report. CLI will map to exit code 1.
return {"ok": False, "results": results}
return {"ok": True, "results": results}

82
src/pobsync/commands/init_host.py
View File

@@ -1,82 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml
from ..errors import ConfigError
from ..paths import PobsyncPaths
from ..util import sanitize_host
def build_host_config(
host: str,
address: str,
retention: dict[str, int],
ssh_user: str | None = None,
ssh_port: int | None = None,
excludes_add: list[str] | None = None,
excludes_replace: list[str] | None = None,
includes: list[str] | None = None,
) -> dict[str, Any]:
cfg: dict[str, Any] = {
"host": host,
"address": address,
"retention": retention,
"includes": includes or [],
}
if ssh_user is not None or ssh_port is not None:
cfg["ssh"] = {}
if ssh_user is not None:
cfg["ssh"]["user"] = ssh_user
if ssh_port is not None:
cfg["ssh"]["port"] = ssh_port
if excludes_replace is not None:
cfg["excludes_replace"] = excludes_replace
else:
cfg["excludes_add"] = excludes_add or []
return cfg
def run_init_host(
prefix: Path,
host: str,
address: str,
retention: dict[str, int],
ssh_user: str | None,
ssh_port: int | None,
excludes_add: list[str],
excludes_replace: list[str] | None,
includes: list[str],
dry_run: bool,
force: bool,
) -> dict[str, Any]:
host = sanitize_host(host)
paths = PobsyncPaths(home=prefix)
target = paths.hosts_dir / f"{host}.yaml"
if target.exists() and not force:
raise ConfigError(f"Host config already exists: {target} (use --force to overwrite)")
cfg = build_host_config(
host=host,
address=address,
retention=retention,
ssh_user=ssh_user,
ssh_port=ssh_port,
excludes_add=excludes_add,
excludes_replace=excludes_replace,
includes=includes,
)
action: str
if dry_run:
action = f"would write {target}"
else:
target.write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8")
action = f"wrote {target}"
return {"ok": True, "action": action, "host_config": str(target)}

152
src/pobsync/commands/install.py
View File

@@ -1,152 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml
from ..errors import InstallError
from ..paths import PobsyncPaths
from ..util import ensure_dir, is_absolute_non_root
DEFAULT_EXCLUDES = [
"/proc/***",
"/sys/***",
"/dev/***",
"/run/***",
"/tmp/***",
"/mnt/***",
"/media/***",
"/lost+found/***",
"/var/cache/***",
"/var/tmp/***",
"/var/run/***",
"/var/lock/***",
"/swapfile",
"/.snapshots/***",
]
DEFAULT_RSYNC_ARGS = [
"--archive",
"--numeric-ids",
"--delete",
"--delete-excluded",
"--partial",
"--partial-dir=.rsync-partial",
"--one-file-system",
"--relative",
"--human-readable",
"--stats",
]
def build_default_global_config(pobsync_home: Path, backup_root: str, retention: dict[str, int]) -> dict[str, Any]:
return {
"backup_root": backup_root,
"pobsync_home": str(pobsync_home),
"ssh": {
"user": "root",
"port": 22,
"options": [
"-oBatchMode=yes",
"-oStrictHostKeyChecking=accept-new",
],
},
"rsync": {
"binary": "rsync",
"args": DEFAULT_RSYNC_ARGS,
"timeout_seconds": 0,
"bwlimit_kbps": 0,
"extra_args": [],
},
"defaults": {
"source_root": "/",
"destination_subdir": "",
},
"excludes_default": DEFAULT_EXCLUDES,
"logging": {
"file": str(pobsync_home / "logs" / "pobsync.log"),
"level": "INFO",
},
"output": {
"default_format": "human",
},
# We store default retention here for init-host convenience; host config still requires retention.
"retention_defaults": retention,
}
def install_layout(paths: PobsyncPaths, dry_run: bool) -> list[str]:
actions: list[str] = []
for d in (paths.home, paths.config_dir, paths.hosts_dir, paths.state_dir, paths.locks_dir, paths.logs_dir):
actions.append(f"mkdir -p {d}")
if not dry_run:
ensure_dir(d)
return actions
def write_yaml(path: Path, data: dict[str, Any], dry_run: bool, force: bool) -> str:
if path.exists() and not force:
return f"skip existing {path}"
if path.exists() and force:
bak = path.with_suffix(path.suffix + ".bak")
if not dry_run:
bak.write_text(path.read_text(encoding="utf-8"), encoding="utf-8")
return f"overwrite {path} (backup {bak})"
if not dry_run:
path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
return f"write {path}"
def _ensure_system_log_dir(dry_run: bool) -> list[str]:
"""
Best-effort: create /var/log/pobsync to match cron redirection.
Not fatal if it fails (e.g., insufficient permissions in a non-root install attempt).
Note: the canonical entrypoint (/opt/pobsync/bin/pobsync) is owned by scripts/deploy.
install only prepares the runtime layout and config.
"""
actions: list[str] = []
log_dir = Path("/var/log/pobsync")
actions.append(f"mkdir -p {log_dir}")
if not dry_run:
try:
ensure_dir(log_dir)
except OSError as e:
actions.append(f"warn: cannot create {log_dir}: {e}")
return actions
def run_install(
prefix: Path,
backup_root: str | None,
retention: dict[str, int],
dry_run: bool,
force: bool,
) -> dict[str, Any]:
if backup_root is None:
raise InstallError("backup_root is required (use --backup-root or interactive mode)")
if not is_absolute_non_root(backup_root):
raise InstallError("backup_root must be an absolute path and must not be '/'")
paths = PobsyncPaths(home=prefix)
actions = install_layout(paths, dry_run=dry_run)
global_cfg = build_default_global_config(paths.home, backup_root=backup_root, retention=retention)
actions.append(write_yaml(paths.global_config_path, global_cfg, dry_run=dry_run, force=force))
# Install polish: ensure cron log directory exists.
# Code + entrypoint deployment is handled by scripts/deploy.
actions.extend(_ensure_system_log_dir(dry_run=dry_run))
return {
"ok": True,
"actions": actions,
"paths": {
"home": str(paths.home),
"global_config": str(paths.global_config_path),
},
}

24
src/pobsync/commands/list_remotes.py
View File

@@ -1,24 +0,0 @@
from __future__ import annotations
from pathlib import Path
from ..paths import PobsyncPaths
from ..util import sanitize_host
def run_list_remotes(prefix: Path) -> dict:
paths = PobsyncPaths(home=prefix)
hosts: list[str] = []
if paths.hosts_dir.exists():
for p in sorted(paths.hosts_dir.glob("*.yaml")):
host = p.stem
try:
sanitize_host(host)
except Exception:
# Ignore invalid filenames; doctor will catch config issues.
continue
hosts.append(host)
return {"ok": True, "hosts": hosts}

162
src/pobsync/commands/schedule_create.py
View File

@@ -1,162 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Any, Optional
from ..errors import ConfigError
from ..schedule import (
build_cron_expr_daily,
build_cron_expr_hourly,
build_cron_expr_monthly,
build_cron_expr_weekly,
normalize_cron_expr,
render_host_block,
upsert_host_block,
validate_cron_expr,
)
from ..util import ensure_dir, sanitize_host, write_text_atomic
def _choose_cron_expr(
*,
cron_expr: Optional[str],
daily: Optional[str],
hourly: Optional[int],
weekly: bool,
dow: Optional[str],
time: Optional[str],
monthly: bool,
day: Optional[int],
) -> str:
modes = [
("cron", cron_expr is not None),
("daily", daily is not None),
("hourly", hourly is not None),
("weekly", bool(weekly)),
("monthly", bool(monthly)),
]
chosen = [name for name, enabled in modes if enabled]
if len(chosen) == 0:
raise ConfigError("One of --cron/--daily/--hourly/--weekly/--monthly must be provided")
if len(chosen) > 1:
raise ConfigError("Choose exactly one of --cron/--daily/--hourly/--weekly/--monthly")
if cron_expr is not None:
validate_cron_expr(cron_expr)
return normalize_cron_expr(cron_expr)
if daily is not None:
return build_cron_expr_daily(daily)
if hourly is not None:
return build_cron_expr_hourly(hourly)
if weekly:
if dow is None or time is None:
raise ConfigError("--weekly requires --dow and --time")
return build_cron_expr_weekly(dow, time)
# monthly
if day is None or time is None:
raise ConfigError("--monthly requires --day and --time")
return build_cron_expr_monthly(day, time)
def run_schedule_create(
*,
host: str,
prefix: Path,
cron_file: Path,
cron_expr: Optional[str],
daily: Optional[str],
hourly: Optional[int],
weekly: bool,
dow: Optional[str],
time: Optional[str],
monthly: bool,
day: Optional[int],
user: str,
prune: bool,
prune_max_delete: int,
prune_protect_bases: bool,
dry_run: bool,
) -> dict[str, Any]:
host = sanitize_host(host)
if prune_max_delete < 0:
raise ConfigError("--prune-max-delete must be >= 0")
expr = _choose_cron_expr(
cron_expr=cron_expr,
daily=daily,
hourly=hourly,
weekly=weekly,
dow=dow,
time=time,
monthly=monthly,
day=day,
)
cmd = f"{prefix}/bin/pobsync --prefix {prefix} run-scheduled {host}"
if prune:
cmd += " --prune"
cmd += f" --prune-max-delete {int(prune_max_delete)}"
if prune_protect_bases:
cmd += " --prune-protect-bases"
log_dir = Path("/var/log/pobsync")
log_path = str(log_dir / f"{host}.cron.log")
block = render_host_block(
host=host,
cron_expr=expr,
user=user,
command=cmd,
log_path=log_path,
include_env=True,
)
try:
existing = cron_file.read_text(encoding="utf-8")
except FileNotFoundError:
existing = ""
except PermissionError as e:
raise ConfigError(f"Permission denied reading {cron_file}: {e}") from e
except OSError as e:
raise ConfigError(f"Failed reading {cron_file}: {e}") from e
had_block = f"# BEGIN POBSYNC host={host}" in existing
new_content = upsert_host_block(existing, host, block)
action_word = "updated" if had_block else "created"
actions = [
f"schedule {action_word} host={host}",
f"file {cron_file}",
f"cron {expr}",
f"user {user}",
]
if prune:
actions.append(f"prune enabled (max_delete={int(prune_max_delete)})")
if prune_protect_bases:
actions.append("prune protect_bases enabled")
if dry_run:
actions.append("dry-run (no file written)")
return {"ok": True, "actions": actions, "host": host, "cron_file": str(cron_file)}
# Best-effort ensure log dir exists
try:
ensure_dir(log_dir)
except Exception:
actions.append(f"warn: could not create {log_dir}")
try:
write_text_atomic(cron_file, new_content)
except PermissionError as e:
raise ConfigError(f"Permission denied writing {cron_file}: {e}") from e
except OSError as e:
raise ConfigError(f"Failed writing {cron_file}: {e}") from e
return {"ok": True, "actions": actions, "host": host, "cron_file": str(cron_file)}

87
src/pobsync/commands/schedule_list.py
View File

@@ -1,87 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List, Optional
from ..errors import ConfigError
from ..schedule import parse_cron_file
from ..util import sanitize_host
def _parse_prune_flags(command: Optional[str]) -> Dict[str, Any]:
"""
Best-effort parse of flags from the command string that we generate.
"""
if not command:
return {"prune": False, "prune_max_delete": None, "prune_protect_bases": False}
tokens = command.split()
prune = "--prune" in tokens
protect = "--prune-protect-bases" in tokens
max_delete = None
if "--prune-max-delete" in tokens:
try:
idx = tokens.index("--prune-max-delete")
if idx + 1 < len(tokens):
max_delete = int(tokens[idx + 1])
except (ValueError, IndexError):
max_delete = None
return {
"prune": bool(prune),
"prune_max_delete": max_delete,
"prune_protect_bases": bool(protect),
}
def run_schedule_list(*, cron_file: Path, host: Optional[str]) -> dict[str, Any]:
if host is not None:
host = sanitize_host(host)
try:
content = cron_file.read_text(encoding="utf-8")
except FileNotFoundError:
content = ""
except PermissionError as e:
raise ConfigError(f"Permission denied reading {cron_file}: {e}") from e
except OSError as e:
raise ConfigError(f"Failed reading {cron_file}: {e}") from e
blocks = parse_cron_file(content)
schedules: List[Dict[str, Any]] = []
if host is not None:
b = blocks.get(host)
if b is None:
return {"ok": True, "cron_file": str(cron_file), "schedules": []}
flags = _parse_prune_flags(b.command)
schedules.append(
{
"host": b.host,
"cron": b.cron_expr,
"user": b.user,
"command": b.command,
"log_path": b.log_path,
**flags,
}
)
return {"ok": True, "cron_file": str(cron_file), "schedules": schedules}
for h in sorted(blocks.keys()):
b = blocks[h]
flags = _parse_prune_flags(b.command)
schedules.append(
{
"host": b.host,
"cron": b.cron_expr,
"user": b.user,
"command": b.command,
"log_path": b.log_path,
**flags,
}
)
return {"ok": True, "cron_file": str(cron_file), "schedules": schedules}

39
src/pobsync/commands/schedule_remove.py
View File

@@ -1,39 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from ..errors import ConfigError
from ..schedule import remove_host_block
from ..util import sanitize_host, write_text_atomic
def run_schedule_remove(*, host: str, cron_file: Path, dry_run: bool) -> dict[str, Any]:
host = sanitize_host(host)
try:
existing = cron_file.read_text(encoding="utf-8")
except FileNotFoundError:
existing = ""
except PermissionError as e:
raise ConfigError(f"Permission denied reading {cron_file}: {e}") from e
except OSError as e:
raise ConfigError(f"Failed reading {cron_file}: {e}") from e
new_content = remove_host_block(existing, host)
actions = [f"schedule remove host={host}", f"file {cron_file}"]
if dry_run:
actions.append("dry-run (no file written)")
return {"ok": True, "actions": actions, "host": host, "cron_file": str(cron_file)}
try:
write_text_atomic(cron_file, new_content)
except PermissionError as e:
raise ConfigError(f"Permission denied writing {cron_file}: {e}") from e
except OSError as e:
raise ConfigError(f"Failed writing {cron_file}: {e}") from e
return {"ok": True, "actions": actions, "host": host, "cron_file": str(cron_file)}

33
src/pobsync/commands/show_config.py
View File

@@ -1,33 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml
from ..config.load import load_global_config, load_host_config
from ..config.merge import build_effective_config
from ..paths import PobsyncPaths
from ..util import sanitize_host
def run_show_config(prefix: Path, host: str, effective: bool) -> dict[str, Any]:
host = sanitize_host(host)
paths = PobsyncPaths(home=prefix)
global_cfg = load_global_config(paths.global_config_path)
host_cfg = load_host_config(paths.hosts_dir / f"{host}.yaml")
cfg = build_effective_config(global_cfg, host_cfg) if effective else host_cfg
return {
"ok": True,
"host": host,
"effective": effective,
"config": cfg,
}
def dump_yaml(data: Any) -> str:
return yaml.safe_dump(data, sort_keys=False)

108
src/pobsync/commands/snapshots_list.py
View File

@@ -1,108 +0,0 @@
from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional, Tuple
from ..config.load import load_global_config, load_host_config
from ..config.merge import build_effective_config
from ..errors import ConfigError
from ..paths import PobsyncPaths
from ..snapshot_meta import iter_snapshot_dirs, normalize_kind, read_snapshot_meta, resolve_host_root
from ..util import sanitize_host
def _parse_iso_z(s: Any) -> Optional[datetime]:
"""
Parse timestamps like '2026-02-02T22:38:07Z' into aware UTC datetime.
Returns None if invalid.
"""
if not isinstance(s, str) or not s:
return None
# Strictly support trailing 'Z' (UTC) to avoid locale/timezone ambiguity.
if not s.endswith("Z"):
return None
try:
dt = datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ")
return dt.replace(tzinfo=timezone.utc)
except ValueError:
return None
def _sort_key(item: dict[str, Any]) -> Tuple[int, datetime, str]:
"""
Sort by:
1) Has started_at meta (1) before missing (0)
2) started_at descending
3) dirname descending (lexicographic)
"""
dt = _parse_iso_z(item.get("started_at"))
has_dt = 1 if dt is not None else 0
# Use epoch for missing to keep key comparable; has_dt separates them anyway.
dt2 = dt if dt is not None else datetime.fromtimestamp(0, tz=timezone.utc)
dirname = item.get("dirname") or ""
return (has_dt, dt2, dirname)
def run_snapshots_list(prefix: Path, host: str, kind: str, limit: int, include_incomplete: bool) -> dict[str, Any]:
host = sanitize_host(host)
k = normalize_kind(kind)
if limit < 1:
raise ConfigError("--limit must be >= 1")
paths = PobsyncPaths(home=prefix)
global_cfg = load_global_config(paths.global_config_path)
host_cfg = load_host_config(paths.hosts_dir / f"{host}.yaml")
cfg = build_effective_config(global_cfg, host_cfg)
backup_root = cfg.get("backup_root")
if not isinstance(backup_root, str) or not backup_root.startswith("/"):
raise ConfigError("Invalid backup_root in effective config")
host_root = resolve_host_root(backup_root, host)
kinds: list[str]
if k == "all":
kinds = ["scheduled", "manual"]
if include_incomplete:
kinds.append("incomplete")
else:
kinds = [k]
items: list[dict[str, Any]] = []
for kk in kinds:
for d in iter_snapshot_dirs(host_root, kk):
meta = read_snapshot_meta(d)
items.append(
{
"kind": kk,
"dirname": d.name,
"path": str(d),
"status": meta.get("status"),
"started_at": meta.get("started_at"),
"ended_at": meta.get("ended_at"),
"duration_seconds": meta.get("duration_seconds"),
"base": meta.get("base"),
"id": meta.get("id"),
}
)
# Global sort: newest first
items.sort(key=_sort_key, reverse=True)
# Apply limit after sorting
out = items[:limit]
return {
"ok": True,
"host": host,
"kind": k,
"include_incomplete": bool(include_incomplete),
"limit": limit,
"snapshots": out,
}

81
src/pobsync/commands/snapshots_show.py
View File

@@ -1,81 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Any, List
from ..config.load import load_global_config, load_host_config
from ..config.merge import build_effective_config
from ..errors import ConfigError
from ..paths import PobsyncPaths
from ..snapshot_meta import (
build_snapshot_ref,
normalize_kind,
read_snapshot_meta,
resolve_host_root,
snapshot_log_path,
)
from ..util import sanitize_host
def _tail_lines(path: Path, n: int) -> List[str]:
"""
Read last n lines of a text file.
Simple and safe; rsync logs are not huge in normal cases.
"""
try:
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
return lines[-n:]
except OSError:
return []
def run_snapshots_show(
prefix: Path,
host: str,
kind: str,
dirname: str,
tail: int | None,
) -> dict[str, Any]:
host = sanitize_host(host)
k = normalize_kind(kind)
if k == "all":
raise ConfigError("kind must be scheduled, manual, or incomplete for show")
if tail is not None and tail < 1:
raise ConfigError("--tail must be >= 1")
paths = PobsyncPaths(home=prefix)
global_cfg = load_global_config(paths.global_config_path)
host_cfg = load_host_config(paths.hosts_dir / f"{host}.yaml")
cfg = build_effective_config(global_cfg, host_cfg)
backup_root = cfg.get("backup_root")
if not isinstance(backup_root, str) or not backup_root.startswith("/"):
raise ConfigError("Invalid backup_root in effective config")
host_root = resolve_host_root(backup_root, host)
ref = build_snapshot_ref(host=host, host_root=host_root, kind=k, dirname=dirname)
if not ref.path.exists():
raise ConfigError(f"Snapshot not found: {k}/{dirname}")
meta = read_snapshot_meta(ref.path)
log_path = snapshot_log_path(ref.path)
log_tail = None
if tail is not None and log_path.exists():
log_tail = _tail_lines(log_path, tail)
return {
"ok": True,
"host": host,
"kind": k,
"dirname": dirname,
"path": str(ref.path),
"meta_path": str(ref.path / "meta" / "meta.yaml"),
"log_path": str(log_path) if log_path.exists() else None,
"meta": meta,
"log_tail": log_tail,
}

32
src/pobsync/config/defaults.py Normal file
View File

@@ -0,0 +1,32 @@
from __future__ import annotations
DEFAULT_EXCLUDES = [
"/proc/***",
"/sys/***",
"/dev/***",
"/run/***",
"/tmp/***",
"/mnt/***",
"/media/***",
"/lost+found/***",
"/var/cache/***",
"/var/tmp/***",
"/var/run/***",
"/var/lock/***",
"/swapfile",
"/.snapshots/***",
]
DEFAULT_RSYNC_ARGS = [
"--archive",
"--numeric-ids",
"--delete",
"--delete-excluded",
"--partial",
"--partial-dir=.rsync-partial",
"--one-file-system",
"--relative",
"--human-readable",
"--stats",
]

21
src/pobsync/config/retention.py Normal file
View File

@@ -0,0 +1,21 @@
from __future__ import annotations
def parse_retention(s: str) -> dict[str, int]:
out: dict[str, int] = {}
parts = [p.strip() for p in s.split(",") if p.strip()]
for part in parts:
if "=" not in part:
raise ValueError(f"Invalid retention component: {part!r}")
k, v = part.split("=", 1)
k = k.strip()
v = v.strip()
if k not in {"daily", "weekly", "monthly", "yearly"}:
raise ValueError(f"Invalid retention key: {k!r}")
n = int(v)
if n < 0:
raise ValueError(f"Retention must be >= 0 for {k}")
out[k] = n
for k in ("daily", "weekly", "monthly", "yearly"):
out.setdefault(k, 0)
return out

228
src/pobsync/doctor_scheduling.py
View File

@@ -1,228 +0,0 @@
from __future__ import annotations
import os
import shutil
import stat
import subprocess
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
CRON_FILE_DEFAULT = "/etc/cron.d/pobsync"
LOG_DIR_DEFAULT = "/var/log/pobsync"
@dataclass(frozen=True)
class DoctorCheck:
name: str
ok: bool
severity: str # "error" | "warning" | "info"
message: str
details: Optional[Dict[str, Any]] = None
def _run(cmd: List[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(
cmd,
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
def _check_cron_service() -> DoctorCheck:
systemctl = shutil.which("systemctl")
if not systemctl:
return DoctorCheck(
name="schedule.cron_service",
ok=True,
severity="warning",
message="systemctl not found; cannot verify cron service status",
details={"hint": "If cron isn't running, schedules won't execute."},
)
# Try both common service names
for svc in ("cron", "crond"):
cp = _run([systemctl, "is-active", svc])
if cp.returncode == 0 and cp.stdout.strip() == "active":
return DoctorCheck(
name="schedule.cron_service",
ok=True,
severity="info",
message=f"cron service is active ({svc})",
)
# Not active / unknown
return DoctorCheck(
name="schedule.cron_service",
ok=False,
severity="error",
message="cron service is not active (tried: cron, crond)",
details={"hint": "Enable/start cron (systemctl enable --now cron) or the equivalent on your distro."},
)
def _check_cron_file_permissions(cron_file: str) -> DoctorCheck:
try:
st = os.stat(cron_file)
except FileNotFoundError:
return DoctorCheck(
name="schedule.cron_file",
ok=True,
severity="warning",
message=f"cron file not found: {cron_file}",
details={"hint": "Create one via: pobsync schedule create <host> ..."},
)
except OSError as e:
return DoctorCheck(
name="schedule.cron_file",
ok=False,
severity="error",
message=f"cannot stat cron file: {cron_file}",
details={"error": str(e)},
)
if not stat.S_ISREG(st.st_mode):
return DoctorCheck(
name="schedule.cron_file",
ok=False,
severity="error",
message=f"cron file is not a regular file: {cron_file}",
)
problems: List[str] = []
if st.st_uid != 0:
problems.append("owner is not root")
# For /etc/cron.d, file must NOT be group/other writable.
# (Mode may be 600 or 644; both are fine as long as not writable by group/other.)
if (st.st_mode & 0o022) != 0:
problems.append("cron file is writable by group/other (must not be)")
mode_octal = oct(st.st_mode & 0o777)
if problems:
return DoctorCheck(
name="schedule.cron_file",
ok=False,
severity="error",
message=f"cron file permissions/ownership look unsafe: {cron_file}",
details={"mode": mode_octal, "uid": st.st_uid, "problems": problems},
)
return DoctorCheck(
name="schedule.cron_file",
ok=True,
severity="info",
message=f"cron file permissions/ownership OK: {cron_file}",
details={"mode": mode_octal},
)
def _check_log_dir(log_dir: str) -> DoctorCheck:
if not os.path.exists(log_dir):
return DoctorCheck(
name="schedule.log_dir",
ok=True,
severity="warning",
message=f"log directory does not exist: {log_dir}",
details={"hint": "Not fatal, but cron output redirection may fail. Backlog item: create in install."},
)
if not os.path.isdir(log_dir):
return DoctorCheck(
name="schedule.log_dir",
ok=False,
severity="error",
message=f"log path exists but is not a directory: {log_dir}",
)
if not os.access(log_dir, os.W_OK):
return DoctorCheck(
name="schedule.log_dir",
ok=False,
severity="error",
message=f"log directory is not writable: {log_dir}",
)
return DoctorCheck(
name="schedule.log_dir",
ok=True,
severity="info",
message=f"log directory OK: {log_dir}",
)
def _check_pobsync_executable(prefix: str) -> DoctorCheck:
exe = os.path.join(prefix, "bin", "pobsync")
if not os.path.exists(exe):
return DoctorCheck(
name="schedule.pobsync_executable",
ok=False,
severity="error",
message=f"pobsync executable not found at {exe}",
details={"hint": "Your cron entry likely points here; verify /opt/pobsync installation."},
)
if not os.access(exe, os.X_OK):
return DoctorCheck(
name="schedule.pobsync_executable",
ok=False,
severity="error",
message=f"pobsync exists but is not executable: {exe}",
)
return DoctorCheck(
name="schedule.pobsync_executable",
ok=True,
severity="info",
message=f"pobsync executable OK: {exe}",
)
def scheduling_checks(prefix: str, cron_file: str = CRON_FILE_DEFAULT) -> List[DoctorCheck]:
return [
_check_cron_service(),
_check_cron_file_permissions(cron_file),
_check_log_dir(LOG_DIR_DEFAULT),
_check_pobsync_executable(prefix),
]
def extend_doctor_result(result: Dict[str, Any], *, prefix: str, cron_file: str = CRON_FILE_DEFAULT) -> Dict[str, Any]:
"""
Add scheduling-related checks into an existing doctor result dict.
This is designed to be additive and low-risk:
- If result has a "checks" list, we append items.
- If result has "ok", we AND it with any error-level failures.
"""
checks = scheduling_checks(prefix=prefix, cron_file=cron_file)
# Normalize result structure
existing = result.get("checks")
if not isinstance(existing, list):
existing = []
result["checks"] = existing
for c in checks:
existing.append(
{
"name": c.name,
"ok": c.ok,
"severity": c.severity,
"message": c.message,
"details": c.details or {},
}
)
# Update overall ok: errors make it false; warnings do not.
overall_ok = bool(result.get("ok", True))
for c in checks:
if c.severity == "error" and not c.ok:
overall_ok = False
result["ok"] = overall_ok
return result

235
src/pobsync/schedule.py
View File

@@ -1,235 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
CRON_FILE_DEFAULT = "/etc/cron.d/pobsync"
BEGIN_PREFIX = "# BEGIN POBSYNC host="
END_PREFIX = "# END POBSYNC host="
@dataclass(frozen=True)
class ScheduleBlock:
host: str
raw_lines: List[str] # full block including begin/end markers
cron_expr: Optional[str] # "m h dom mon dow"
user: Optional[str]
command: Optional[str]
log_path: Optional[str]
def normalize_cron_expr(expr: str) -> str:
return " ".join(expr.strip().split())
def validate_cron_expr(expr: str) -> None:
parts = normalize_cron_expr(expr).split(" ")
if len(parts) != 5:
raise ValueError("cron expression must have exactly 5 fields (m h dom mon dow)")
def parse_hhmm(s: str) -> Tuple[int, int]:
s = s.strip()
if ":" not in s:
raise ValueError("time must be HH:MM")
a, b = s.split(":", 1)
if not a.isdigit() or not b.isdigit():
raise ValueError("time must be HH:MM")
h = int(a)
m = int(b)
if h < 0 or h > 23:
raise ValueError("hour must be 0..23")
if m < 0 or m > 59:
raise ValueError("minute must be 0..59")
return h, m
def parse_dow(s: str) -> int:
"""
Accept: mon,tue,wed,thu,fri,sat,sun (case-insensitive)
Return cron day-of-week number: 0=sun, 1=mon, ... 6=sat
"""
x = s.strip().lower()
mapping = {
"sun": 0,
"mon": 1,
"tue": 2,
"wed": 3,
"thu": 4,
"fri": 5,
"sat": 6,
}
if x not in mapping:
raise ValueError("dow must be one of: mon,tue,wed,thu,fri,sat,sun")
return mapping[x]
def build_cron_expr_daily(hhmm: str) -> str:
h, m = parse_hhmm(hhmm)
return f"{m} {h} * * *"
def build_cron_expr_hourly(minute: int = 0) -> str:
if minute < 0 or minute > 59:
raise ValueError("minute must be 0..59")
return f"{minute} * * * *"
def build_cron_expr_weekly(dow: str, hhmm: str) -> str:
h, m = parse_hhmm(hhmm)
dow_num = parse_dow(dow)
return f"{m} {h} * * {dow_num}"
def build_cron_expr_monthly(day: int, hhmm: str) -> str:
if day < 1 or day > 31:
raise ValueError("day must be 1..31")
h, m = parse_hhmm(hhmm)
return f"{m} {h} {day} * *"
def render_host_block(
host: str,
cron_expr: str,
user: str,
command: str,
log_path: Optional[str],
include_env: bool = True,
) -> str:
validate_cron_expr(cron_expr)
cron_expr = normalize_cron_expr(cron_expr)
lines: List[str] = []
lines.append(f"{BEGIN_PREFIX}{host}")
lines.append("# managed-by=pobsync")
if include_env:
lines.append("SHELL=/bin/sh")
lines.append("PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin")
cron_line = f"{cron_expr} {user} {command}"
if log_path:
cron_line += f" >>{log_path} 2>&1"
lines.append(cron_line)
lines.append(f"{END_PREFIX}{host}")
return "\n".join(lines) + "\n"
def parse_cron_file(content: str) -> Dict[str, ScheduleBlock]:
blocks: Dict[str, ScheduleBlock] = {}
lines = content.splitlines()
i = 0
while i < len(lines):
line = lines[i]
if line.startswith(BEGIN_PREFIX):
host = line[len(BEGIN_PREFIX) :].strip()
block_lines = [line]
i += 1
while i < len(lines):
block_lines.append(lines[i])
if lines[i].strip() == f"{END_PREFIX}{host}":
break
i += 1
cron_expr, user, command, log_path = _extract_cron_line(block_lines)
blocks[host] = ScheduleBlock(
host=host,
raw_lines=block_lines,
cron_expr=cron_expr,
user=user,
command=command,
log_path=log_path,
)
i += 1
return blocks
def _extract_cron_line(block_lines: List[str]) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
for raw in block_lines:
line = raw.strip()
if not line:
continue
if line.startswith("#"):
continue
# skip env-like lines
if "=" in line and line.split("=", 1)[0].isidentifier():
continue
parts = line.split()
if len(parts) < 7:
continue
cron_expr = " ".join(parts[0:5])
user = parts[5]
cmd = " ".join(parts[6:])
log_path = None
if ">>" in cmd:
before, after = cmd.split(">>", 1)
cmd = before.rstrip()
after_parts = after.strip().split()
if after_parts:
log_path = after_parts[0]
return cron_expr, user, cmd, log_path
return None, None, None, None
def upsert_host_block(content: str, host: str, new_block: str) -> str:
lines = content.splitlines()
out: List[str] = []
i = 0
replaced = False
begin = f"{BEGIN_PREFIX}{host}"
end = f"{END_PREFIX}{host}"
while i < len(lines):
if lines[i].strip() == begin:
replaced = True
# skip until end marker (inclusive)
i += 1
while i < len(lines) and lines[i].strip() != end:
i += 1
if i < len(lines):
i += 1 # skip end marker
out.extend(new_block.rstrip("\n").splitlines())
continue
out.append(lines[i])
i += 1
if not replaced:
if out and out[-1].strip() != "":
out.append("")
out.extend(new_block.rstrip("\n").splitlines())
return "\n".join(out).rstrip() + "\n"
def remove_host_block(content: str, host: str) -> str:
lines = content.splitlines()
out: List[str] = []
i = 0
begin = f"{BEGIN_PREFIX}{host}"
end = f"{END_PREFIX}{host}"
while i < len(lines):
if lines[i].strip() == begin:
i += 1
while i < len(lines) and lines[i].strip() != end:
i += 1
if i < len(lines):
i += 1 # skip end marker
continue
out.append(lines[i])
i += 1
return "\n".join(out).rstrip() + "\n"

4
src/pobsync_backend/management/commands/configure_pobsync_global.py
View File

@@ -6,8 +6,8 @@ from typing import Any
from django.conf import settings
from django.core.management.base import BaseCommand, CommandError
from pobsync.cli import parse_retention
from pobsync.commands.install import DEFAULT_EXCLUDES, DEFAULT_RSYNC_ARGS
from pobsync.config.retention import parse_retention
from pobsync.config.defaults import DEFAULT_EXCLUDES, DEFAULT_RSYNC_ARGS
from pobsync.util import is_absolute_non_root
from pobsync_backend.models import GlobalConfig

2
src/pobsync_backend/management/commands/configure_pobsync_host.py
View File

@@ -4,7 +4,7 @@ from typing import Any
from django.core.management.base import BaseCommand, CommandError
from pobsync.cli import parse_retention
from pobsync.config.retention import parse_retention
from pobsync.util import sanitize_host
from pobsync_backend.models import GlobalConfig, HostConfig

55
src/pobsync_backend/management/commands/configure_pobsync_schedule.py Normal file
View File

@@ -0,0 +1,55 @@
from __future__ import annotations
from typing import Any
from django.core.management.base import BaseCommand, CommandError
from pobsync_backend.models import HostConfig, ScheduleConfig
from pobsync_backend.scheduler import parse_cron_expr
class Command(BaseCommand):
help = "Create, update, disable, or remove a SQL-backed pobsync schedule."
def add_arguments(self, parser) -> None:
parser.add_argument("host")
parser.add_argument("--cron", help='Cron expression, e.g. "15 2 * * *"')
parser.add_argument("--user", default="root")
parser.add_argument("--prune", action="store_true")
parser.add_argument("--prune-max-delete", type=int, default=10)
parser.add_argument("--prune-protect-bases", action="store_true")
parser.add_argument("--disabled", action="store_true")
parser.add_argument("--delete", action="store_true")
def handle(self, *args: Any, **options: Any) -> None:
try:
host = HostConfig.objects.get(host=options["host"])
except HostConfig.DoesNotExist as exc:
raise CommandError(f"Missing HostConfig {options['host']!r}") from exc
if options["delete"]:
deleted, _details = ScheduleConfig.objects.filter(host=host).delete()
self.stdout.write(self.style.SUCCESS(f"Deleted {deleted} schedule row(s) for {host.host!r}."))
return
if not options["cron"]:
raise CommandError("--cron is required unless --delete is used")
try:
parse_cron_expr(options["cron"])
except ValueError as exc:
raise CommandError(str(exc)) from exc
schedule, created = ScheduleConfig.objects.update_or_create(
host=host,
defaults={
"cron_expr": options["cron"],
"user": options["user"],
"enabled": not options["disabled"],
"prune": bool(options["prune"]),
"prune_max_delete": int(options["prune_max_delete"]),
"prune_protect_bases": bool(options["prune_protect_bases"]),
},
)
action = "Created" if created else "Updated"
state = "enabled" if schedule.enabled else "disabled"
self.stdout.write(self.style.SUCCESS(f"{action} {state} schedule for {host.host!r}."))

55
src/pobsync_backend/management/commands/run_pobsync_retention.py Normal file
View File

@@ -0,0 +1,55 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from django.conf import settings
from django.core.management.base import BaseCommand, CommandError
from pobsync.commands.retention_apply import run_retention_apply
from pobsync.commands.retention_plan import run_retention_plan
from pobsync_backend.config_source import DjangoConfigSource
from pobsync_backend.models import HostConfig
class Command(BaseCommand):
help = "Plan or apply retention using SQL-backed pobsync configuration."
def add_arguments(self, parser) -> None:
parser.add_argument("host")
parser.add_argument("--prefix", default=settings.POBSYNC_HOME)
parser.add_argument("--kind", default="scheduled", choices=["scheduled", "manual", "all"])
parser.add_argument("--protect-bases", action="store_true")
parser.add_argument("--apply", action="store_true")
parser.add_argument("--yes", action="store_true")
parser.add_argument("--max-delete", type=int, default=10)
def handle(self, *args: Any, **options: Any) -> None:
host = options["host"]
if not HostConfig.objects.filter(host=host, enabled=True).exists():
raise CommandError(f"Missing enabled HostConfig {host!r}")
config_source = DjangoConfigSource()
if options["apply"]:
if not options["yes"]:
raise CommandError("--yes is required with --apply")
result = run_retention_apply(
prefix=Path(options["prefix"]),
host=host,
kind=options["kind"],
protect_bases=bool(options["protect_bases"]),
yes=True,
max_delete=int(options["max_delete"]),
config_source=config_source,
)
else:
result = run_retention_plan(
prefix=Path(options["prefix"]),
host=host,
kind=options["kind"],
protect_bases=bool(options["protect_bases"]),
config_source=config_source,
)
self.stdout.write(json.dumps(result, indent=2, sort_keys=False))

18
src/pobsync_backend/tests/test_configure_commands.py
View File

@@ -6,7 +6,7 @@ from django.core.management import call_command
from django.test import TestCase
from pobsync_backend.config_source import DjangoConfigSource
from pobsync_backend.models import GlobalConfig, HostConfig
from pobsync_backend.models import GlobalConfig, HostConfig, ScheduleConfig
class ConfigureCommandsTests(TestCase):
@@ -54,3 +54,19 @@ class ConfigureCommandsTests(TestCase):
effective = DjangoConfigSource().effective_config_for_host("web-01")
self.assertEqual(effective["retention"]["yearly"], 2)
self.assertEqual(effective["excludes_effective"], ["/tmp/***"])
def test_configure_schedule_creates_sql_schedule(self) -> None:
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
out = StringIO()
call_command(
"configure_pobsync_schedule",
host.host,
cron="15 2 * * *",
prune=True,
stdout=out,
)
schedule = ScheduleConfig.objects.get(host=host)
self.assertEqual(schedule.cron_expr, "15 2 * * *")
self.assertTrue(schedule.prune)

41
src/pobsync_backend/tests/test_console_entrypoint.py Normal file
View File

@@ -0,0 +1,41 @@
from __future__ import annotations
from io import StringIO
from unittest.mock import patch
from django.test import SimpleTestCase
from pobsync.cli import main
class ConsoleEntrypointTests(SimpleTestCase):
def test_maps_backup_alias_to_django_command(self) -> None:
with patch("pobsync.cli.execute_from_command_line") as execute:
exit_code = main(["backup", "web-01", "--dry-run"])
self.assertEqual(exit_code, 0)
execute.assert_called_once_with(["pobsync", "run_pobsync_backup", "web-01", "--dry-run"])
def test_unknown_command_returns_usage_error(self) -> None:
stderr = StringIO()
with patch("sys.stderr", stderr):
exit_code = main(["run-scheduled", "web-01"])
self.assertEqual(exit_code, 2)
self.assertIn("Unknown pobsync command", stderr.getvalue())
def test_django_passthrough_keeps_management_command_name(self) -> None:
with patch("pobsync.cli.execute_from_command_line") as execute:
exit_code = main(["django", "check"])
self.assertEqual(exit_code, 0)
execute.assert_called_once_with(["pobsync", "check"])
def test_maps_schedule_alias_to_django_command(self) -> None:
with patch("pobsync.cli.execute_from_command_line") as execute:
exit_code = main(["schedule", "web-01", "--cron", "15 2 * * *"])
self.assertEqual(exit_code, 0)
execute.assert_called_once_with(
["pobsync", "configure_pobsync_schedule", "web-01", "--cron", "15 2 * * *"]
)