mirror of
https://github.com/Joulenap/joulenap.git
synced 2026-08-11 13:21:43 +02:00
feat(i18n): localize run step details, and notify when a PBS never wakes
Step details were stored as English sentences, so the run timeline stayed English whatever the UI language was. `run_steps` now carries a `detail_key`/`detail_params` pair alongside the text, the same seam `runs.error_key` already had: the English rendering stays on the row for pre-1.0 rows and for foreign strings (a task UPID, another product's error text), while the API rebuilds Joulenap's own copy in the user's language on read. `set_detail()` writes both halves in one call so they cannot drift. Fixes carried in the same files: - a run whose PBS never came up recorded a history row and nothing else. The unreachable handler re-raised, unwinding past both the lease release and the notification, so the one failure this product exists to have an opinion about sent no Telegram, no ntfy, no email — a regression against 0.9, which did notify. On a sync route it also left an already-woken box burning power with no POWEROFF step. It now falls through the same release-and-notify path a completed run uses. - the notification's duration breakdown dropped the backup phase on every real backup route: the lookup matched bare step names, but a route labels its backup step per source (`backup:pve-alpha`). Matched with `_step_is` now, and summed per phase. - a PBS sync remote outlived the route that created it, parking the peer's full API token in the executing box's `remote.cfg`. Remote and sync job are both removed once the sync is over. - a step that set a detail and then failed rendered the detail instead of the error, because a resolvable key wins over the raw string.
This commit is contained in:
Catubba
+17
-10
@@ -8,20 +8,20 @@ from datetime import datetime
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from ..db.models import LogEvent, Run, RunStep, TaskLogLine
|
||||
from ..notify.messages import render_error
|
||||
from ..notify.messages import render_detail, render_error
|
||||
|
||||
|
||||
def _error_params(run: Run) -> dict[str, object] | None:
|
||||
"""``run.error_params`` decoded, or ``None`` if it is absent or unreadable.
|
||||
def _params(raw: str | None) -> dict[str, object] | None:
|
||||
"""A stored ``*_params`` column decoded, or ``None`` if it is absent or unreadable.
|
||||
|
||||
Never raises: the column is free-form JSON written by an older version of the app, and a
|
||||
history page must not 500 because one row's payload is malformed. ``render_error`` falls
|
||||
history page must not 500 because one row's payload is malformed. The renderers fall
|
||||
back to the stored English text in that case.
|
||||
"""
|
||||
if not run.error_params:
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
params = json.loads(run.error_params)
|
||||
params = json.loads(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
return params if isinstance(params, dict) else None
|
||||
@@ -51,7 +51,7 @@ class RunSummary(BaseModel):
|
||||
@classmethod
|
||||
def of(cls, run: Run, language: str = "en") -> RunSummary:
|
||||
summary = cls.model_validate(run)
|
||||
summary.error = render_error(language, run.error_key, _error_params(run), run.error)
|
||||
summary.error = render_error(language, run.error_key, _params(run.error_params), run.error)
|
||||
return summary
|
||||
|
||||
|
||||
@@ -76,11 +76,18 @@ class StepInfo(BaseModel):
|
||||
status: str
|
||||
started_at: datetime
|
||||
finished_at: datetime | None
|
||||
#: Already rendered in ``app.language``, like ``RunSummary.error``: the row stores
|
||||
#: ``detail_key``/``detail_params`` and keeps the English ``detail`` as the fallback for
|
||||
#: pre-1.0 rows, task ids and other software's error text.
|
||||
detail: str | None = None
|
||||
|
||||
@classmethod
|
||||
def of(cls, step: RunStep) -> StepInfo:
|
||||
return cls.model_validate(step)
|
||||
def of(cls, step: RunStep, language: str = "en") -> StepInfo:
|
||||
info = cls.model_validate(step)
|
||||
info.detail = render_detail(
|
||||
language, step.detail_key, _params(step.detail_params), step.detail
|
||||
)
|
||||
return info
|
||||
|
||||
|
||||
class TaskLogLineSchema(BaseModel):
|
||||
@@ -117,6 +124,6 @@ class RunDetail(RunSummary):
|
||||
def of(cls, run: Run, language: str = "en") -> RunDetail:
|
||||
return cls(
|
||||
**RunSummary.of(run, language).model_dump(),
|
||||
steps=[StepInfo.of(s) for s in run.steps],
|
||||
steps=[StepInfo.of(s, language) for s in run.steps],
|
||||
logs=[LogLine.of(e) for e in run.logs],
|
||||
)
|
||||
|
||||
@@ -237,6 +237,18 @@ class PbsClient:
|
||||
if any(e.get("id") == job_id for e in existing):
|
||||
self._api.request("DELETE", f"/config/sync/{job_id}")
|
||||
|
||||
def delete_remote(self, name: str) -> None:
|
||||
"""Drop the remote ``name`` if it exists.
|
||||
|
||||
A remote holds the *other* PBS's API token — id and secret — in that box's
|
||||
``remote.cfg``, so it must not outlive the route that created it: revoking the
|
||||
credential in Joulenap otherwise leaves a working copy of it on the peer. Call
|
||||
:meth:`delete_sync_job` first; PBS refuses to delete a remote a job still references.
|
||||
"""
|
||||
existing = self._api.request("GET", "/config/remote") or []
|
||||
if any((e.get("name") or e.get("id")) == name for e in existing):
|
||||
self._api.request("DELETE", f"/config/remote/{name}")
|
||||
|
||||
def ensure_sync_job(
|
||||
self,
|
||||
job_id: str,
|
||||
|
||||
@@ -161,6 +161,12 @@ class RunStep(Base):
|
||||
started_at: Mapped[datetime] = mapped_column(UtcDateTime(), default=_utcnow)
|
||||
finished_at: Mapped[datetime | None] = mapped_column(UtcDateTime(), default=None)
|
||||
detail: Mapped[str | None] = mapped_column(Text, default=None)
|
||||
# The localisation seam, same shape as ``Run.error_key``/``error_params``: ``detail``
|
||||
# keeps the English rendering (and is all a pre-1.0 row has), while the key and its
|
||||
# JSON parameters let the API rebuild the line in the user's language on read. Only
|
||||
# details Joulenap authored carry a key — a task UPID has none.
|
||||
detail_key: Mapped[str | None] = mapped_column(String(64), default=None)
|
||||
detail_params: Mapped[str | None] = mapped_column(Text, default=None)
|
||||
|
||||
run: Mapped[Run] = relationship(back_populates="steps")
|
||||
|
||||
|
||||
@@ -41,6 +41,7 @@ def sweep_orphaned_runs(session: Session, *, now: datetime | None = None) -> lis
|
||||
step.finished_at = ts
|
||||
if not step.detail:
|
||||
step.detail = _INTERRUPTED_STEP
|
||||
step.detail_key = "interrupted"
|
||||
return list(orphaned)
|
||||
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ from ..db.guest_backups import upsert_last_backups
|
||||
from ..db.models import LogLevel, RunStatus, StepName
|
||||
from ..notify.messages import GuestSummary, LocalizedError, RunContext
|
||||
from .deps import CycleDeps
|
||||
from .recorder import RunRecorder
|
||||
from .recorder import RunRecorder, set_detail
|
||||
|
||||
# Poll cadence while tailing a task's log — snappier than the plain wait default so the
|
||||
# live Task-log panel narrates in near-real-time (Proxmox has no push API).
|
||||
@@ -326,7 +326,9 @@ def _route_preflight(
|
||||
with deps.connect_pbs(target) as pbs:
|
||||
ds = pbs.datastore_status()
|
||||
_cache_route_datastore(target, recorder, ds)
|
||||
step.detail = f"{ds.avail_pct:.1f}% free ({ds.avail / 1_000_000_000:.0f} GB)"
|
||||
set_detail(
|
||||
step, "free_space", free=f"{ds.avail_pct:.1f}", avail=f"{ds.avail / 1_000_000_000:.0f}"
|
||||
)
|
||||
if ds.avail_pct < threshold:
|
||||
raise CycleAbort(
|
||||
"datastore_full",
|
||||
@@ -472,7 +474,7 @@ def run_route_backup(
|
||||
if route.options.gc:
|
||||
_route_gc_step(target, recorder, deps)
|
||||
else:
|
||||
recorder.skip_step(StepName.GC, "GC disabled for this route")
|
||||
recorder.skip_step(StepName.GC, "gc_disabled")
|
||||
|
||||
if deps.cancelled():
|
||||
raise CycleCancelled("Run cancelled")
|
||||
@@ -480,7 +482,7 @@ def run_route_backup(
|
||||
if route.options.verify_after:
|
||||
_route_verify_step(target, recorder, deps, outdated_after=None)
|
||||
else:
|
||||
recorder.skip_step(StepName.VERIFY, "verify disabled for this route")
|
||||
recorder.skip_step(StepName.VERIFY, "verify_disabled")
|
||||
|
||||
datastore = _route_read_datastore(target, recorder, deps)
|
||||
_refresh_route_backup_cache(target, covered, recorder, deps)
|
||||
|
||||
@@ -40,7 +40,8 @@ class ReleaseOutcome(StrEnum):
|
||||
Only :attr:`LEFT_ON` means "still burning power, and nothing is going to fix that" —
|
||||
the others are boxes Joulenap either put to sleep, never powers, or will power off at
|
||||
the end of the run that still holds them. The value doubles as the POWEROFF step's
|
||||
detail in the run timeline.
|
||||
English detail in the run timeline; :attr:`key` is the same fact as a translatable
|
||||
code, which is what the step actually stores alongside it.
|
||||
"""
|
||||
|
||||
POWERED_OFF = "powered off"
|
||||
@@ -48,6 +49,15 @@ class ReleaseOutcome(StrEnum):
|
||||
UNMANAGED = "left on: Joulenap does not manage this box's power"
|
||||
LEFT_ON = "left powered on"
|
||||
|
||||
@property
|
||||
def key(self) -> str:
|
||||
"""This outcome's key in ``notify.messages._DETAILS`` — the member name, lowercased.
|
||||
|
||||
Derived rather than mapped so a new member cannot be added without a key: the
|
||||
i18n parity test walks the enum and fails on any member the packs don't cover.
|
||||
"""
|
||||
return self.name.lower()
|
||||
|
||||
|
||||
# --- device-shaped connector calls -------------------------------------------
|
||||
# jobs/deps.py has the same four operations bound to the 0.9 single-PBS ``Config``. These
|
||||
|
||||
@@ -27,7 +27,7 @@ from ..db.models import (
|
||||
StepStatus,
|
||||
TaskLogLine,
|
||||
)
|
||||
from ..notify.messages import LocalizedError
|
||||
from ..notify.messages import LocalizedError, render_detail
|
||||
|
||||
|
||||
def _utcnow() -> datetime:
|
||||
@@ -48,6 +48,18 @@ def _error_code(error: str | BaseException) -> tuple[str | None, dict[str, objec
|
||||
return None, None
|
||||
|
||||
|
||||
def set_detail(step: RunStep, key: str, /, **params: object) -> None:
|
||||
"""Set a step's ``detail`` from the ``_DETAILS`` catalogue, in English and as a code.
|
||||
|
||||
Both halves in one call, so the two can never drift: ``detail`` keeps the English
|
||||
sentence every existing reader (and every pre-1.0 row) expects, while the key and its
|
||||
parameters let ``api.schemas.StepInfo`` rebuild the line in the user's language.
|
||||
"""
|
||||
step.detail = render_detail("en", key, params) or key
|
||||
step.detail_key = key
|
||||
step.detail_params = json.dumps(params) if params else None
|
||||
|
||||
|
||||
class RunRecorder:
|
||||
"""Records a single run. Use as a context manager so the session is always closed
|
||||
and an unfinished run is marked failed even if the caller crashes unexpectedly."""
|
||||
@@ -130,7 +142,13 @@ class RunRecorder:
|
||||
except Exception as exc:
|
||||
step.status = StepStatus.FAILURE
|
||||
step.finished_at = _utcnow()
|
||||
# Clear the key as well as the text: a step that set a localized detail and *then*
|
||||
# failed (the pre-flight guard reports free space, then aborts on it) would
|
||||
# otherwise keep rendering that detail — ``render_detail`` prefers a resolvable
|
||||
# key over the raw string, so the failure message would never reach the timeline.
|
||||
step.detail = str(exc)
|
||||
step.detail_key = None
|
||||
step.detail_params = None
|
||||
self.log(LogLevel.ERROR, f"{full}: {exc}")
|
||||
self._session.commit()
|
||||
raise
|
||||
@@ -144,22 +162,27 @@ class RunRecorder:
|
||||
self.log(LogLevel.OK, f"{full}: done")
|
||||
self._session.commit()
|
||||
|
||||
def skip_step(self, name: StepName, detail: str | None = None) -> None:
|
||||
"""Record a step that was intentionally not run (e.g. GC when the toggle is off)."""
|
||||
def skip_step(self, name: StepName, detail_key: str | None = None, **params: object) -> None:
|
||||
"""Record a step that was intentionally not run (e.g. GC when the toggle is off).
|
||||
|
||||
Takes a ``_DETAILS`` key rather than a sentence: a skipped step's reason is always
|
||||
Joulenap's own copy, so it is always translatable.
|
||||
"""
|
||||
# Both ends from one clock read: letting ``started_at`` fall back to its column
|
||||
# default means it is evaluated at flush, i.e. *after* this call, and a skipped step
|
||||
# ends up finishing before it started — a negative duration in the timeline.
|
||||
now = _utcnow()
|
||||
self._session.add(
|
||||
RunStep(
|
||||
run_id=self.run.id,
|
||||
name=name,
|
||||
status=StepStatus.SKIPPED,
|
||||
started_at=now,
|
||||
finished_at=now,
|
||||
detail=detail,
|
||||
)
|
||||
step = RunStep(
|
||||
run_id=self.run.id,
|
||||
name=name,
|
||||
status=StepStatus.SKIPPED,
|
||||
started_at=now,
|
||||
finished_at=now,
|
||||
)
|
||||
if detail_key:
|
||||
set_detail(step, detail_key, **params)
|
||||
detail = step.detail
|
||||
self._session.add(step)
|
||||
if detail:
|
||||
self.log(LogLevel.INFO, f"{name.value}: skipped ({detail})")
|
||||
else:
|
||||
|
||||
@@ -15,9 +15,9 @@ from collections.abc import Callable
|
||||
|
||||
from ..config import Config, PbsDevice, Route
|
||||
from ..connectors.errors import TaskError
|
||||
from ..connectors.pbs import DatastoreStatus
|
||||
from ..connectors.pbs import DatastoreStatus, PbsClient
|
||||
from ..db.models import LogLevel, RunKind, RunStatus, StepName
|
||||
from ..notify.messages import LocalizedError, RunContext
|
||||
from ..notify.messages import LocalizedError, RunContext, render_detail
|
||||
from .backup_cycle import (
|
||||
CycleAbort,
|
||||
CycleCancelled,
|
||||
@@ -31,7 +31,7 @@ from .backup_cycle import (
|
||||
watch_external_tasks,
|
||||
)
|
||||
from .deps import CycleDeps
|
||||
from .recorder import RunRecorder
|
||||
from .recorder import RunRecorder, set_detail
|
||||
|
||||
# ``(config, route, target, recorder, deps) -> datastore status for the notification``.
|
||||
RouteBody = Callable[[Config, Route, PbsDevice, RunRecorder, CycleDeps], "DatastoreStatus | None"]
|
||||
@@ -76,6 +76,25 @@ def _task_reason(pbs, upid: str) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
def _drop_sync_config(pbs: PbsClient, name: str, recorder: RunRecorder) -> None:
|
||||
"""Remove the remote and its sync job from the executing PBS once the sync is over.
|
||||
|
||||
A PBS remote stores the **peer's API token, id and secret**, in the executing box's
|
||||
``/etc/proxmox-backup/remote.cfg``. Joulenap rebuilds the pair on every run anyway, so
|
||||
leaving them behind buys nothing and parks a working credential for one backup server on
|
||||
another — one that survives changing the token in Joulenap, and that anyone with root on
|
||||
that box, or a copy of its ``/etc``, can read.
|
||||
|
||||
Job before remote: PBS refuses to delete a remote a job still references. Best-effort and
|
||||
never fatal — it runs in a ``finally``, so raising here would mask the run's real failure.
|
||||
"""
|
||||
try:
|
||||
pbs.delete_sync_job(name)
|
||||
pbs.delete_remote(name)
|
||||
except Exception as exc: # noqa: BLE001 - cleanup must never replace the real outcome
|
||||
recorder.log(LogLevel.WARN, f"could not remove sync config '{name}': {exc}")
|
||||
|
||||
|
||||
def _sync_body(
|
||||
config: Config, route: Route, target: PbsDevice, recorder: RunRecorder, deps: CycleDeps
|
||||
) -> DatastoreStatus | None:
|
||||
@@ -116,34 +135,37 @@ def _sync_body(
|
||||
store=executor.datastore,
|
||||
direction=route.sync_direction,
|
||||
)
|
||||
upid = pbs.run_sync_job(name)
|
||||
step.detail = upid
|
||||
try:
|
||||
_wait_or_stop(pbs, upid, recorder, deps, StepName.SYNC.value, "pbs")
|
||||
except TaskError as exc:
|
||||
# The bare "Task UPID:… finished with status 'WARNINGS: 1'" is unreadable and
|
||||
# is what the run row, the history and the notification all end up showing.
|
||||
reason = _task_reason(pbs, upid)
|
||||
raise LocalizedError(
|
||||
"sync_failed",
|
||||
direction=route.sync_direction,
|
||||
source=source.id,
|
||||
target=target.id,
|
||||
status=exc.exit_status or "unknown status",
|
||||
reason=f": {reason}" if reason else "",
|
||||
) from exc
|
||||
upid = pbs.run_sync_job(name)
|
||||
step.detail = upid
|
||||
try:
|
||||
_wait_or_stop(pbs, upid, recorder, deps, StepName.SYNC.value, "pbs")
|
||||
except TaskError as exc:
|
||||
# The bare "Task UPID:… finished with status 'WARNINGS: 1'" is unreadable
|
||||
# and is what the run row, the history and the notification all show.
|
||||
reason = _task_reason(pbs, upid)
|
||||
raise LocalizedError(
|
||||
"sync_failed",
|
||||
direction=route.sync_direction,
|
||||
source=source.id,
|
||||
target=target.id,
|
||||
status=exc.exit_status or "unknown status",
|
||||
reason=f": {reason}" if reason else "",
|
||||
) from exc
|
||||
finally:
|
||||
_drop_sync_config(pbs, name, recorder)
|
||||
|
||||
# Maintenance runs on the target: it is the box that just gained snapshots, whichever
|
||||
# side pushed or pulled them.
|
||||
if route.options.gc:
|
||||
_route_gc_step(target, recorder, deps)
|
||||
else:
|
||||
recorder.skip_step(StepName.GC, "GC disabled for this route")
|
||||
recorder.skip_step(StepName.GC, "gc_disabled")
|
||||
|
||||
if route.options.verify_after:
|
||||
_route_verify_step(target, recorder, deps, outdated_after=None)
|
||||
else:
|
||||
recorder.skip_step(StepName.VERIFY, "verify disabled for this route")
|
||||
recorder.skip_step(StepName.VERIFY, "verify_disabled")
|
||||
|
||||
# No last-backup cache refresh: a synced snapshot carries no hint of which PVE created
|
||||
# it, and guessing would attribute another PBS's guests to a local one.
|
||||
@@ -153,16 +175,24 @@ def _sync_body(
|
||||
# --- external ----------------------------------------------------------------
|
||||
|
||||
|
||||
def monitor_key(observed: int | None) -> str:
|
||||
"""The MONITOR step's key in ``notify.messages._DETAILS``."""
|
||||
return "no_tasks_observed" if observed is None else "tasks_observed"
|
||||
|
||||
|
||||
def monitor_detail(observed: int | None) -> str:
|
||||
"""The MONITOR step's ``detail``, in the one place that defines its shape.
|
||||
"""The MONITOR step's **English** ``detail``, in the one place that defines its shape.
|
||||
|
||||
``notify.messages.build_run_message`` reads the count back out of this string
|
||||
(``int(detail.split()[0])``, falling through to the "no PBS job ran" warning on
|
||||
``ValueError``), so the format is a contract between two modules rather than free text.
|
||||
A test pins the round-trip; reword it here and that test fails instead of the
|
||||
notification quietly reporting the wrong thing.
|
||||
|
||||
The notifier reads the *stored* English detail, not the translated one, so this contract
|
||||
is unaffected by which language the UI renders the step in.
|
||||
"""
|
||||
return "no tasks observed" if observed is None else f"{observed} task(s) observed"
|
||||
return render_detail("en", monitor_key(observed), {"count": observed}) or ""
|
||||
|
||||
|
||||
def _external_source_pve(config: Config, target: PbsDevice, recorder: RunRecorder) -> str | None:
|
||||
@@ -195,14 +225,14 @@ def _external_body(
|
||||
with deps.connect_pbs(target) as pbs:
|
||||
observed = watch_external_tasks(pbs, target.external, cancelled=deps.cancelled)
|
||||
if observed is None:
|
||||
step.detail = monitor_detail(None)
|
||||
set_detail(step, monitor_key(None))
|
||||
recorder.log(
|
||||
LogLevel.WARN,
|
||||
f"no PBS task appeared within {target.external.first_task_wait}s "
|
||||
"— check the schedules on PVE/PBS; powering off",
|
||||
)
|
||||
else:
|
||||
step.detail = monitor_detail(observed)
|
||||
set_detail(step, monitor_key(observed), count=observed)
|
||||
|
||||
# Someone else's jobs (hopefully) wrote new snapshots — refresh the caches the dashboard
|
||||
# serves while the PBS sleeps, exactly like a managed route does.
|
||||
|
||||
@@ -32,7 +32,7 @@ from ..db.prune import PruneResult, prune_history
|
||||
from ..notify.messages import LocalizedError, RunContext
|
||||
from .deps import CycleDeps
|
||||
from .lease import LeaseDeps, PbsUnreachableError, PowerLease, ReleaseOutcome
|
||||
from .recorder import RunRecorder
|
||||
from .recorder import RunRecorder, set_detail
|
||||
from .route_cycle import RUN_KINDS, run_pbs_maintenance, run_route
|
||||
|
||||
log = logging.getLogger("joulenap.jobs")
|
||||
@@ -299,6 +299,7 @@ class JobService:
|
||||
recorder = self._start(item.kind, item.trigger, route)
|
||||
item.run_id = recorder.run_id
|
||||
held: list[PbsDevice] = []
|
||||
multi = len(devices) > 1 # labels every WAIT/POWEROFF step of this run, consistently
|
||||
try:
|
||||
with recorder:
|
||||
log.info(
|
||||
@@ -306,20 +307,39 @@ class JobService:
|
||||
)
|
||||
try:
|
||||
for device in devices:
|
||||
self._acquire_step(device, recorder, multi=len(devices) > 1)
|
||||
self._acquire_step(device, recorder, multi=multi)
|
||||
held.append(device)
|
||||
except PbsUnreachableError as exc:
|
||||
# A cancel abandons the wake wait the same way a timeout does, so say
|
||||
# which one it was rather than filing every stopped run as a failure.
|
||||
if self._cancel.is_set():
|
||||
cancelled = self._cancel.is_set()
|
||||
if cancelled:
|
||||
recorder.finish(RunStatus.ABORTED, error=LocalizedError("cancelled"))
|
||||
else:
|
||||
recorder.finish(RunStatus.FAILURE, error=exc)
|
||||
raise
|
||||
# Fall through the *same* release-and-notify path a completed run uses,
|
||||
# rather than re-raising. Re-raising unwound past both of them, so
|
||||
# "the backup server never came up" — the one failure this product
|
||||
# exists to have an opinion about — left a history row and nothing
|
||||
# else, no Telegram, no ntfy, no email. 0.9 notified here, so it was a
|
||||
# regression, and the `pbs_unreachable` entry in the message catalogue
|
||||
# was built for a message that could never be sent.
|
||||
#
|
||||
# Releasing matters just as much on a sync route: the target can be
|
||||
# awake when the source never answers, and that box was otherwise left
|
||||
# burning power with no POWEROFF step in the timeline.
|
||||
left_on = self._release_all(item, held, recorder, multi=multi)
|
||||
held = []
|
||||
# A stopped run stays silent on purpose — the user is standing at the UI.
|
||||
if not cancelled:
|
||||
ctx = RunContext(config=config, run=recorder.run, route=route)
|
||||
ctx.left_on = left_on
|
||||
self._notify(ctx, recorder)
|
||||
return
|
||||
# The cycle sets the run's final status itself and hands back what to say
|
||||
# about it (None = cancelled, say nothing).
|
||||
ctx = item.job(config, subject, recorder, self.deps)
|
||||
left_on = self._release_all(item, held, recorder)
|
||||
left_on = self._release_all(item, held, recorder, multi=multi)
|
||||
held = []
|
||||
if ctx is not None:
|
||||
ctx.left_on = left_on
|
||||
@@ -366,10 +386,15 @@ class JobService:
|
||||
"""
|
||||
with recorder.step(StepName.WAIT, label=device.id if multi else None) as step:
|
||||
was_awake = self.lease.acquire(device)
|
||||
step.detail = "already awake" if was_awake else "woken by Wake-on-LAN"
|
||||
set_detail(step, "already_awake" if was_awake else "woken")
|
||||
|
||||
def _release_all(
|
||||
self, item: QueuedRun, devices: list[PbsDevice], recorder: RunRecorder | None
|
||||
self,
|
||||
item: QueuedRun,
|
||||
devices: list[PbsDevice],
|
||||
recorder: RunRecorder | None,
|
||||
*,
|
||||
multi: bool | None = None,
|
||||
) -> list[str]:
|
||||
"""Drop every lease this run holds, recording each power-off decision. Returns the
|
||||
ids of the boxes genuinely left burning power, for the notification's warning.
|
||||
@@ -379,10 +404,16 @@ class JobService:
|
||||
is why the warning cannot be a single fact about the run.
|
||||
``recorder=None`` is the crash path — the leases must still be dropped, but the run
|
||||
row is already being finalised, so there is nothing left to record against.
|
||||
|
||||
``multi`` labels the POWEROFF steps and must describe *the run*, not this call's
|
||||
slice of it: when a wake fails half way, only the boxes that came up are released,
|
||||
and deriving it from ``devices`` here would pair a labelled ``wait:pbs-02`` with a
|
||||
bare ``poweroff`` in the same timeline. The caller passes what ``_acquire_step``
|
||||
used; ``None`` keeps the old derive-it-here behaviour for the crash path.
|
||||
"""
|
||||
succeeded = recorder is not None and recorder.run.status == RunStatus.SUCCESS
|
||||
power_off = self._power_off_policy(item, succeeded=succeeded)
|
||||
multi = len(devices) > 1
|
||||
multi = len(devices) > 1 if multi is None else multi
|
||||
left_on: list[str] = []
|
||||
for device in devices:
|
||||
if recorder is None:
|
||||
@@ -391,7 +422,7 @@ class JobService:
|
||||
continue
|
||||
with recorder.step(StepName.POWEROFF, label=device.id if multi else None) as step:
|
||||
outcome = self.lease.release(device, power_off=power_off)
|
||||
step.detail = str(outcome)
|
||||
set_detail(step, outcome.key)
|
||||
if outcome is ReleaseOutcome.POWERED_OFF:
|
||||
continue
|
||||
# Not a failure: leaving the box on is the *correct* outcome after a failed
|
||||
|
||||
@@ -275,6 +275,44 @@ _ERRORS: dict[str, dict[str, str]] = {
|
||||
},
|
||||
}
|
||||
|
||||
#: Step ``detail`` strings Joulenap authored, i.e. the ones it can translate. The English
|
||||
#: rendering is still stored on the row (``run_steps.detail``) and is the fallback here, the
|
||||
#: same contract ``_ERRORS`` has with ``runs.error``.
|
||||
#:
|
||||
#: Deliberately **not** covered, and deliberately keyless: a PVE/PBS task UPID and the text
|
||||
#: of someone else's exception. Those are identifiers and foreign strings, not copy.
|
||||
_DETAILS: dict[str, dict[str, str]] = {
|
||||
"en": {
|
||||
"already_awake": "already awake",
|
||||
"woken": "woken by Wake-on-LAN",
|
||||
"powered_off": "powered off",
|
||||
"still_needed": "left on: still needed by another run",
|
||||
"unmanaged": "left on: Joulenap does not manage this box's power",
|
||||
"left_on": "left powered on",
|
||||
"gc_disabled": "GC disabled for this route",
|
||||
"verify_disabled": "verify disabled for this route",
|
||||
"free_space": "{free}% free ({avail} GB)",
|
||||
"tasks_observed": "{count} task(s) observed",
|
||||
"no_tasks_observed": "no tasks observed",
|
||||
"interrupted": "Interrupted at startup",
|
||||
},
|
||||
"it": {
|
||||
"already_awake": "già acceso",
|
||||
"woken": "acceso con Wake-on-LAN",
|
||||
"powered_off": "spento",
|
||||
"still_needed": "lasciato acceso: serve ancora a un'altra esecuzione",
|
||||
"unmanaged": "lasciato acceso: Joulenap non gestisce l'alimentazione di questa macchina",
|
||||
"left_on": "lasciato acceso",
|
||||
"gc_disabled": "GC disattivata per questa route",
|
||||
"verify_disabled": "verifica disattivata per questa route",
|
||||
"free_space": "{free}% libero ({avail} GB)",
|
||||
"tasks_observed": "{count} task osservati",
|
||||
"no_tasks_observed": "nessun task osservato",
|
||||
"interrupted": "Interrotto al riavvio",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _run_error(language: str, run: Run) -> str | None:
|
||||
"""A run's failure message in ``language``, from the stored key when it has one.
|
||||
|
||||
@@ -318,13 +356,14 @@ def _pack(language: str) -> dict[str, dict[str, str]]:
|
||||
return _MESSAGES.get(language, _MESSAGES["en"])
|
||||
|
||||
|
||||
def render_error(
|
||||
def _render(
|
||||
catalogue: dict[str, dict[str, str]],
|
||||
language: str,
|
||||
key: str | None,
|
||||
params: Mapping[str, object] | None,
|
||||
raw: str | None = None,
|
||||
raw: str | None,
|
||||
) -> str | None:
|
||||
"""The failure message in ``language``, falling back to ``raw`` whenever it cannot be built.
|
||||
"""Look ``key`` up in ``catalogue`` and fill it in, degrading to ``raw`` at every step.
|
||||
|
||||
Every branch degrades to readable text rather than raising: a pre-1.0 row has no key, a
|
||||
key added in a later version may be missing from a pack, and a template whose parameters
|
||||
@@ -333,7 +372,7 @@ def render_error(
|
||||
"""
|
||||
if not key:
|
||||
return raw
|
||||
template = _ERRORS.get(language, _ERRORS["en"]).get(key) or _ERRORS["en"].get(key)
|
||||
template = catalogue.get(language, catalogue["en"]).get(key) or catalogue["en"].get(key)
|
||||
if template is None:
|
||||
return raw
|
||||
try:
|
||||
@@ -342,6 +381,31 @@ def render_error(
|
||||
return raw or template
|
||||
|
||||
|
||||
def render_error(
|
||||
language: str,
|
||||
key: str | None,
|
||||
params: Mapping[str, object] | None,
|
||||
raw: str | None = None,
|
||||
) -> str | None:
|
||||
"""The failure message in ``language``, falling back to ``raw`` whenever it cannot be built."""
|
||||
return _render(_ERRORS, language, key, params, raw)
|
||||
|
||||
|
||||
def render_detail(
|
||||
language: str,
|
||||
key: str | None,
|
||||
params: Mapping[str, object] | None,
|
||||
raw: str | None = None,
|
||||
) -> str | None:
|
||||
"""A step's ``detail`` in ``language``, falling back to the stored English ``raw``.
|
||||
|
||||
The same seam as :func:`render_error`, one level down. Only details Joulenap authored
|
||||
carry a key; a task UPID or a connector's own error text has none and comes back
|
||||
verbatim, which is what ``raw`` is for.
|
||||
"""
|
||||
return _render(_DETAILS, language, key, params, raw)
|
||||
|
||||
|
||||
def _title_for(pack: dict[str, dict[str, str]], kind: str, event: str) -> str:
|
||||
"""Title for a finished run, worded for the kind of cycle it was.
|
||||
|
||||
@@ -380,15 +444,21 @@ def _phase_breakdown(labels: dict[str, str], run: Run) -> str:
|
||||
Skipped steps (GC turned off) and steps still running contribute nothing, so the
|
||||
parentheses never advertise work that didn't happen. A ``StepName`` added later simply
|
||||
doesn't appear rather than raising.
|
||||
|
||||
Matched with ``_step_is`` rather than by equality, and summed per phase: a backup route
|
||||
records one step **per source PVE** (``backup:pve-alpha``), so an equality test matched
|
||||
nothing at all and every backup notification silently lost the one slice that mattered.
|
||||
"""
|
||||
parts = []
|
||||
totals: dict[str, float] = {}
|
||||
for step in run.steps: # the relationship is ordered by started_at
|
||||
key = _PHASE_LABEL.get(step.name)
|
||||
if key is None or step.status == StepStatus.SKIPPED or not step.finished_at:
|
||||
if step.status == StepStatus.SKIPPED or not step.finished_at:
|
||||
continue
|
||||
key = next((k for n, k in _PHASE_LABEL.items() if _step_is(step, n)), None)
|
||||
if key is None:
|
||||
continue
|
||||
seconds = (step.finished_at - step.started_at).total_seconds()
|
||||
parts.append(f"{labels[key]} {_format_duration(seconds)}")
|
||||
return " · ".join(parts)
|
||||
totals[key] = totals.get(key, 0.0) + seconds
|
||||
return " · ".join(f"{labels[k]} {_format_duration(s)}" for k, s in totals.items())
|
||||
|
||||
|
||||
def human_bytes(n: int) -> str:
|
||||
|
||||
@@ -116,7 +116,11 @@ class FakePbs:
|
||||
self.verify_args: dict | None = None
|
||||
# Sync route bookkeeping: what a route asked this box to set up and run.
|
||||
self.remotes: dict[str, dict] = {}
|
||||
# What ensure_remote was *asked* for, kept even after the run tears the remote down
|
||||
# again — the payload is the proof the right peer and credentials were used.
|
||||
self.remotes_created: dict[str, dict] = {}
|
||||
self.sync_jobs: dict[str, dict] = {}
|
||||
self.sync_jobs_created: dict[str, dict] = {}
|
||||
self.sync_runs: list[dict] = []
|
||||
# Ordered method names: PBS refuses to delete a remote a sync job still references,
|
||||
# so the sequence is load-bearing and a test pins it.
|
||||
@@ -152,14 +156,20 @@ class FakePbs:
|
||||
def ensure_remote(self, name: str, **kwargs) -> None:
|
||||
self.sync_calls.append("ensure_remote")
|
||||
self.remotes[name] = kwargs
|
||||
self.remotes_created[name] = kwargs
|
||||
|
||||
def delete_sync_job(self, job_id: str) -> None:
|
||||
self.sync_calls.append("delete_sync_job")
|
||||
self.sync_jobs.pop(job_id, None)
|
||||
|
||||
def delete_remote(self, name: str) -> None:
|
||||
self.sync_calls.append("delete_remote")
|
||||
self.remotes.pop(name, None)
|
||||
|
||||
def ensure_sync_job(self, job_id: str, **kwargs) -> None:
|
||||
self.sync_calls.append("ensure_sync_job")
|
||||
self.sync_jobs[job_id] = kwargs
|
||||
self.sync_jobs_created[job_id] = kwargs
|
||||
|
||||
def run_sync_job(self, job_id: str) -> str:
|
||||
self.sync_runs.append({"id": job_id})
|
||||
@@ -219,6 +229,8 @@ class FakeBox:
|
||||
|
||||
``reachable`` is either a constant or a list of answers consumed one probe at a time
|
||||
(the last one repeats), so a test can say "down, then up after the first wake".
|
||||
``unreachable`` names specific pbs ids that never answer, whatever ``reachable`` says —
|
||||
for a multi-box run where one wakes and the other does not.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -227,8 +239,10 @@ class FakeBox:
|
||||
idle: bool = True,
|
||||
idle_error: Exception | None = None,
|
||||
poweroff_error: Exception | None = None,
|
||||
unreachable: set[str] | None = None,
|
||||
):
|
||||
self._reachable = reachable
|
||||
self._unreachable = unreachable or set()
|
||||
self.idle = idle
|
||||
self.idle_error = idle_error
|
||||
self.poweroff_error = poweroff_error
|
||||
@@ -244,6 +258,8 @@ class FakeBox:
|
||||
def deps(self) -> LeaseDeps:
|
||||
def wait_reachable(pbs, timeout, _should_cancel=None) -> bool:
|
||||
self.waits.append(timeout)
|
||||
if pbs.id in self._unreachable:
|
||||
return False
|
||||
return self._answer()
|
||||
|
||||
def send_wol(pbs) -> None:
|
||||
|
||||
@@ -14,11 +14,14 @@ from sqlalchemy import select
|
||||
from app.config import Config, Route
|
||||
from app.db import session_scope
|
||||
from app.db.models import Run, RunKind, RunStatus, RunStep, RunTrigger, StepName, StepStatus
|
||||
from app.db.startup import _INTERRUPTED_STEP
|
||||
from app.jobs.lease import ReleaseOutcome
|
||||
from app.jobs.route_cycle import monitor_detail
|
||||
from app.main import create_app
|
||||
from app.notify import NotificationService
|
||||
from app.notify.apprise_urls import Channel, build_channels
|
||||
from app.notify.messages import (
|
||||
_DETAILS,
|
||||
_ERRORS,
|
||||
_KIND_LABEL,
|
||||
_MESSAGES,
|
||||
@@ -223,9 +226,13 @@ def _send(svc, config, run, **kwargs):
|
||||
return svc.send_run_result(RunContext(config=config, run=run, **kwargs))
|
||||
|
||||
|
||||
def _step(name: StepName, status: StepStatus, seconds: int) -> RunStep:
|
||||
"""A finished step that took ``seconds``, for the duration breakdown."""
|
||||
step = RunStep(name=name, status=status)
|
||||
def _step(name: StepName, status: StepStatus, seconds: int, label: str | None = None) -> RunStep:
|
||||
"""A finished step that took ``seconds``, for the duration breakdown.
|
||||
|
||||
``label`` produces the ``backup:pve-alpha`` / ``poweroff:pbs-02`` form a multi-device run
|
||||
records — which is what the route cycles actually write.
|
||||
"""
|
||||
step = RunStep(name=f"{name.value}:{label}" if label else name, status=status)
|
||||
step.started_at = datetime(2026, 6, 28, 4, 0, 0, tzinfo=UTC)
|
||||
step.finished_at = step.started_at + timedelta(seconds=seconds)
|
||||
return step
|
||||
@@ -305,7 +312,9 @@ def test_run_message_duration_breaks_down_the_work_phases():
|
||||
run = _run(RunStatus.SUCCESS)
|
||||
run.steps = [
|
||||
_step(StepName.WAIT, StepStatus.SUCCESS, 40),
|
||||
_step(StepName.BACKUP, StepStatus.SUCCESS, 70),
|
||||
# Labelled, because that is the only shape a backup route can produce: one step per
|
||||
# source PVE. An equality lookup matched nothing here and dropped the phase entirely.
|
||||
_step(StepName.BACKUP, StepStatus.SUCCESS, 70, label="pve-alpha"),
|
||||
RunStep(name=StepName.GC, status=StepStatus.SKIPPED),
|
||||
_step(StepName.POWEROFF, StepStatus.SUCCESS, 9),
|
||||
]
|
||||
@@ -313,6 +322,21 @@ def test_run_message_duration_breaks_down_the_work_phases():
|
||||
assert "Duration: 1m 23s (backup 1m 10s)" in body
|
||||
|
||||
|
||||
def test_run_message_sums_one_backup_phase_across_several_sources():
|
||||
"""A fan-in route records ``backup:pve-alpha`` and ``backup:pve-beta``; the line should
|
||||
read one ``backup`` slice, not two entries and not just the last one."""
|
||||
run = _run(RunStatus.SUCCESS)
|
||||
run.steps = [
|
||||
_step(StepName.WAIT, StepStatus.SUCCESS, 40),
|
||||
_step(StepName.BACKUP, StepStatus.SUCCESS, 70, label="pve-alpha"),
|
||||
_step(StepName.BACKUP, StepStatus.SUCCESS, 50, label="pve-beta"),
|
||||
_step(StepName.GC, StepStatus.SUCCESS, 66),
|
||||
_step(StepName.POWEROFF, StepStatus.SUCCESS, 9),
|
||||
]
|
||||
_title, body = _msg(Config(), run)
|
||||
assert "(backup 2m 0s · GC 1m 6s)" in body
|
||||
|
||||
|
||||
def test_run_message_trigger_and_next_run_are_localized():
|
||||
cfg = Config()
|
||||
cfg.app.language = "it"
|
||||
@@ -624,12 +648,29 @@ def test_every_language_pack_holds_the_same_keys():
|
||||
("_MESSAGES", _MESSAGES),
|
||||
("_KIND_LABEL", _KIND_LABEL),
|
||||
("_ERRORS", _ERRORS),
|
||||
("_DETAILS", _DETAILS),
|
||||
):
|
||||
english = sorted(_flatten(packs["en"]))
|
||||
for language, pack in packs.items():
|
||||
assert sorted(_flatten(pack)) == english, f"{name}: '{language}' differs from 'en'"
|
||||
|
||||
|
||||
def test_every_release_outcome_has_a_detail_string():
|
||||
"""``ReleaseOutcome.key`` is derived from the member name, so a new member silently
|
||||
renders as the raw English until the packs learn about it. This is what notices."""
|
||||
for outcome in ReleaseOutcome:
|
||||
assert outcome.key in _DETAILS["en"], outcome
|
||||
# The enum's value doubles as the stored English detail: the two must agree, or the
|
||||
# timeline reads one thing in English and another in Italian.
|
||||
assert _DETAILS["en"][outcome.key] == outcome.value
|
||||
|
||||
|
||||
def test_the_interrupted_step_detail_matches_the_catalogue():
|
||||
"""``db.startup`` writes the English text and the key by hand (importing the recorder
|
||||
there would be a cycle), so nothing but this keeps the two in step."""
|
||||
assert _INTERRUPTED_STEP == _DETAILS["en"]["interrupted"]
|
||||
|
||||
|
||||
def test_error_keys_are_rendered_in_the_configured_language():
|
||||
cfg = Config()
|
||||
cfg.app.language = "it"
|
||||
|
||||
@@ -274,8 +274,10 @@ def test_a_sync_route_leases_both_boxes(temp_config, temp_db):
|
||||
def test_an_unreachable_pbs_fails_the_run_without_running_the_job(temp_config, temp_db):
|
||||
service, box = make_service(FakeBox(reachable=False))
|
||||
ran = []
|
||||
seen: list[RunContext] = []
|
||||
service.deps.notify = seen.append
|
||||
|
||||
enqueue(service, "r1", lambda *_a: ran.append(1, trigger=RunTrigger.SCHEDULED))
|
||||
enqueue(service, "r1", lambda *_a: ran.append(1), trigger=RunTrigger.SCHEDULED)
|
||||
drain(service)
|
||||
|
||||
assert ran == []
|
||||
@@ -285,6 +287,38 @@ def test_an_unreachable_pbs_fails_the_run_without_running_the_job(temp_config, t
|
||||
assert run.status == RunStatus.FAILURE
|
||||
assert "not reachable" in (run.error or "")
|
||||
|
||||
# "The backup server never came up" is the failure this product exists to report, and
|
||||
# it used to send nothing at all: the handler re-raised, unwinding past _notify.
|
||||
assert len(seen) == 1
|
||||
assert seen[0].run.status == RunStatus.FAILURE
|
||||
assert seen[0].route is not None and seen[0].route.id == "r1"
|
||||
|
||||
|
||||
def test_a_sync_route_releases_the_box_that_did_wake_when_the_other_does_not(
|
||||
temp_config, temp_db
|
||||
):
|
||||
"""The second-order half of the same bug: one lease held, the other unreachable.
|
||||
|
||||
The re-raise skipped ``_release_all`` too, so the box that *did* answer was left awake
|
||||
with no POWEROFF step in the timeline and nothing queued to shut it down.
|
||||
"""
|
||||
# The sync route acquires its target (pbs2) first, then its source (pbs1).
|
||||
service, box = make_service(FakeBox(unreachable={"pbs1"}))
|
||||
seen: list[RunContext] = []
|
||||
service.deps.notify = seen.append
|
||||
|
||||
enqueue(service, "sync", lambda *_a: None, trigger=RunTrigger.SCHEDULED)
|
||||
drain(service)
|
||||
|
||||
with session_scope() as session:
|
||||
run = session.scalars(select(Run)).one()
|
||||
assert run.status == RunStatus.FAILURE
|
||||
steps = {s.name: s.status for s in run.steps}
|
||||
# pbs2 woke, so its release is recorded; a failed run leaves it on for inspection.
|
||||
assert "poweroff:pbs2" in steps
|
||||
assert box.poweroffs == []
|
||||
assert seen and seen[0].left_on == ["pbs2"]
|
||||
|
||||
|
||||
# --- what the run reports as left powered on ---------------------------------
|
||||
#
|
||||
|
||||
@@ -2,11 +2,14 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from app.api.schemas import StepInfo
|
||||
from app.db import session_scope
|
||||
from app.db.models import Run, RunKind, RunStatus, RunTrigger, StepName, StepStatus
|
||||
from app.jobs.recorder import RunRecorder
|
||||
from app.jobs.recorder import RunRecorder, set_detail
|
||||
|
||||
|
||||
def test_step_body_can_record_non_fatal_failure(temp_db):
|
||||
@@ -38,7 +41,7 @@ def test_a_skipped_step_does_not_finish_before_it_started(temp_db):
|
||||
# started_at used to come from the column default, which SQLAlchemy evaluates at flush —
|
||||
# i.e. after the finished_at passed here — so every skipped step had a negative duration.
|
||||
with RunRecorder(RunKind.CYCLE, RunTrigger.MANUAL) as recorder:
|
||||
recorder.skip_step(StepName.GC, "GC disabled for this route")
|
||||
recorder.skip_step(StepName.GC, "gc_disabled")
|
||||
run_id = recorder.run_id
|
||||
recorder.finish(RunStatus.SUCCESS)
|
||||
|
||||
@@ -101,3 +104,68 @@ def test_a_run_without_a_route_is_allowed(temp_db):
|
||||
with session_scope() as session:
|
||||
run = session.get(Run, run_id)
|
||||
assert run.route_id is None and run.route_name is None
|
||||
|
||||
|
||||
def test_a_step_detail_is_stored_in_english_and_rendered_in_the_users_language(temp_db):
|
||||
"""The whole seam end to end: what the recorder writes is what the API renders.
|
||||
|
||||
``detail`` stays English on the row — it is what the notifier reads and what a pre-1.0
|
||||
row has — while ``detail_key``/``detail_params`` let ``StepInfo`` rebuild the line in
|
||||
Italian on the way out.
|
||||
"""
|
||||
with RunRecorder(RunKind.CYCLE, RunTrigger.MANUAL) as recorder:
|
||||
with recorder.step(StepName.PRECHECK) as step:
|
||||
set_detail(step, "free_space", free="12.5", avail="640")
|
||||
recorder.skip_step(StepName.GC, "gc_disabled")
|
||||
run_id = recorder.run_id
|
||||
recorder.finish(RunStatus.SUCCESS)
|
||||
|
||||
with session_scope() as session:
|
||||
steps = {s.name: s for s in session.get(Run, run_id).steps}
|
||||
precheck, gc = steps[StepName.PRECHECK], steps[StepName.GC]
|
||||
|
||||
assert precheck.detail == "12.5% free (640 GB)"
|
||||
assert precheck.detail_key == "free_space"
|
||||
assert json.loads(precheck.detail_params) == {"free": "12.5", "avail": "640"}
|
||||
assert StepInfo.of(precheck, "it").detail == "12.5% libero (640 GB)"
|
||||
assert StepInfo.of(precheck).detail == precheck.detail # English is the default
|
||||
|
||||
assert gc.detail == "GC disabled for this route" and gc.detail_params is None
|
||||
assert StepInfo.of(gc, "it").detail == "GC disattivata per questa route"
|
||||
|
||||
|
||||
def test_a_step_detail_with_no_key_is_passed_through_untouched(temp_db):
|
||||
"""A task UPID (or someone else's error text) has no key and must not be mangled."""
|
||||
upid = "UPID:pve:0000ABCD:...:vzdump:101:root@pam:"
|
||||
with RunRecorder(RunKind.CYCLE, RunTrigger.MANUAL) as recorder:
|
||||
with recorder.step(StepName.BACKUP) as step:
|
||||
step.detail = upid
|
||||
run_id = recorder.run_id
|
||||
recorder.finish(RunStatus.SUCCESS)
|
||||
|
||||
with session_scope() as session:
|
||||
step = next(s for s in session.get(Run, run_id).steps if s.name == StepName.BACKUP)
|
||||
assert step.detail_key is None
|
||||
assert StepInfo.of(step, "it").detail == upid
|
||||
|
||||
|
||||
def test_a_step_that_set_a_detail_and_then_failed_shows_the_error(temp_db):
|
||||
"""The pre-flight guard's shape: report free space, then abort on it.
|
||||
|
||||
``render_detail`` prefers a resolvable key over the raw string, so leaving the key in
|
||||
place would render "12.5% free" on a FAILURE step and drop the reason entirely.
|
||||
"""
|
||||
with RunRecorder(RunKind.CYCLE, RunTrigger.MANUAL) as recorder:
|
||||
with pytest.raises(RuntimeError):
|
||||
with recorder.step(StepName.PRECHECK) as step:
|
||||
set_detail(step, "free_space", free="12.5", avail="640")
|
||||
raise RuntimeError("datastore too full")
|
||||
run_id = recorder.run_id
|
||||
recorder.finish(RunStatus.FAILURE)
|
||||
|
||||
with session_scope() as session:
|
||||
step = next(s for s in session.get(Run, run_id).steps if s.name == StepName.PRECHECK)
|
||||
assert step.status == StepStatus.FAILURE
|
||||
assert step.detail == "datastore too full"
|
||||
assert step.detail_key is None and step.detail_params is None
|
||||
assert StepInfo.of(step, "it").detail == "datastore too full"
|
||||
|
||||
@@ -109,8 +109,8 @@ def test_pull_sync_runs_the_job_on_the_target(temp_db):
|
||||
|
||||
run_id = _run(config, deps)
|
||||
|
||||
# The target pulls: remote + job live on it, pointing at the source.
|
||||
assert pbs1.remotes == {
|
||||
# The target pulls: remote + job are created on it, pointing at the source.
|
||||
assert pbs1.remotes_created == {
|
||||
"joulenap-r1": {
|
||||
"host": "192.0.2.21",
|
||||
"port": 8007,
|
||||
@@ -119,7 +119,7 @@ def test_pull_sync_runs_the_job_on_the_target(temp_db):
|
||||
"fingerprint": "cc:dd",
|
||||
}
|
||||
}
|
||||
assert pbs1.sync_jobs == {
|
||||
assert pbs1.sync_jobs_created == {
|
||||
"joulenap-r1": {
|
||||
"remote": "joulenap-r1",
|
||||
"remote_store": "offsite", # the peer's datastore
|
||||
@@ -133,7 +133,16 @@ def test_pull_sync_runs_the_job_on_the_target(temp_db):
|
||||
# The job is dropped before the remote is rebuilt. PBS refuses to delete a remote that a
|
||||
# sync job still references, so the reverse order fails on every run after the first —
|
||||
# a sync route that works once and then never again.
|
||||
assert pbs1.sync_calls == ["delete_sync_job", "ensure_remote", "ensure_sync_job"]
|
||||
# ...and both are dropped again once the sync is over: the remote holds the *peer's* API
|
||||
# token, so leaving it behind parks a working credential for one PBS on another.
|
||||
assert pbs1.sync_calls == [
|
||||
"delete_sync_job",
|
||||
"ensure_remote",
|
||||
"ensure_sync_job",
|
||||
"delete_sync_job",
|
||||
"delete_remote",
|
||||
]
|
||||
assert pbs1.remotes == {} and pbs1.sync_jobs == {}
|
||||
# The source is never touched by the job — it is only kept awake.
|
||||
assert (pbs2.remotes, pbs2.sync_jobs, pbs2.sync_runs) == ({}, {}, [])
|
||||
status, steps = _load(run_id)
|
||||
@@ -149,15 +158,16 @@ def test_push_sync_runs_the_job_on_the_source(temp_db):
|
||||
run_id = _run(config, deps)
|
||||
|
||||
# Mirror image: the source sends, so it executes and its remote is the target.
|
||||
assert pbs2.remotes["joulenap-r1"]["host"] == "192.0.2.20"
|
||||
assert pbs2.remotes["joulenap-r1"]["auth_id"] == "root@pam!jn1"
|
||||
assert pbs2.sync_jobs["joulenap-r1"] == {
|
||||
assert pbs2.remotes_created["joulenap-r1"]["host"] == "192.0.2.20"
|
||||
assert pbs2.remotes_created["joulenap-r1"]["auth_id"] == "root@pam!jn1"
|
||||
assert pbs2.sync_jobs_created["joulenap-r1"] == {
|
||||
"remote": "joulenap-r1",
|
||||
"remote_store": "backup",
|
||||
"store": "offsite",
|
||||
"direction": "push",
|
||||
}
|
||||
assert pbs2.sync_runs == [{"id": "joulenap-r1"}]
|
||||
assert pbs2.remotes == {} and pbs2.sync_jobs == {} # torn down after the run
|
||||
assert (pbs1.remotes, pbs1.sync_jobs, pbs1.sync_runs) == ({}, {}, [])
|
||||
assert _load(run_id)[0] == RunStatus.SUCCESS
|
||||
|
||||
|
||||
Reference in New Issue
Block a user