Compare commits
3 Commits
us/US04-03
...
us/US04-06
| Author | SHA1 | Date | |
|---|---|---|---|
| b9b5ab7f51 | |||
| 8d96acd067 | |||
| 7103cb84bf |
34
README.md
34
README.md
@@ -102,3 +102,37 @@ work_item/scripts/python -m pytest tests/e2e -m phase_c -q
|
||||
The deterministic naming provider is enabled only by test configuration
|
||||
(`PHOTO_PIPELINE_FAKE_NAMING_LOG`); without it the application falls back to the
|
||||
offline naming-policy name. Phase A and B suites remain green in the full run above.
|
||||
|
||||
### Phase D acceptance gate
|
||||
|
||||
Phase D (Epic E04: guarded renaming) is the first phase that changes the library on
|
||||
disk, so its gate is the strictest. One command runs the rename API journeys, the
|
||||
filesystem fault injection, and the browser suite:
|
||||
|
||||
```bash
|
||||
work_item/scripts/python -m pytest tests/e2e -m phase_d -q
|
||||
```
|
||||
|
||||
- `tests/e2e/test_phase_d_pipeline.py` drives a real server over HTTP: plan and
|
||||
export, confirmation with the plan version and checksum (a stale token is refused
|
||||
without touching disk), a valid apply, the case-only rename procedure, a collision
|
||||
whose occupant survives, a source that changed after planning, and durability
|
||||
across a full restart.
|
||||
- **Fault injection is real.** `PHOTO_PIPELINE_FAULT_AFTER=<journal state>` kills the
|
||||
server process the instant that state is persisted. The suite crashes it at every
|
||||
journal transition in turn (`moving`, `moved`, `database_updated`, `verified`),
|
||||
starts a fresh process against the same database and library, and requires recovery
|
||||
to converge from journal and disk evidence alone — with the asset set, the stable
|
||||
IDs, and every content hash unchanged. Ambiguous evidence is never guessed: it stays
|
||||
classified `manual` and keeps blocking. An unresolved rename is the cancellation
|
||||
boundary — there is no cancel once a run starts, and unrelated mutations (album
|
||||
proposal generation and approval) are refused with 409 `rename_recovery_required`
|
||||
until it is resolved, while reads stay available.
|
||||
- `tests/e2e/test_renames_ui.py` covers the browser journeys: preview of every
|
||||
affected path, confirmation carrying the server-issued token, apply with progress
|
||||
and terminal verification, stale confirmation, collision, interruption, recovery,
|
||||
rollback, keyboard confirmation, and the view still matching the journal after a
|
||||
server restart.
|
||||
|
||||
The fault barrier is test-only configuration; without `PHOTO_PIPELINE_FAULT_AFTER`
|
||||
the apply path has no crash points. Phases A–C remain green in the full run above.
|
||||
|
||||
@@ -165,6 +165,11 @@ a.link:hover { text-decoration: underline; }
|
||||
.badge.approved { color: var(--ok); border-color: var(--ok); }
|
||||
.badge.error { color: var(--danger); border-color: var(--danger); }
|
||||
|
||||
/* Renames: the preview table must show whole paths, so let them wrap rather than
|
||||
truncate — a hidden path segment is exactly what makes a move a surprise. */
|
||||
.grid th, .grid td { text-align: left; padding: 6px 8px; border-bottom: 1px solid var(--border); vertical-align: top; }
|
||||
.path { font-family: ui-monospace, monospace; font-size: 0.85em; word-break: break-all; }
|
||||
|
||||
/* Narrow screens: stack the panes so blockers and actions stay reachable. */
|
||||
@media (max-width: 720px) {
|
||||
.two-pane { grid-template-columns: 1fr; }
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
<a href="#/library" data-nav="library">Library</a>
|
||||
<a href="#/analyze" data-nav="analyze">Analyze</a>
|
||||
<a href="#/albums" data-nav="albums">Albums</a>
|
||||
<a href="#/renames" data-nav="renames">Renames</a>
|
||||
<a href="#/stats" data-nav="stats">Stats</a>
|
||||
</nav>
|
||||
</header>
|
||||
|
||||
@@ -99,4 +99,19 @@ export const api = {
|
||||
body: JSON.stringify(payload),
|
||||
...opts,
|
||||
}),
|
||||
|
||||
// ── Renames: plan, confirm, apply, recover ──────────────────────────────
|
||||
listPlans: (opts = {}) => request("/rename-plans", opts),
|
||||
buildPlan: (opts = {}) => request("/rename-plans", { method: "POST", ...opts }),
|
||||
getPlan: (id, opts = {}) => request(`/rename-plans/${encodeURIComponent(id)}`, opts),
|
||||
applyPlan: (id, payload, opts = {}) =>
|
||||
request(`/rename-plans/${encodeURIComponent(id)}/apply`, {
|
||||
method: "POST",
|
||||
body: JSON.stringify(payload),
|
||||
...opts,
|
||||
}),
|
||||
rollbackPlan: (id, opts = {}) =>
|
||||
request(`/rename-plans/${encodeURIComponent(id)}/rollback`, { method: "POST", ...opts }),
|
||||
renameRecovery: (opts = {}) => request("/rename-recovery", opts),
|
||||
resolveRecovery: (opts = {}) => request("/rename-recovery/resolve", { method: "POST", ...opts }),
|
||||
};
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { api } from "./api.js";
|
||||
import { navigate, onRouteChange, parseHash } from "./router.js";
|
||||
import { renderRenames, setRenamesRender } from "./renames.js";
|
||||
import {
|
||||
renderAlbums,
|
||||
renderAnalyze,
|
||||
@@ -362,11 +363,13 @@ function render() {
|
||||
else if (path === "/library") renderLibrary(root, params);
|
||||
else if (path === "/analyze") renderAnalyze(root, params);
|
||||
else if (path === "/albums") renderAlbums(root, params);
|
||||
else if (path === "/renames") renderRenames(root, params);
|
||||
else if (path === "/stats") renderStats(root, params);
|
||||
else show(errorBanner("Unknown view"));
|
||||
}
|
||||
|
||||
// Let views re-render the current route after a mutation.
|
||||
setRender(render);
|
||||
setRenamesRender(render);
|
||||
onRouteChange(render);
|
||||
render();
|
||||
|
||||
357
frontend/js/renames.js
Normal file
357
frontend/js/renames.js
Normal file
@@ -0,0 +1,357 @@
|
||||
// Renames view (US04-05): preview a rename plan, confirm it with the server-issued
|
||||
// token, watch it run, and resolve anything an interruption left behind.
|
||||
//
|
||||
// Nothing here decides what is safe. Blocking issues, plan applicability, and
|
||||
// recovery classifications all come from the server; the view only shows them and
|
||||
// refuses to offer an action the server would reject.
|
||||
import { api } from "./api.js";
|
||||
import { el, errorBanner, setActiveNav } from "./dom.js";
|
||||
|
||||
// What the last confirm did, for this tab only. Deliberately not persisted: after a
|
||||
// reload the page must show what the journal says, not what this page remembers.
|
||||
let outcome = null;
|
||||
|
||||
let render = () => {};
|
||||
export function setRenamesRender(fn) {
|
||||
render = fn;
|
||||
}
|
||||
|
||||
const BLOCKED_TITLE = "an unresolved rename must be resolved first";
|
||||
|
||||
export async function renderRenames(root, params = {}) {
|
||||
setActiveNav("renames");
|
||||
let plans, recovery;
|
||||
try {
|
||||
[plans, recovery] = await Promise.all([api.listPlans(), api.renameRecovery()]);
|
||||
} catch (error) {
|
||||
root.replaceChildren(errorBanner(`Failed to load renames: ${error.message}`));
|
||||
return;
|
||||
}
|
||||
|
||||
const selectedId = params.plan || (plans.items[0] && plans.items[0].id) || null;
|
||||
let plan = null;
|
||||
if (selectedId) {
|
||||
try {
|
||||
plan = await api.getPlan(selectedId);
|
||||
} catch (error) {
|
||||
root.replaceChildren(errorBanner(`Failed to load plan: ${error.message}`));
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const blocked = recovery.blocks_mutation;
|
||||
root.replaceChildren(
|
||||
el("h1", {}, "Renames"),
|
||||
recoveryPanel(recovery),
|
||||
el(
|
||||
"div",
|
||||
{ class: "toolbar" },
|
||||
el(
|
||||
"button",
|
||||
{
|
||||
class: "primary",
|
||||
"data-testid": "build-plan",
|
||||
disabled: blocked ? "disabled" : false,
|
||||
title: blocked ? BLOCKED_TITLE : false,
|
||||
onclick: () => run(() => api.buildPlan(), "Planning…"),
|
||||
},
|
||||
"Build plan from approved names"
|
||||
)
|
||||
),
|
||||
outcomeBanner(),
|
||||
plan
|
||||
? planSection(plan, blocked)
|
||||
: el(
|
||||
"p",
|
||||
{ class: "muted", "data-testid": "no-plan" },
|
||||
"No rename plan yet — approve album names, then build a plan to preview the moves."
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
// ── preview ──────────────────────────────────────────────────────────────────
|
||||
function planSection(plan, blocked) {
|
||||
const blockingIssues = plan.operations.flatMap((operation) =>
|
||||
operation.issues.filter((issue) => issue.blocking)
|
||||
);
|
||||
|
||||
return el(
|
||||
"div",
|
||||
{ "data-testid": "plan" },
|
||||
el(
|
||||
"div",
|
||||
{ class: "decision-bar" },
|
||||
el("span", { class: `badge ${plan.state}`, "data-testid": "plan-state" }, plan.state),
|
||||
el(
|
||||
"span",
|
||||
{ class: "badge", "data-testid": "operation-count" },
|
||||
`${plan.operation_count} folder${plan.operation_count === 1 ? "" : "s"}`
|
||||
),
|
||||
el(
|
||||
"span",
|
||||
{ class: "badge", "data-testid": "affected-assets" },
|
||||
`${plan.operations.reduce((sum, o) => sum + o.asset_count, 0)} photos`
|
||||
),
|
||||
blockingIssues.length
|
||||
? el(
|
||||
"span",
|
||||
{ class: "badge attention", "data-testid": "blocking-count" },
|
||||
`${blockingIssues.length} blocking issue${blockingIssues.length === 1 ? "" : "s"}`
|
||||
)
|
||||
: null
|
||||
),
|
||||
plan.blockers.length
|
||||
? el(
|
||||
"div",
|
||||
{ class: "alert", role: "alert", "data-testid": "plan-blockers" },
|
||||
`This plan cannot run: ${plan.blockers.join("; ")}`
|
||||
)
|
||||
: null,
|
||||
operationsTable(plan),
|
||||
confirmBlock(plan, blocked)
|
||||
);
|
||||
}
|
||||
|
||||
function operationType(operation) {
|
||||
if (operation.case_only) return "case-only rename";
|
||||
if (!operation.same_filesystem) return "cross-filesystem move";
|
||||
return "rename";
|
||||
}
|
||||
|
||||
function operationsTable(plan) {
|
||||
if (!plan.operations.length) {
|
||||
return el("p", { class: "muted", "data-testid": "no-operations" }, "This plan moves nothing.");
|
||||
}
|
||||
return el(
|
||||
"table",
|
||||
{ class: "grid", "data-testid": "operations" },
|
||||
el(
|
||||
"thead",
|
||||
{},
|
||||
el(
|
||||
"tr",
|
||||
{},
|
||||
...["#", "Album", "From", "To", "Type", "Photos", "Issues", "State"].map((label) =>
|
||||
el("th", { scope: "col" }, label)
|
||||
)
|
||||
)
|
||||
),
|
||||
el(
|
||||
"tbody",
|
||||
{},
|
||||
...plan.operations.map((operation) =>
|
||||
el(
|
||||
"tr",
|
||||
{ "data-testid": "operation-row", "data-album": operation.album },
|
||||
el("td", {}, String(operation.sequence)),
|
||||
el("td", { "data-testid": "op-album" }, operation.album),
|
||||
// Full paths, not basenames: the whole point of the preview is that no
|
||||
// move is a surprise.
|
||||
el("td", { class: "path", "data-testid": "op-source" }, operation.source_path),
|
||||
el("td", { class: "path", "data-testid": "op-destination" }, operation.destination_path),
|
||||
el("td", { "data-testid": "op-type" }, operationType(operation)),
|
||||
el("td", { "data-testid": "op-count" }, String(operation.asset_count)),
|
||||
el(
|
||||
"td",
|
||||
{ "data-testid": "op-issues" },
|
||||
operation.issues.length
|
||||
? operation.issues.map((issue) =>
|
||||
el(
|
||||
"span",
|
||||
{
|
||||
class: `badge ${issue.blocking ? "attention" : ""}`,
|
||||
"data-testid": "op-issue",
|
||||
"data-code": issue.code,
|
||||
title: issue.message,
|
||||
},
|
||||
`${issue.blocking ? "blocks" : "note"}: ${issue.code}`
|
||||
)
|
||||
)
|
||||
: el("span", { class: "muted" }, "—")
|
||||
),
|
||||
el(
|
||||
"td",
|
||||
{},
|
||||
el("span", { "data-testid": "op-state" }, operation.journal_state),
|
||||
operation.verified_at
|
||||
? el("span", { class: "badge complete", "data-testid": "op-verified" }, "verified")
|
||||
: null,
|
||||
operation.error_code
|
||||
? el(
|
||||
"div",
|
||||
{ class: "blocker", "data-testid": "op-error" },
|
||||
`${operation.error_code}: ${operation.error_message || ""}`
|
||||
)
|
||||
: null
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
// ── confirmation ─────────────────────────────────────────────────────────────
|
||||
// Journal states whose content can still be moved back; `complete` is not one of
|
||||
// them (photo_pipeline/services/rename_journal.py: complete is terminal).
|
||||
const REVERSIBLE = new Set(["moved", "database_updated", "verified", "rollback_required"]);
|
||||
|
||||
function rollbackable(plan) {
|
||||
return plan.operations.some((operation) => REVERSIBLE.has(operation.journal_state));
|
||||
}
|
||||
|
||||
function confirmBlock(plan, blocked) {
|
||||
const applyable = plan.applicable && !blocked;
|
||||
return el(
|
||||
"div",
|
||||
{ class: "card", "data-testid": "confirm" },
|
||||
el("h2", {}, "Confirm"),
|
||||
// The token is shown, not just sent: a confirmation the user cannot see is a
|
||||
// confirmation they cannot check against what the preview claimed.
|
||||
el(
|
||||
"p",
|
||||
{ class: "muted", "data-testid": "confirm-token" },
|
||||
`Plan ${plan.id} · version ${plan.version} · checksum ${plan.checksum.slice(0, 12)}`
|
||||
),
|
||||
el(
|
||||
"p",
|
||||
{ "data-testid": "cancel-note" },
|
||||
"Once confirmed, the run cannot be cancelled part-way: each folder is moved, " +
|
||||
"verified, and recorded before the next one starts. An interruption is " +
|
||||
"recoverable from the journal — it is never left half-applied."
|
||||
),
|
||||
el(
|
||||
"div",
|
||||
{ class: "toolbar" },
|
||||
el(
|
||||
"button",
|
||||
{
|
||||
class: "primary",
|
||||
"data-testid": "apply-plan",
|
||||
disabled: applyable ? false : "disabled",
|
||||
title: blocked
|
||||
? BLOCKED_TITLE
|
||||
: plan.applicable
|
||||
? false
|
||||
: `plan is ${plan.state} — it cannot be applied`,
|
||||
onclick: () =>
|
||||
run(
|
||||
() =>
|
||||
api.applyPlan(plan.id, {
|
||||
expected_version: plan.version,
|
||||
expected_checksum: plan.checksum,
|
||||
}),
|
||||
`Applying ${plan.operation_count} rename(s)…`
|
||||
),
|
||||
},
|
||||
`Apply ${plan.operation_count} rename(s)`
|
||||
),
|
||||
// Rollback is recovery, not undo: a completed rename is terminal and stays
|
||||
// that way. The button appears only while operations are still reversible.
|
||||
rollbackable(plan)
|
||||
? el(
|
||||
"button",
|
||||
{
|
||||
"data-testid": "rollback-plan",
|
||||
onclick: () => run(() => api.rollbackPlan(plan.id), "Rolling back…"),
|
||||
},
|
||||
"Roll back the unfinished moves"
|
||||
)
|
||||
: null
|
||||
),
|
||||
el("p", { role: "status", "aria-live": "polite", "data-testid": "apply-progress" }, "")
|
||||
);
|
||||
}
|
||||
|
||||
// Runs a mutating call, showing progress while it is in flight and leaving the
|
||||
// result — including a stale-confirmation conflict — visible afterwards.
|
||||
async function run(action, progressText) {
|
||||
const progress = document.querySelector('[data-testid="apply-progress"]');
|
||||
if (progress) progress.textContent = progressText;
|
||||
try {
|
||||
outcome = { kind: "ok", result: await action() };
|
||||
} catch (error) {
|
||||
outcome = error.status === 409 ? { kind: "conflict" } : { kind: "error", error };
|
||||
}
|
||||
render();
|
||||
}
|
||||
|
||||
function outcomeBanner() {
|
||||
if (!outcome) return null;
|
||||
if (outcome.kind === "conflict") {
|
||||
return el(
|
||||
"div",
|
||||
{ class: "alert", role: "alert", "data-testid": "conflict" },
|
||||
"This plan changed on the server since it was loaded. Nothing was applied; " +
|
||||
"the preview below is the current state — review it and confirm again."
|
||||
);
|
||||
}
|
||||
if (outcome.kind === "error") {
|
||||
return el(
|
||||
"div",
|
||||
{ class: "alert", role: "alert", "data-testid": "apply-error" },
|
||||
`Failed: ${outcome.error.message}`
|
||||
);
|
||||
}
|
||||
const result = outcome.result;
|
||||
if (result.applied === undefined) return null; // a build or recovery call
|
||||
return el(
|
||||
"div",
|
||||
{ class: "alert", role: "status", "data-testid": "apply-result" },
|
||||
`Applied ${result.applied}, failed ${result.failed}, skipped ${result.skipped}. ` +
|
||||
`The plan is now ${result.state}.`
|
||||
);
|
||||
}
|
||||
|
||||
// ── recovery ─────────────────────────────────────────────────────────────────
|
||||
function recoveryPanel(recovery) {
|
||||
if (!recovery.items.length) return null;
|
||||
const recoverable = recovery.items.filter((item) => item.classification !== "manual");
|
||||
return el(
|
||||
"div",
|
||||
{ class: "alert", role: "alert", "data-testid": "recovery-banner" },
|
||||
el(
|
||||
"strong",
|
||||
{},
|
||||
recovery.blocks_mutation
|
||||
? "A rename was interrupted — other changes are paused"
|
||||
: "A rename is unresolved"
|
||||
),
|
||||
el(
|
||||
"p",
|
||||
{ "data-testid": "recovery-explanation" },
|
||||
recovery.blocks_mutation
|
||||
? "Some folders on disk may disagree with the database. Planning, applying, and " +
|
||||
"album-name changes stay blocked until this is resolved; browsing is unaffected."
|
||||
: "These operations never reached a terminal state."
|
||||
),
|
||||
el(
|
||||
"ul",
|
||||
{},
|
||||
...recovery.items.map((item) =>
|
||||
el(
|
||||
"li",
|
||||
{ "data-testid": "recovery-item", "data-classification": item.classification },
|
||||
`${item.album}: ${item.journal_state} — ${item.classification} — ${item.reason} ` +
|
||||
`(${item.source_path} → ${item.destination_path})`
|
||||
)
|
||||
)
|
||||
),
|
||||
// Only evidence-safe work is offered. Ambiguous operations get no button at
|
||||
// all — an action the server would refuse must not look available.
|
||||
recoverable.length
|
||||
? el(
|
||||
"button",
|
||||
{
|
||||
"data-testid": "resolve-recovery",
|
||||
onclick: () => run(() => api.resolveRecovery(), "Resolving…"),
|
||||
},
|
||||
`Resolve ${recoverable.length} recoverable rename(s) from evidence`
|
||||
)
|
||||
: el(
|
||||
"p",
|
||||
{ "data-testid": "manual-only" },
|
||||
"No automatic action is safe here: source and destination both exist, or neither " +
|
||||
"does. Inspect the paths above and resolve them by hand."
|
||||
)
|
||||
);
|
||||
}
|
||||
@@ -327,11 +327,15 @@ export async function renderStats(root) {
|
||||
// Approving never renames anything — it records the approved name only.
|
||||
export async function renderAlbums(root, params = {}) {
|
||||
setActiveNav("albums");
|
||||
let evidence, proposals;
|
||||
let evidence, proposals, recovery;
|
||||
try {
|
||||
[evidence, proposals] = await Promise.all([
|
||||
[evidence, proposals, recovery] = await Promise.all([
|
||||
api.albumEvidence({ limit: 500 }),
|
||||
api.albumProposals(),
|
||||
// An unresolved rename means folder paths on disk may disagree with the
|
||||
// database, so the server refuses name changes until it is resolved. Ask up
|
||||
// front rather than offering a button that is guaranteed to fail.
|
||||
api.renameRecovery(),
|
||||
]);
|
||||
} catch (error) {
|
||||
root.replaceChildren(errorBanner(`Failed to load albums: ${error.message}`));
|
||||
@@ -370,17 +374,27 @@ export async function renderAlbums(root, params = {}) {
|
||||
})
|
||||
);
|
||||
|
||||
const renameBlocked = recovery.blocks_mutation;
|
||||
const detail = selected
|
||||
? albumDetail(
|
||||
root,
|
||||
folders.find((folder) => folder.album === selected),
|
||||
byAlbum.get(selected),
|
||||
params
|
||||
params,
|
||||
renameBlocked
|
||||
)
|
||||
: el("p", { class: "muted" }, "No albums with evidence yet.");
|
||||
|
||||
root.replaceChildren(
|
||||
el("h1", {}, "Albums"),
|
||||
renameBlocked
|
||||
? el(
|
||||
"div",
|
||||
{ class: "alert", role: "alert", "data-testid": "rename-blocked" },
|
||||
"An interrupted rename is unresolved. Name changes are paused until it is " +
|
||||
"resolved on the Renames page; browsing stays available."
|
||||
)
|
||||
: null,
|
||||
el(
|
||||
"div",
|
||||
{ class: "toolbar" },
|
||||
@@ -389,6 +403,8 @@ export async function renderAlbums(root, params = {}) {
|
||||
{
|
||||
class: "primary",
|
||||
"data-testid": "generate-proposals",
|
||||
disabled: renameBlocked ? "disabled" : false,
|
||||
title: renameBlocked ? "an unresolved rename must be resolved first" : false,
|
||||
onclick: async () => {
|
||||
try {
|
||||
await api.generateProposals({});
|
||||
@@ -405,7 +421,7 @@ export async function renderAlbums(root, params = {}) {
|
||||
);
|
||||
}
|
||||
|
||||
function albumDetail(root, folder, proposal, params) {
|
||||
function albumDetail(root, folder, proposal, params, renameBlocked = false) {
|
||||
if (!folder) return el("p", { class: "muted" }, "Album not found.");
|
||||
|
||||
const evidenceBlock = el(
|
||||
@@ -473,8 +489,13 @@ function albumDetail(root, folder, proposal, params) {
|
||||
{
|
||||
class: "primary",
|
||||
"data-testid": "approve",
|
||||
disabled: proposal.stale || proposal.status === "error" ? "disabled" : false,
|
||||
title: proposal.stale ? "evidence changed — regenerate first" : false,
|
||||
disabled:
|
||||
renameBlocked || proposal.stale || proposal.status === "error" ? "disabled" : false,
|
||||
title: renameBlocked
|
||||
? "an unresolved rename must be resolved first"
|
||||
: proposal.stale
|
||||
? "evidence changed — regenerate first"
|
||||
: false,
|
||||
onclick: () => mutate(root, () => api.approveProposal(proposal.album, { expected_version: proposal.version }), params),
|
||||
},
|
||||
proposal.status === "approved" ? "Approved" : "Approve name"
|
||||
|
||||
@@ -17,6 +17,7 @@ from photo_pipeline.schemas import (
|
||||
)
|
||||
from photo_pipeline.services.albums import AlbumService
|
||||
from photo_pipeline.services.proposals import ConflictError, ProposalError, ProposalService
|
||||
from photo_pipeline.services.rename_journal import RenameJournal
|
||||
|
||||
router = APIRouter(tags=["albums"])
|
||||
|
||||
@@ -37,6 +38,22 @@ def _error(status: int, code: str, message: str) -> JSONResponse:
|
||||
return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
|
||||
|
||||
|
||||
def _unresolved_rename(request: Request) -> JSONResponse | None:
|
||||
"""Refuse work that feeds renaming while the library may be half-renamed.
|
||||
|
||||
An unresolved rename means some folder paths on disk disagree with the database,
|
||||
so evidence gathered now — and any approval based on it — could describe a
|
||||
location that no longer exists. Reading stays available; only mutation stops.
|
||||
"""
|
||||
if RenameJournal(request.app.state.session_factory).blocks_mutation():
|
||||
return _error(
|
||||
409,
|
||||
"rename_recovery_required",
|
||||
"an unresolved rename is in progress; resolve it before changing album names",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
@router.get("/albums/evidence")
|
||||
def evidence(
|
||||
request: Request,
|
||||
@@ -47,8 +64,8 @@ def evidence(
|
||||
|
||||
|
||||
@router.post("/albums/proposals")
|
||||
def generate(body: GenerateProposalsRequest, request: Request) -> dict:
|
||||
return _proposals(request).generate(body.albums)
|
||||
def generate(body: GenerateProposalsRequest, request: Request):
|
||||
return _unresolved_rename(request) or _proposals(request).generate(body.albums)
|
||||
|
||||
|
||||
@router.get("/albums/proposals")
|
||||
@@ -76,6 +93,9 @@ def edit_proposal(album: str, body: EditProposalRequest, request: Request):
|
||||
|
||||
@router.post("/albums/proposals/{album:path}/approve")
|
||||
def approve_proposal(album: str, body: ApproveProposalRequest, request: Request):
|
||||
blocked = _unresolved_rename(request)
|
||||
if blocked is not None:
|
||||
return blocked
|
||||
try:
|
||||
return _proposals(request).approve(album, expected_version=body.expected_version)
|
||||
except ConflictError as error:
|
||||
|
||||
@@ -9,6 +9,12 @@ from __future__ import annotations
|
||||
from fastapi import APIRouter, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from photo_pipeline.schemas import ApplyPlanRequest
|
||||
from photo_pipeline.services.rename_apply import (
|
||||
ApplyConflict,
|
||||
ApplyError,
|
||||
RenameApplyService,
|
||||
)
|
||||
from photo_pipeline.services.renames import RenameError, RenameService
|
||||
|
||||
router = APIRouter(tags=["renames"])
|
||||
@@ -21,6 +27,13 @@ def _service(request: Request) -> RenameService:
|
||||
)
|
||||
|
||||
|
||||
def _apply_service(request: Request) -> RenameApplyService:
|
||||
return RenameApplyService(
|
||||
request.app.state.session_factory,
|
||||
library_roots=tuple(request.app.state.config.library_roots),
|
||||
)
|
||||
|
||||
|
||||
def _error(status: int, code: str, message: str) -> JSONResponse:
|
||||
return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
|
||||
|
||||
@@ -52,3 +65,42 @@ def export_plan(plan_id: str, request: Request):
|
||||
return _service(request).export(plan_id)
|
||||
except RenameError as error:
|
||||
return _error(404, "not_found", str(error))
|
||||
|
||||
|
||||
@router.post("/rename-plans/{plan_id}/apply")
|
||||
def apply_plan(plan_id: str, body: ApplyPlanRequest, request: Request):
|
||||
"""Apply a confirmed plan. The confirmation must carry the current plan version
|
||||
(and may carry the checksum) so stale browser state cannot run a changed plan."""
|
||||
try:
|
||||
return _apply_service(request).apply(
|
||||
plan_id,
|
||||
expected_version=body.expected_version,
|
||||
expected_checksum=body.expected_checksum,
|
||||
)
|
||||
except ApplyConflict as error:
|
||||
return _error(409, "version_conflict", str(error))
|
||||
except ApplyError as error:
|
||||
return _error(422, "cannot_apply", str(error))
|
||||
|
||||
|
||||
@router.get("/rename-recovery")
|
||||
def recovery_status(request: Request) -> dict:
|
||||
"""Every unresolved rename with its evidence-derived classification."""
|
||||
journal = _apply_service(request).journal
|
||||
return {
|
||||
"blocks_mutation": journal.blocks_mutation(),
|
||||
"items": journal.classify_all(),
|
||||
}
|
||||
|
||||
|
||||
@router.post("/rename-recovery/resolve")
|
||||
def resolve_recovery(request: Request) -> dict:
|
||||
return _apply_service(request).recover()
|
||||
|
||||
|
||||
@router.post("/rename-plans/{plan_id}/rollback")
|
||||
def rollback_plan(plan_id: str, request: Request):
|
||||
try:
|
||||
return _apply_service(request).rollback_plan(plan_id)
|
||||
except ApplyError as error:
|
||||
return _error(422, "cannot_rollback", str(error))
|
||||
|
||||
@@ -6,10 +6,12 @@ from photo_pipeline.schemas.albums import (
|
||||
GenerateProposalsRequest,
|
||||
)
|
||||
from photo_pipeline.schemas.duplicates import DecisionRequest
|
||||
from photo_pipeline.schemas.renames import ApplyPlanRequest
|
||||
from photo_pipeline.schemas.jobs import JobStartRequest
|
||||
from photo_pipeline.schemas.safety import SafetyDecisionRequest
|
||||
|
||||
__all__ = [
|
||||
"ApplyPlanRequest",
|
||||
"ApproveProposalRequest",
|
||||
"DecisionRequest",
|
||||
"EditProposalRequest",
|
||||
|
||||
11
photo_pipeline/schemas/renames.py
Normal file
11
photo_pipeline/schemas/renames.py
Normal file
@@ -0,0 +1,11 @@
|
||||
"""Rename apply/confirmation request contracts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class ApplyPlanRequest(BaseModel):
|
||||
# The server-issued plan version the user confirmed; a mismatch is a 409.
|
||||
expected_version: int
|
||||
expected_checksum: str | None = None
|
||||
493
photo_pipeline/services/rename_apply.py
Normal file
493
photo_pipeline/services/rename_apply.py
Normal file
@@ -0,0 +1,493 @@
|
||||
"""Apply and recover guarded renames (US04-03, US04-04).
|
||||
|
||||
This is the only module that moves a photo library's files. Every mutation is
|
||||
bracketed by the journal (US04-02): intent is recorded **before** the move, and the
|
||||
journal plus the real files on disk are the sole basis for recovery afterwards. The
|
||||
two halves live together because they are one crash-safe transaction — apply writes
|
||||
exactly the evidence recovery reads.
|
||||
|
||||
Per operation the sequence is:
|
||||
|
||||
```
|
||||
journal.begin (moving) ← intent persisted BEFORE any disk change
|
||||
recheck preconditions ← immediately before the mutation, never trusting the plan
|
||||
move ← same-fs rename | staged case-only | copy-verify-delete
|
||||
journal → moved
|
||||
update assets.current_path + path occurrences (one transaction)
|
||||
journal → database_updated
|
||||
verify bytes and paths on disk
|
||||
journal → verified → complete
|
||||
```
|
||||
|
||||
Safety rules that are never relaxed:
|
||||
|
||||
- An unexpected occupant at the destination is **never** overwritten; the operation
|
||||
fails and the source is left untouched.
|
||||
- Cross-filesystem moves are copy → verify hash → delete source, so the bytes exist
|
||||
in two places until they are proven identical.
|
||||
- A case-only rename goes through a staged temporary name, because on a
|
||||
case-insensitive filesystem source and destination are the same directory entry.
|
||||
- Recovery never guesses: a ``manual`` verdict blocks unrelated mutations instead of
|
||||
retrying. Resume and rollback revalidate the recorded hashes and refuse a path
|
||||
whose bytes changed.
|
||||
- Recovery and rollback are idempotent — replaying them after repeated restarts
|
||||
converges on the same state.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline.models import Asset, AssetPath, RenamePlan
|
||||
from photo_pipeline.services import hashing
|
||||
from photo_pipeline.services.rename_journal import (
|
||||
MANUAL,
|
||||
RESUMABLE,
|
||||
JournalState,
|
||||
RenameJournal,
|
||||
)
|
||||
|
||||
# ``applied`` is included so a repeated apply is a safe no-op rather than an error:
|
||||
# every operation is already complete and simply skips (idempotent confirmation).
|
||||
# ``planned`` is what recovery leaves behind after resetting a resumable operation,
|
||||
# so excluding it would make "resumable" impossible to actually resume. An invalid
|
||||
# plan is still refused — by the separate blockers check, not by state.
|
||||
APPLYABLE_PLAN_STATES = frozenset({"planned", "validated", "applying", "failed", "applied"})
|
||||
|
||||
|
||||
class ApplyError(RuntimeError):
|
||||
"""The request cannot be applied (unknown plan, invalid state, blocked)."""
|
||||
|
||||
|
||||
class ApplyConflict(RuntimeError):
|
||||
"""Optimistic confirmation failed; the plan changed since it was previewed."""
|
||||
|
||||
|
||||
class PreconditionFailed(RuntimeError):
|
||||
"""A precondition rechecked immediately before mutation no longer holds."""
|
||||
|
||||
def __init__(self, code: str, message: str) -> None:
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _maybe_fault(state: str) -> None:
|
||||
"""Test-only crash barrier (concept §18 fault injection).
|
||||
|
||||
When ``PHOTO_PIPELINE_FAULT_AFTER`` names a journal state, the process dies
|
||||
abruptly the moment that state has been persisted — modelling a real kill at
|
||||
exactly that transition. Never set outside tests.
|
||||
"""
|
||||
if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state:
|
||||
os._exit(9)
|
||||
|
||||
|
||||
class RenameApplyService:
|
||||
def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None:
|
||||
self._session_factory = session_factory
|
||||
self._roots = tuple(Path(root) for root in library_roots)
|
||||
self.journal = RenameJournal(session_factory)
|
||||
|
||||
# ── apply ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def apply(
|
||||
self,
|
||||
plan_id: str,
|
||||
*,
|
||||
expected_version: int,
|
||||
expected_checksum: str | None = None,
|
||||
worker_id: str = "apply",
|
||||
) -> dict:
|
||||
"""Apply a confirmed plan. Requires the current version (and optionally the
|
||||
checksum) so a stale browser confirmation cannot run a changed plan."""
|
||||
plan = self._require_plan(plan_id)
|
||||
if plan["version"] != expected_version:
|
||||
raise ApplyConflict(
|
||||
f"plan {plan_id} is at version {plan['version']}, expected {expected_version}"
|
||||
)
|
||||
if expected_checksum is not None and plan["checksum"] != expected_checksum:
|
||||
raise ApplyConflict(f"plan {plan_id} checksum changed since it was previewed")
|
||||
if plan["state"] not in APPLYABLE_PLAN_STATES:
|
||||
raise ApplyError(f"plan {plan_id} is {plan['state']} and cannot be applied")
|
||||
if plan["blockers"]:
|
||||
raise ApplyError(f"plan {plan_id} has unresolved blockers: {plan['blockers']}")
|
||||
|
||||
# Exclusive mutation lease: refuse while any other rename may have the
|
||||
# library half-moved (concept §16 lock hierarchy).
|
||||
blocking = [row for row in self.journal.incomplete() if row["plan_id"] != plan_id]
|
||||
if blocking:
|
||||
raise ApplyError(
|
||||
f"another rename is unresolved ({blocking[0]['id']}); recover it first"
|
||||
)
|
||||
|
||||
token = self._claim_plan(plan_id)
|
||||
applied = failed = skipped = 0
|
||||
for operation in self.journal.operations(plan_id):
|
||||
if operation["journal_state"] == JournalState.COMPLETE:
|
||||
skipped += 1 # repeated apply is a no-op for finished work
|
||||
continue
|
||||
try:
|
||||
self._apply_one(operation, token=token, worker_id=worker_id)
|
||||
applied += 1
|
||||
except PreconditionFailed as error:
|
||||
self.journal.transition(
|
||||
operation["id"],
|
||||
JournalState.FAILED,
|
||||
fencing_token=token,
|
||||
error=(error.code, str(error)),
|
||||
)
|
||||
failed += 1
|
||||
except Exception as error: # unexpected: record and stop touching disk
|
||||
self.journal.transition(
|
||||
operation["id"],
|
||||
JournalState.FAILED,
|
||||
fencing_token=token,
|
||||
error=("apply_error", str(error)),
|
||||
)
|
||||
failed += 1
|
||||
state = self.journal.sync_plan_state(plan_id)
|
||||
return {
|
||||
"plan_id": plan_id,
|
||||
"applied": applied,
|
||||
"failed": failed,
|
||||
"skipped": skipped,
|
||||
"state": state,
|
||||
}
|
||||
|
||||
def _apply_one(self, operation: dict, *, token: int, worker_id: str) -> None:
|
||||
source = Path(operation["source_path"])
|
||||
destination = Path(operation["destination_path"])
|
||||
|
||||
# 1. Intent first — after this point a crash is recoverable from evidence.
|
||||
self.journal.begin(operation["id"], worker_id=worker_id, fencing_token=token)
|
||||
_maybe_fault(JournalState.MOVING)
|
||||
|
||||
# 2. Recheck preconditions immediately before mutating, never trusting the
|
||||
# plan's snapshot: files can change between preview and confirmation.
|
||||
self._recheck(operation, source, destination)
|
||||
|
||||
# 3. The move itself.
|
||||
if operation["case_only"]:
|
||||
self._staged_case_rename(source, destination)
|
||||
elif operation["same_filesystem"] is False:
|
||||
self._copy_verify_delete(source, destination)
|
||||
else:
|
||||
os.rename(source, destination)
|
||||
|
||||
self.journal.transition(operation["id"], JournalState.MOVED, fencing_token=token)
|
||||
_maybe_fault(JournalState.MOVED)
|
||||
|
||||
# 4. Database: stable IDs keep their identity, paths are re-pointed and the
|
||||
# old occurrence is closed — all in one transaction.
|
||||
self._reconcile_paths(operation, source, destination)
|
||||
self.journal.transition(operation["id"], JournalState.DATABASE_UPDATED, fencing_token=token)
|
||||
_maybe_fault(JournalState.DATABASE_UPDATED)
|
||||
|
||||
# 5. Postconditions: the bytes really are at the new paths.
|
||||
self._verify(operation, destination)
|
||||
self.journal.transition(operation["id"], JournalState.VERIFIED, fencing_token=token)
|
||||
_maybe_fault(JournalState.VERIFIED)
|
||||
|
||||
self.journal.transition(operation["id"], JournalState.COMPLETE, fencing_token=token)
|
||||
_maybe_fault(JournalState.COMPLETE)
|
||||
|
||||
def _recheck(self, operation: dict, source: Path, destination: Path) -> None:
|
||||
if not source.exists():
|
||||
raise PreconditionFailed("source_missing", f"source {source} disappeared")
|
||||
if source.is_symlink() or destination.is_symlink():
|
||||
raise PreconditionFailed("symlink", "refusing to rename through a symlink")
|
||||
# An unexpected occupant is never overwritten. For a case-only rename the
|
||||
# destination resolves to the same entry as the source, so its existence is
|
||||
# expected and not a collision.
|
||||
if destination.exists() and not operation["case_only"]:
|
||||
raise PreconditionFailed("destination_exists", f"destination {destination} is occupied")
|
||||
if not self._within_roots(source) or not self._within_roots(destination):
|
||||
raise PreconditionFailed("root_escape", "path is outside the library roots")
|
||||
|
||||
expected = operation.get("expected_sha256") or {}
|
||||
with self._session_factory() as session:
|
||||
assets = session.scalars(
|
||||
select(Asset).where(Asset.id.in_(operation["asset_ids"]))
|
||||
).all()
|
||||
current = {asset.id: asset.current_path for asset in assets}
|
||||
for asset_id, path in current.items():
|
||||
file_path = Path(path) if path else None
|
||||
if file_path is None or not file_path.exists():
|
||||
raise PreconditionFailed("source_changed", f"asset {asset_id} is missing")
|
||||
wanted = expected.get(asset_id)
|
||||
if wanted and hashing.sha256_file(file_path) != wanted:
|
||||
raise PreconditionFailed(
|
||||
"source_changed", f"asset {asset_id} changed since the plan was built"
|
||||
)
|
||||
|
||||
def _within_roots(self, path: Path) -> bool:
|
||||
if not self._roots:
|
||||
return False
|
||||
return any(path == root or root in path.parents for root in self._roots)
|
||||
|
||||
# ── move procedures ───────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _staged_case_rename(source: Path, destination: Path) -> None:
|
||||
"""Case-only rename via a temporary name: on a case-insensitive filesystem
|
||||
source and destination are the same entry, so a direct rename is a no-op or
|
||||
an error depending on the platform."""
|
||||
staging = source.with_name(f".rename-{uuid.uuid4().hex[:8]}")
|
||||
os.rename(source, staging)
|
||||
try:
|
||||
os.rename(staging, destination)
|
||||
except OSError:
|
||||
os.rename(staging, source) # put it back rather than leave it staged
|
||||
raise
|
||||
|
||||
@staticmethod
|
||||
def _copy_verify_delete(source: Path, destination: Path) -> None:
|
||||
"""Cross-filesystem move: the bytes exist in both places until they are
|
||||
proven identical, and only then is the source removed."""
|
||||
staging = destination.with_name(f".rename-{uuid.uuid4().hex[:8]}")
|
||||
shutil.copytree(source, staging)
|
||||
for original in sorted(p for p in source.rglob("*") if p.is_file()):
|
||||
copied = staging / original.relative_to(source)
|
||||
if not copied.exists():
|
||||
shutil.rmtree(staging, ignore_errors=True)
|
||||
raise PreconditionFailed("copy_incomplete", f"{original} was not copied")
|
||||
if hashing.sha256_file(original) != hashing.sha256_file(copied):
|
||||
shutil.rmtree(staging, ignore_errors=True)
|
||||
raise PreconditionFailed("copy_mismatch", f"{original} copied with wrong bytes")
|
||||
os.rename(staging, destination)
|
||||
shutil.rmtree(source)
|
||||
|
||||
# ── database reconciliation ───────────────────────────────────────────────
|
||||
|
||||
def _reconcile_paths(self, operation: dict, source: Path, destination: Path) -> None:
|
||||
"""Re-point every asset in the operation at its new path, keeping the stable
|
||||
ID and recording the path history."""
|
||||
now = _now()
|
||||
with self._session_factory() as session:
|
||||
assets = session.scalars(
|
||||
select(Asset).where(Asset.id.in_(operation["asset_ids"]))
|
||||
).all()
|
||||
for asset in assets:
|
||||
if not asset.current_path:
|
||||
continue
|
||||
old = Path(asset.current_path)
|
||||
try:
|
||||
relative = old.relative_to(source)
|
||||
except ValueError:
|
||||
continue # already re-pointed by an earlier partial run
|
||||
new_path = destination / relative
|
||||
for row in session.scalars(
|
||||
select(AssetPath).where(
|
||||
AssetPath.asset_id == asset.id,
|
||||
AssetPath.path == asset.current_path,
|
||||
AssetPath.valid_until.is_(None),
|
||||
)
|
||||
):
|
||||
row.valid_until = now
|
||||
session.add(
|
||||
AssetPath(
|
||||
asset_id=asset.id,
|
||||
path=str(new_path),
|
||||
valid_from=now,
|
||||
reason="rename",
|
||||
)
|
||||
)
|
||||
asset.current_path = str(new_path)
|
||||
asset.state_version += 1
|
||||
asset.updated_at = now
|
||||
session.commit()
|
||||
|
||||
def _verify(self, operation: dict, destination: Path) -> None:
|
||||
"""Postcondition: every asset's recorded path exists and its bytes are the
|
||||
ones the plan expected."""
|
||||
expected = operation.get("expected_sha256") or {}
|
||||
with self._session_factory() as session:
|
||||
assets = session.scalars(
|
||||
select(Asset).where(Asset.id.in_(operation["asset_ids"]))
|
||||
).all()
|
||||
paths = {asset.id: asset.current_path for asset in assets}
|
||||
for asset_id, path in paths.items():
|
||||
if not path or not Path(path).exists():
|
||||
raise PreconditionFailed("verify_missing", f"asset {asset_id} is not at {path}")
|
||||
wanted = expected.get(asset_id)
|
||||
if wanted and hashing.sha256_file(path) != wanted:
|
||||
raise PreconditionFailed("verify_bytes", f"asset {asset_id} bytes changed")
|
||||
|
||||
# ── recovery (US04-04) ────────────────────────────────────────────────────
|
||||
|
||||
def recover(self, *, worker_id: str = "recovery") -> dict:
|
||||
"""Resolve every incomplete operation from journal + disk evidence.
|
||||
|
||||
Idempotent: running it repeatedly converges. Ambiguous (``manual``) work is
|
||||
left untouched and keeps blocking unrelated mutations.
|
||||
"""
|
||||
results = {"resumed": 0, "completed": 0, "manual": 0}
|
||||
touched_plans: set[str] = set()
|
||||
for verdict in self.journal.classify_all():
|
||||
operation = self.journal.get(verdict["operation_id"])
|
||||
touched_plans.add(operation["plan_id"])
|
||||
token = (operation["fencing_token"] or 0) + 1
|
||||
if verdict["classification"] == MANUAL:
|
||||
results["manual"] += 1
|
||||
continue
|
||||
if verdict["classification"] == RESUMABLE:
|
||||
# Nothing moved: return the operation to the planned state so a
|
||||
# later apply can retry it cleanly.
|
||||
self._to_planned(operation)
|
||||
results["resumed"] += 1
|
||||
continue
|
||||
# rollback_safe: the content is at the destination. Finish the remaining
|
||||
# bookkeeping deterministically rather than moving anything again.
|
||||
self._finish_moved(operation, token=token, worker_id=worker_id)
|
||||
results["completed"] += 1
|
||||
for plan_id in touched_plans:
|
||||
self.journal.sync_plan_state(plan_id)
|
||||
return results
|
||||
|
||||
def _to_planned(self, operation: dict) -> None:
|
||||
state = operation["journal_state"]
|
||||
if state == JournalState.PLANNED:
|
||||
return
|
||||
if state in (JournalState.MOVING, JournalState.FAILED):
|
||||
self.journal.transition(operation["id"], JournalState.PLANNED)
|
||||
elif state == JournalState.ROLLBACK_REQUIRED:
|
||||
self.journal.transition(operation["id"], JournalState.ROLLED_BACK)
|
||||
|
||||
def _finish_moved(self, operation: dict, *, token: int, worker_id: str) -> None:
|
||||
"""Drive an operation whose bytes are already at the destination through the
|
||||
remaining states, re-running the idempotent bookkeeping."""
|
||||
operation_id = operation["id"]
|
||||
source = Path(operation["source_path"])
|
||||
destination = Path(operation["destination_path"])
|
||||
state = operation["journal_state"]
|
||||
|
||||
if state == JournalState.ROLLBACK_REQUIRED:
|
||||
self.rollback_operation(operation_id, token=token)
|
||||
return
|
||||
|
||||
if state == JournalState.MOVING:
|
||||
self.journal.transition(operation_id, JournalState.MOVED, fencing_token=token)
|
||||
state = JournalState.MOVED
|
||||
if state == JournalState.MOVED:
|
||||
self._reconcile_paths(operation, source, destination)
|
||||
self.journal.transition(
|
||||
operation_id, JournalState.DATABASE_UPDATED, fencing_token=token
|
||||
)
|
||||
state = JournalState.DATABASE_UPDATED
|
||||
if state == JournalState.DATABASE_UPDATED:
|
||||
self._verify(operation, destination)
|
||||
self.journal.transition(operation_id, JournalState.VERIFIED, fencing_token=token)
|
||||
state = JournalState.VERIFIED
|
||||
if state == JournalState.VERIFIED:
|
||||
self.journal.transition(operation_id, JournalState.COMPLETE, fencing_token=token)
|
||||
|
||||
# ── rollback ──────────────────────────────────────────────────────────────
|
||||
|
||||
def rollback_operation(self, operation_id: str, *, token: int | None = None) -> dict:
|
||||
"""Move an operation's content back to its source, revalidating the recorded
|
||||
hashes first. Refuses when the source is occupied or the bytes changed."""
|
||||
operation = self.journal.get(operation_id)
|
||||
if operation is None:
|
||||
raise ApplyError(f"unknown rename operation {operation_id!r}")
|
||||
source = Path(operation["source_path"])
|
||||
destination = Path(operation["destination_path"])
|
||||
token = token if token is not None else (operation["fencing_token"] or 0) + 1
|
||||
|
||||
if operation["journal_state"] not in (
|
||||
JournalState.MOVED,
|
||||
JournalState.DATABASE_UPDATED,
|
||||
JournalState.VERIFIED,
|
||||
JournalState.ROLLBACK_REQUIRED,
|
||||
JournalState.MOVING,
|
||||
):
|
||||
raise ApplyError(
|
||||
f"operation {operation_id} is {operation['journal_state']}; nothing to roll back"
|
||||
)
|
||||
if source.exists() and not operation["case_only"]:
|
||||
raise ApplyError(f"source {source} is occupied; refusing to overwrite it")
|
||||
if not destination.exists():
|
||||
raise ApplyError(f"destination {destination} is missing; cannot roll back")
|
||||
|
||||
# Revalidate before moving anything back: a changed file is not ours to move.
|
||||
expected = operation.get("expected_sha256") or {}
|
||||
if expected:
|
||||
for original in sorted(p for p in destination.rglob("*") if p.is_file()):
|
||||
digest = hashing.sha256_file(original)
|
||||
if expected and digest not in expected.values():
|
||||
raise ApplyError(
|
||||
f"{original} does not match the recorded hashes; manual recovery required"
|
||||
)
|
||||
|
||||
if operation["journal_state"] != JournalState.ROLLBACK_REQUIRED:
|
||||
self.journal.transition(
|
||||
operation_id, JournalState.ROLLBACK_REQUIRED, fencing_token=token
|
||||
)
|
||||
if operation["case_only"]:
|
||||
self._staged_case_rename(destination, source)
|
||||
elif operation["same_filesystem"] is False:
|
||||
self._copy_verify_delete(destination, source)
|
||||
else:
|
||||
os.rename(destination, source)
|
||||
|
||||
self._reconcile_paths(operation, destination, source)
|
||||
self.journal.transition(operation_id, JournalState.ROLLED_BACK, fencing_token=token)
|
||||
self.journal.sync_plan_state(operation["plan_id"])
|
||||
return self.journal.get(operation_id)
|
||||
|
||||
def rollback_plan(self, plan_id: str) -> dict:
|
||||
"""Roll back every applied operation of a plan, newest first."""
|
||||
rolled = 0
|
||||
for operation in reversed(self.journal.operations(plan_id)):
|
||||
if operation["journal_state"] in (
|
||||
JournalState.MOVED,
|
||||
JournalState.DATABASE_UPDATED,
|
||||
JournalState.VERIFIED,
|
||||
JournalState.COMPLETE,
|
||||
JournalState.ROLLBACK_REQUIRED,
|
||||
):
|
||||
if operation["journal_state"] == JournalState.COMPLETE:
|
||||
# A completed operation must be reopened before it can move back.
|
||||
continue
|
||||
self.rollback_operation(operation["id"])
|
||||
rolled += 1
|
||||
state = self.journal.sync_plan_state(plan_id)
|
||||
return {"plan_id": plan_id, "rolled_back": rolled, "state": state}
|
||||
|
||||
# ── helpers ───────────────────────────────────────────────────────────────
|
||||
|
||||
def _require_plan(self, plan_id: str) -> dict:
|
||||
with self._session_factory() as session:
|
||||
plan = session.get(RenamePlan, plan_id)
|
||||
if plan is None:
|
||||
raise ApplyError(f"unknown rename plan {plan_id!r}")
|
||||
return {
|
||||
"id": plan.id,
|
||||
"state": plan.state,
|
||||
"version": plan.version,
|
||||
"checksum": plan.checksum,
|
||||
"blockers": json.loads(plan.blockers or "[]"),
|
||||
}
|
||||
|
||||
def _claim_plan(self, plan_id: str) -> int:
|
||||
"""Bump the plan version and use it as this attempt's fencing token, so a
|
||||
worker from a superseded attempt cannot commit."""
|
||||
with self._session_factory() as session:
|
||||
plan = session.get(RenamePlan, plan_id)
|
||||
plan.version += 1
|
||||
plan.state = "applying"
|
||||
plan.updated_at = _now()
|
||||
token = plan.version
|
||||
session.commit()
|
||||
return token
|
||||
@@ -229,6 +229,12 @@ class RenameJournal:
|
||||
classification, reason = _classify(row["journal_state"], source_exists, destination_exists)
|
||||
return {
|
||||
"operation_id": operation_id,
|
||||
"plan_id": row["plan_id"],
|
||||
"album": row["album"],
|
||||
# The paths travel with the verdict so a reviewer can see which folder is
|
||||
# unresolved without correlating an opaque operation id by hand.
|
||||
"source_path": row["source_path"],
|
||||
"destination_path": row["destination_path"],
|
||||
"journal_state": row["journal_state"],
|
||||
"classification": classification,
|
||||
"reason": reason,
|
||||
|
||||
@@ -418,8 +418,17 @@ def _plan_dict(plan: RenamePlan, operations: list[RenameOperation]) -> dict:
|
||||
"asset_count": operation.asset_count,
|
||||
"same_filesystem": operation.same_filesystem,
|
||||
"case_only": operation.case_only,
|
||||
"issues": json.loads(operation.issues or "[]"),
|
||||
# Severity is decided here, not in the browser: a client that has to
|
||||
# keep its own copy of BLOCKING_CODES will eventually disagree with
|
||||
# the validator about whether a plan can run.
|
||||
"issues": [
|
||||
{**issue, "blocking": issue["code"] in BLOCKING_CODES}
|
||||
for issue in json.loads(operation.issues or "[]")
|
||||
],
|
||||
"journal_state": operation.journal_state,
|
||||
"verified_at": operation.verified_at.isoformat() if operation.verified_at else None,
|
||||
"error_code": operation.error_code,
|
||||
"error_message": operation.error_message,
|
||||
}
|
||||
for operation in operations
|
||||
],
|
||||
|
||||
@@ -30,4 +30,5 @@ testpaths = ["tests"]
|
||||
markers = [
|
||||
"phase_b: Phase B end-to-end acceptance (US02-07) — API, worker-recovery, and browser journeys",
|
||||
"phase_c: Phase C end-to-end acceptance (US03-05) — album proposal API and browser journeys",
|
||||
"phase_d: Phase D end-to-end acceptance (US04-06) — guarded rename API, fault, and browser journeys",
|
||||
]
|
||||
|
||||
@@ -14,7 +14,9 @@ import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
@@ -159,3 +161,88 @@ def wait_until(predicate, *, timeout: float = 20, interval: float = 0.1):
|
||||
return value
|
||||
time.sleep(interval)
|
||||
raise AssertionError("condition not met before timeout")
|
||||
|
||||
|
||||
# ── Phase D: an analysed album, ready to be named and renamed ────────────────
|
||||
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
class session_factory:
|
||||
"""Session factory against a seeded database, for the few things a test has to
|
||||
set up or inspect below the API — journal states, mainly."""
|
||||
|
||||
def __init__(self, seeded: Seeded) -> None:
|
||||
self._seeded = seeded
|
||||
|
||||
def __enter__(self):
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
|
||||
config = Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(self._seeded.data),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(self._seeded.lib),
|
||||
}
|
||||
)
|
||||
run_migrations(config.database_url)
|
||||
self._engine = create_db_engine(config.database_url)
|
||||
return create_session_factory(self._engine)
|
||||
|
||||
def __exit__(self, *_):
|
||||
self._engine.dispose()
|
||||
return False
|
||||
|
||||
|
||||
def seed_album(tmp_path: Path, album: str = "rome", names: tuple[str, ...] = ("a.jpg", "b.jpg")):
|
||||
"""A library holding one album folder whose photos are confirmed SFW and analysed
|
||||
— the state a naming proposal, and therefore a rename plan, is built from."""
|
||||
from sqlalchemy import select
|
||||
|
||||
from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
|
||||
from photo_pipeline.services.inventory import InventoryService
|
||||
|
||||
seeded = seed_library(tmp_path, {}, {})
|
||||
folder = seeded.lib / album
|
||||
folder.mkdir(parents=True)
|
||||
for index, name in enumerate(names):
|
||||
image(folder / name, index + 1)
|
||||
|
||||
with session_factory(seeded) as sf:
|
||||
InventoryService(sf).scan(seeded.lib)
|
||||
with sf() as session:
|
||||
rows = list(session.execute(select(Asset.id, Asset.current_path)).all())
|
||||
for asset_id, path in rows:
|
||||
session.add(
|
||||
SafetyReview(
|
||||
id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
AnalysisResult(
|
||||
asset_id=asset_id,
|
||||
status="analyzed",
|
||||
description=f"a view of {path}",
|
||||
tags='["ruins", "city"]',
|
||||
approx_year=2019,
|
||||
location_hint="Rome",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
seeded.asset_ids.update({Path(path).stem: aid for aid, path in rows})
|
||||
return seeded
|
||||
|
||||
|
||||
def approve_album(base: str, *, album: str = "rome", name: str) -> None:
|
||||
"""Generate a proposal, set its final name, and approve it over HTTP."""
|
||||
httpx.post(f"{base}/api/v1/albums/proposals", json={}, timeout=10).raise_for_status()
|
||||
for payload, route in (
|
||||
({"name": name}, "edit"),
|
||||
({}, "approve"),
|
||||
):
|
||||
current = httpx.get(f"{base}/api/v1/albums/proposals/{album}", timeout=10).json()
|
||||
httpx.post(
|
||||
f"{base}/api/v1/albums/proposals/{album}/{route}",
|
||||
json={**payload, "expected_version": current["version"]},
|
||||
timeout=10,
|
||||
).raise_for_status()
|
||||
|
||||
377
tests/e2e/test_phase_d_pipeline.py
Normal file
377
tests/e2e/test_phase_d_pipeline.py
Normal file
@@ -0,0 +1,377 @@
|
||||
"""Phase D end-to-end acceptance (US04-06): guarded renaming, black box.
|
||||
|
||||
Every journey here drives a real ``photo_pipeline serve`` child process over HTTP —
|
||||
plan, export, confirm, apply, collide, go stale, crash, recover, roll back. The
|
||||
crashes are real: the server is killed by the ``PHOTO_PIPELINE_FAULT_AFTER`` barrier
|
||||
at each persisted journal transition in turn, then a fresh process is started against
|
||||
the same database and library and has to reconcile the wreckage from evidence alone.
|
||||
|
||||
Photos really move. After every journey the assertions read the filesystem and the
|
||||
inventory back: the asset set, the stable IDs, and the content hashes must be exactly
|
||||
what they were before, only at new paths.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from tests.e2e._pipeline_harness import Server, approve_album, seed_album, session_factory
|
||||
|
||||
pytestmark = pytest.mark.phase_d
|
||||
|
||||
TIMEOUT = 10
|
||||
APPROVED = "2019 Rome"
|
||||
CRASH_POINTS = ["moving", "moved", "database_updated", "verified"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def server(tmp_path):
|
||||
seeded = seed_album(tmp_path)
|
||||
running = Server(seeded).start()
|
||||
running.seeded = seeded
|
||||
try:
|
||||
yield running
|
||||
finally:
|
||||
running.stop()
|
||||
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _plan(base) -> dict:
|
||||
response = httpx.post(f"{base}/api/v1/rename-plans", timeout=TIMEOUT)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
def _get_plan(base, plan_id) -> dict:
|
||||
return httpx.get(f"{base}/api/v1/rename-plans/{plan_id}", timeout=TIMEOUT).json()
|
||||
|
||||
|
||||
def _apply(base, plan, **body):
|
||||
payload = {"expected_version": plan["version"], **body}
|
||||
return httpx.post(
|
||||
f"{base}/api/v1/rename-plans/{plan['id']}/apply", json=payload, timeout=TIMEOUT
|
||||
)
|
||||
|
||||
|
||||
def _inventory(base) -> dict[str, dict]:
|
||||
"""Every asset by stable ID, so identity can be compared across a rename."""
|
||||
items = httpx.get(
|
||||
f"{base}/api/v1/inventory/assets", params={"limit": 200}, timeout=TIMEOUT
|
||||
).json()["items"]
|
||||
return {item["id"]: item for item in items}
|
||||
|
||||
|
||||
def _content(root) -> dict[str, bytes]:
|
||||
return {
|
||||
str(path.relative_to(root)): path.read_bytes()
|
||||
for path in sorted(root.rglob("*"))
|
||||
if path.is_file()
|
||||
}
|
||||
|
||||
|
||||
def _recovery(base) -> dict:
|
||||
return httpx.get(f"{base}/api/v1/rename-recovery", timeout=TIMEOUT).json()
|
||||
|
||||
|
||||
def _journal_states(seeded, plan_id) -> list[str]:
|
||||
from photo_pipeline.services.rename_journal import RenameJournal
|
||||
|
||||
with session_factory(seeded) as sf:
|
||||
return [row["journal_state"] for row in RenameJournal(sf).operations(plan_id)]
|
||||
|
||||
|
||||
# ── US04-01: plan and export ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_plan_and_export_describe_every_move_without_touching_the_library(server):
|
||||
before = _content(server.seeded.lib)
|
||||
approve_album(server.base, name=APPROVED)
|
||||
plan = _plan(server.base)
|
||||
|
||||
assert plan["state"] == "validated" and plan["operation_count"] == 1
|
||||
operation = plan["operations"][0]
|
||||
assert operation["source_path"].endswith("/rome")
|
||||
assert operation["destination_path"].endswith(f"/{APPROVED}")
|
||||
assert operation["asset_count"] == 2
|
||||
assert set(operation["asset_ids"]) == set(_inventory(server.base))
|
||||
|
||||
export = httpx.get(
|
||||
f"{server.base}/api/v1/rename-plans/{plan['id']}/export", timeout=TIMEOUT
|
||||
).json()
|
||||
assert export["schema_version"] == 1
|
||||
assert export["checksum"] == plan["checksum"]
|
||||
assert [op["source_path"] for op in export["operations"]] == [operation["source_path"]]
|
||||
# Portable evidence must not smuggle out anything sensitive.
|
||||
assert "token" not in str(export).lower() and "key" not in str(export).lower()
|
||||
|
||||
# Planning is a preview: not one byte moved.
|
||||
assert _content(server.seeded.lib) == before
|
||||
|
||||
|
||||
# ── US04-03: confirmation and apply ──────────────────────────────────────────
|
||||
|
||||
|
||||
def test_apply_requires_the_current_confirmation_token(server):
|
||||
approve_album(server.base, name=APPROVED)
|
||||
plan = _plan(server.base)
|
||||
|
||||
stale = _apply(server.base, {**plan, "version": plan["version"] + 7})
|
||||
assert stale.status_code == 409 and stale.json()["error"]["code"] == "version_conflict"
|
||||
|
||||
wrong_checksum = _apply(server.base, plan, expected_checksum="0" * 64)
|
||||
assert wrong_checksum.status_code == 409
|
||||
|
||||
assert (server.seeded.lib / "rome").is_dir(), "a refused confirmation moves nothing"
|
||||
|
||||
|
||||
def test_a_valid_apply_preserves_ids_hashes_and_the_asset_set(server):
|
||||
approve_album(server.base, name=APPROVED)
|
||||
before = _inventory(server.base)
|
||||
before_content = _content(server.seeded.lib)
|
||||
plan = _plan(server.base)
|
||||
|
||||
applied = _apply(server.base, plan, expected_checksum=plan["checksum"]).json()
|
||||
assert applied["applied"] == 1 and applied["failed"] == 0 and applied["state"] == "applied"
|
||||
|
||||
after = _inventory(server.base)
|
||||
assert set(after) == set(before), "renaming must not change asset identity"
|
||||
assert {item["current_sha256"] for item in after.values()} == {
|
||||
item["current_sha256"] for item in before.values()
|
||||
}
|
||||
assert all(APPROVED in item["current_path"] for item in after.values())
|
||||
# Same bytes, new folder — nothing was rewritten in the move.
|
||||
assert _content(server.seeded.lib) == {
|
||||
key.replace("rome/", f"{APPROVED}/"): value for key, value in before_content.items()
|
||||
}
|
||||
assert _journal_states(server.seeded, plan["id"]) == ["complete"]
|
||||
|
||||
|
||||
def test_a_case_only_rename_applies_on_a_case_insensitive_filesystem(tmp_path):
|
||||
seeded = seed_album(tmp_path, album="rome")
|
||||
running = Server(seeded).start()
|
||||
try:
|
||||
approve_album(running.base, name="Rome")
|
||||
plan = _plan(running.base)
|
||||
assert plan["operations"][0]["case_only"] is True
|
||||
|
||||
assert _apply(running.base, plan).json()["applied"] == 1
|
||||
entries = {path.name for path in seeded.lib.iterdir()}
|
||||
assert "Rome" in entries
|
||||
# The staged intermediate name must not survive the procedure.
|
||||
assert not any(name.startswith(".rename-") for name in entries)
|
||||
assert all("/Rome/" in item["current_path"] for item in _inventory(running.base).values())
|
||||
finally:
|
||||
running.stop()
|
||||
|
||||
|
||||
def test_a_collision_is_refused_and_the_occupant_survives(server):
|
||||
approve_album(server.base, name=APPROVED)
|
||||
occupied = server.seeded.lib / APPROVED
|
||||
occupied.mkdir()
|
||||
(occupied / "precious.jpg").write_bytes(b"do not lose me")
|
||||
|
||||
plan = _plan(server.base)
|
||||
assert plan["state"] == "invalid" and "destination_exists" in plan["blockers"]
|
||||
|
||||
refused = _apply(server.base, plan)
|
||||
assert refused.status_code == 422 and refused.json()["error"]["code"] == "cannot_apply"
|
||||
assert (occupied / "precious.jpg").read_bytes() == b"do not lose me"
|
||||
assert (server.seeded.lib / "rome").is_dir()
|
||||
|
||||
|
||||
def test_a_source_that_changed_after_planning_is_refused(server):
|
||||
approve_album(server.base, name=APPROVED)
|
||||
plan = _plan(server.base)
|
||||
# The plan recorded per-asset hashes; the file changes before confirmation.
|
||||
(server.seeded.lib / "rome" / "a.jpg").write_bytes(b"tampered")
|
||||
|
||||
result = _apply(server.base, plan).json()
|
||||
assert result["failed"] == 1 and result["applied"] == 0
|
||||
assert (server.seeded.lib / "rome").is_dir(), "a failed precondition leaves the source alone"
|
||||
operation = _get_plan(server.base, plan["id"])["operations"][0]
|
||||
assert operation["journal_state"] == "failed"
|
||||
assert operation["error_code"] == "source_changed"
|
||||
|
||||
|
||||
# ── US04-04: crash points, recovery, rollback ────────────────────────────────
|
||||
|
||||
|
||||
def _crash_during_apply(seeded, plan, state):
|
||||
"""Apply with the fault barrier armed: the server dies at ``state``, mid-move."""
|
||||
crashing = Server(seeded, extra_env={"PHOTO_PIPELINE_FAULT_AFTER": state}).start()
|
||||
try:
|
||||
with pytest.raises(httpx.HTTPError):
|
||||
_apply(crashing.base, plan)
|
||||
finally:
|
||||
crashing.stop()
|
||||
assert crashing.proc is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("crash_point", CRASH_POINTS)
|
||||
def test_every_journal_crash_point_recovers_without_losing_content(tmp_path, crash_point):
|
||||
seeded = seed_album(tmp_path)
|
||||
first = Server(seeded).start()
|
||||
try:
|
||||
approve_album(first.base, name=APPROVED)
|
||||
before = _inventory(first.base)
|
||||
before_bytes = sorted(_content(seeded.lib).values())
|
||||
plan = _plan(first.base)
|
||||
finally:
|
||||
first.stop()
|
||||
|
||||
_crash_during_apply(seeded, plan, crash_point)
|
||||
|
||||
# A brand-new process, no in-memory state: everything comes from the journal.
|
||||
restarted = Server(seeded).start()
|
||||
try:
|
||||
recovery = _recovery(restarted.base)
|
||||
assert recovery["items"], f"a crash at {crash_point} must leave visible evidence"
|
||||
assert recovery["items"][0]["journal_state"] == crash_point
|
||||
# Only a crash that could have left the library half-renamed blocks other
|
||||
# work. `verified` is past every filesystem and database change — the move
|
||||
# is done and checked, just not flagged complete — so it blocks nothing.
|
||||
assert recovery["blocks_mutation"] is (crash_point != "verified")
|
||||
|
||||
resolved = httpx.post(
|
||||
f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT
|
||||
).json()
|
||||
assert resolved["manual"] == 0, "an interrupted rename must be decidable from evidence"
|
||||
|
||||
# Recovery is idempotent: running it again changes nothing.
|
||||
assert _recovery(restarted.base)["blocks_mutation"] is False
|
||||
httpx.post(f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT)
|
||||
|
||||
# A resumable crash is left ready to run again; finish it so every crash
|
||||
# point converges on the same observable end state.
|
||||
current = _get_plan(restarted.base, plan["id"])
|
||||
if current["state"] != "applied":
|
||||
_apply(restarted.base, current)
|
||||
|
||||
after = _inventory(restarted.base)
|
||||
assert set(after) == set(before), "no asset may be lost or invented by a crash"
|
||||
assert {item["current_sha256"] for item in after.values()} == {
|
||||
item["current_sha256"] for item in before.values()
|
||||
}
|
||||
assert sorted(_content(seeded.lib).values()) == before_bytes
|
||||
assert (seeded.lib / APPROVED).is_dir() and not (seeded.lib / "rome").exists()
|
||||
assert all(APPROVED in item["current_path"] for item in after.values())
|
||||
assert _journal_states(seeded, plan["id"]) == ["complete"]
|
||||
finally:
|
||||
restarted.stop()
|
||||
|
||||
|
||||
def test_ambiguous_evidence_is_kept_for_a_human_and_keeps_blocking(tmp_path):
|
||||
seeded = seed_album(tmp_path)
|
||||
first = Server(seeded).start()
|
||||
try:
|
||||
approve_album(first.base, name=APPROVED)
|
||||
plan = _plan(first.base)
|
||||
finally:
|
||||
first.stop()
|
||||
|
||||
_crash_during_apply(seeded, plan, "moving")
|
||||
# Someone creates the destination while the operation is unresolved: now both
|
||||
# paths exist and nothing can tell which one holds the truth.
|
||||
(seeded.lib / APPROVED).mkdir(exist_ok=True)
|
||||
|
||||
restarted = Server(seeded).start()
|
||||
try:
|
||||
assert _recovery(restarted.base)["items"][0]["classification"] == "manual"
|
||||
resolved = httpx.post(
|
||||
f"{restarted.base}/api/v1/rename-recovery/resolve", timeout=TIMEOUT
|
||||
).json()
|
||||
assert resolved["manual"] == 1 and resolved["resumed"] == 0 and resolved["completed"] == 0
|
||||
# Still blocking, and still nothing guessed.
|
||||
assert _recovery(restarted.base)["blocks_mutation"] is True
|
||||
assert (seeded.lib / "rome").is_dir()
|
||||
finally:
|
||||
restarted.stop()
|
||||
|
||||
|
||||
def test_an_unresolved_rename_is_the_cancellation_boundary(tmp_path):
|
||||
"""There is no cancel button once a rename starts. The boundary is that nothing
|
||||
else may mutate the library until the interrupted work is resolved."""
|
||||
seeded = seed_album(tmp_path)
|
||||
first = Server(seeded).start()
|
||||
try:
|
||||
approve_album(first.base, name=APPROVED)
|
||||
plan = _plan(first.base)
|
||||
finally:
|
||||
first.stop()
|
||||
|
||||
_crash_during_apply(seeded, plan, "moving")
|
||||
|
||||
restarted = Server(seeded).start()
|
||||
try:
|
||||
assert _recovery(restarted.base)["blocks_mutation"] is True
|
||||
|
||||
refused = httpx.post(f"{restarted.base}/api/v1/albums/proposals", json={}, timeout=TIMEOUT)
|
||||
assert refused.status_code == 409
|
||||
assert refused.json()["error"]["code"] == "rename_recovery_required"
|
||||
|
||||
# Reading stays available throughout — only mutation is paused.
|
||||
assert httpx.get(f"{restarted.base}/api/v1/albums/evidence", timeout=TIMEOUT).status_code
|
||||
assert len(_inventory(restarted.base)) == 2
|
||||
finally:
|
||||
restarted.stop()
|
||||
|
||||
|
||||
def test_rollback_returns_an_interrupted_move_to_its_source(tmp_path):
|
||||
seeded = seed_album(tmp_path)
|
||||
first = Server(seeded).start()
|
||||
try:
|
||||
approve_album(first.base, name=APPROVED)
|
||||
before = _inventory(first.base)
|
||||
plan = _plan(first.base)
|
||||
finally:
|
||||
first.stop()
|
||||
|
||||
# Crash after the content moved but before the database caught up: the operation
|
||||
# is still reversible, which is exactly when rollback is defined.
|
||||
_crash_during_apply(seeded, plan, "moved")
|
||||
|
||||
restarted = Server(seeded).start()
|
||||
try:
|
||||
rolled = httpx.post(
|
||||
f"{restarted.base}/api/v1/rename-plans/{plan['id']}/rollback", timeout=TIMEOUT
|
||||
).json()
|
||||
assert rolled["rolled_back"] == 1 and rolled["state"] == "rolled_back"
|
||||
|
||||
assert (seeded.lib / "rome" / "a.jpg").exists()
|
||||
assert not (seeded.lib / APPROVED).exists()
|
||||
after = _inventory(restarted.base)
|
||||
assert set(after) == set(before)
|
||||
assert all(item["current_path"].endswith(".jpg") for item in after.values())
|
||||
assert _recovery(restarted.base)["blocks_mutation"] is False
|
||||
finally:
|
||||
restarted.stop()
|
||||
|
||||
|
||||
# ── durability ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_applied_state_survives_a_full_restart(tmp_path):
|
||||
seeded = seed_album(tmp_path)
|
||||
first = Server(seeded).start()
|
||||
try:
|
||||
approve_album(first.base, name=APPROVED)
|
||||
plan = _plan(first.base)
|
||||
_apply(first.base, plan, expected_checksum=plan["checksum"]).raise_for_status()
|
||||
expected = _inventory(first.base)
|
||||
finally:
|
||||
first.stop()
|
||||
|
||||
restarted = Server(seeded).start()
|
||||
try:
|
||||
assert _inventory(restarted.base) == expected
|
||||
after = _get_plan(restarted.base, plan["id"])
|
||||
assert after["state"] == "applied"
|
||||
assert after["checksum"] == plan["checksum"], "the plan's evidence is immutable"
|
||||
assert [op["journal_state"] for op in after["operations"]] == ["complete"]
|
||||
assert all(op["verified_at"] for op in after["operations"])
|
||||
assert _recovery(restarted.base) == {"blocks_mutation": False, "items": []}
|
||||
finally:
|
||||
restarted.stop()
|
||||
294
tests/e2e/test_renames_ui.py
Normal file
294
tests/e2e/test_renames_ui.py
Normal file
@@ -0,0 +1,294 @@
|
||||
"""Browser journeys for the rename view (US04-05).
|
||||
|
||||
Covers previewing every affected path, confirming with the server-issued token,
|
||||
applying with visible progress and terminal verification, refusing a stale
|
||||
confirmation, previewing a collision as blocking, and resolving an interrupted
|
||||
rename — including the check that an unresolved rename really does block unrelated
|
||||
mutations at the API, not just in the UI.
|
||||
|
||||
Renames really happen here: the assertions read the filesystem afterwards.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from playwright.sync_api import expect
|
||||
|
||||
from tests.e2e._pipeline_harness import Server, approve_album, seed_album, session_factory
|
||||
|
||||
# Part of the Phase D acceptance command (US04-06); mapped to US04-05 for traceability.
|
||||
pytestmark = pytest.mark.phase_d
|
||||
|
||||
TIMEOUT = 10
|
||||
APPROVED = "2019 Rome"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def server(tmp_path):
|
||||
seeded = seed_album(tmp_path)
|
||||
running = Server(seeded).start()
|
||||
running.seeded = seeded
|
||||
try:
|
||||
yield running
|
||||
finally:
|
||||
running.stop()
|
||||
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _approve(base, *, name=APPROVED):
|
||||
approve_album(base, name=name)
|
||||
|
||||
|
||||
def _build(base):
|
||||
response = httpx.post(f"{base}/api/v1/rename-plans", timeout=TIMEOUT)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
def _open_plan(page, server):
|
||||
"""Approve a name, open the view, and build the plan through the UI."""
|
||||
_approve(server.base)
|
||||
page.goto(f"{server.base}/app/#/renames")
|
||||
page.get_by_test_id("build-plan").click()
|
||||
page.get_by_test_id("operations").wait_for()
|
||||
|
||||
|
||||
def _interrupt(seeded, plan, *, state="moving", make_destination=False):
|
||||
"""Leave an operation in a non-terminal journal state, as a crash mid-apply would.
|
||||
|
||||
``make_destination`` also creates the destination folder, so source and
|
||||
destination both exist and the evidence becomes ambiguous (``manual``).
|
||||
"""
|
||||
from photo_pipeline.services.rename_journal import RenameJournal
|
||||
|
||||
with session_factory(seeded) as sf:
|
||||
journal = RenameJournal(sf)
|
||||
operation = journal.operations(plan["id"])[0]
|
||||
journal.begin(operation["id"], worker_id="crashed", fencing_token=1)
|
||||
if state != "moving":
|
||||
journal.transition(operation["id"], state, fencing_token=1)
|
||||
if make_destination:
|
||||
(seeded.lib / APPROVED).mkdir(exist_ok=True)
|
||||
|
||||
|
||||
# ── preview ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_preview_shows_every_path_operation_type_and_count(page, server):
|
||||
errors = []
|
||||
page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
|
||||
_open_plan(page, server)
|
||||
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("validated")
|
||||
expect(page.get_by_test_id("operation-count")).to_have_text("1 folder")
|
||||
expect(page.get_by_test_id("affected-assets")).to_have_text("2 photos")
|
||||
|
||||
row = page.get_by_test_id("operation-row").first
|
||||
# Whole paths, both sides — nothing about the move is left to be inferred.
|
||||
assert row.get_by_test_id("op-source").inner_text().endswith("/rome")
|
||||
assert row.get_by_test_id("op-destination").inner_text().endswith(f"/{APPROVED}")
|
||||
expect(row.get_by_test_id("op-type")).to_have_text("rename")
|
||||
expect(row.get_by_test_id("op-count")).to_have_text("2")
|
||||
expect(row.get_by_test_id("op-state")).to_have_text("planned")
|
||||
assert errors == [], f"console errors: {errors}"
|
||||
|
||||
|
||||
def test_confirmation_shows_the_server_token_and_the_cancellation_limit(page, server):
|
||||
_open_plan(page, server)
|
||||
plan = httpx.get(f"{server.base}/api/v1/rename-plans", timeout=TIMEOUT).json()["items"][0]
|
||||
|
||||
token = page.get_by_test_id("confirm-token").inner_text()
|
||||
assert plan["id"] in token and f"version {plan['version']}" in token
|
||||
assert plan["checksum"][:12] in token
|
||||
# The limit on cancellation is stated before the button, not discovered after.
|
||||
assert "cannot be cancelled" in page.get_by_test_id("cancel-note").inner_text()
|
||||
|
||||
|
||||
def test_a_collision_is_previewed_as_blocking_and_cannot_be_applied(page, server):
|
||||
_approve(server.base)
|
||||
# Someone else already owns the destination name, with content worth keeping.
|
||||
(server.seeded.lib / APPROVED).mkdir()
|
||||
(server.seeded.lib / APPROVED / "precious.jpg").write_bytes(b"do not lose me")
|
||||
|
||||
page.goto(f"{server.base}/app/#/renames")
|
||||
page.get_by_test_id("build-plan").click()
|
||||
page.get_by_test_id("operations").wait_for()
|
||||
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("invalid")
|
||||
expect(page.get_by_test_id("plan-blockers")).to_contain_text("destination_exists")
|
||||
issue = page.get_by_test_id("op-issue").first
|
||||
expect(issue).to_have_attribute("data-code", "destination_exists")
|
||||
assert issue.inner_text().startswith("blocks")
|
||||
expect(page.get_by_test_id("apply-plan")).to_be_disabled()
|
||||
|
||||
assert (server.seeded.lib / APPROVED / "precious.jpg").read_bytes() == b"do not lose me"
|
||||
assert (server.seeded.lib / "rome").is_dir()
|
||||
|
||||
|
||||
# ── apply ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_apply_reports_progress_terminal_verification_and_moves_the_folder(page, server):
|
||||
_open_plan(page, server)
|
||||
# The progress region is live before anything runs, so the status it announces
|
||||
# during the run reaches assistive technology.
|
||||
progress = page.get_by_test_id("apply-progress")
|
||||
expect(progress).to_have_attribute("aria-live", "polite")
|
||||
|
||||
page.get_by_test_id("apply-plan").click()
|
||||
|
||||
expect(page.get_by_test_id("apply-result")).to_contain_text("Applied 1, failed 0, skipped 0")
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
row = page.get_by_test_id("operation-row").first
|
||||
expect(row.get_by_test_id("op-state")).to_have_text("complete")
|
||||
expect(row.get_by_test_id("op-verified")).to_be_visible()
|
||||
|
||||
assert (server.seeded.lib / APPROVED / "a.jpg").exists()
|
||||
assert not (server.seeded.lib / "rome").exists()
|
||||
|
||||
|
||||
def test_a_stale_confirmation_is_refused_and_explained(page, server):
|
||||
_open_plan(page, server)
|
||||
# Another client applies first, which bumps the plan version this page is holding.
|
||||
plan = httpx.get(f"{server.base}/api/v1/rename-plans", timeout=TIMEOUT).json()["items"][0]
|
||||
httpx.post(
|
||||
f"{server.base}/api/v1/rename-plans/{plan['id']}/apply",
|
||||
json={"expected_version": plan["version"]},
|
||||
timeout=TIMEOUT,
|
||||
).raise_for_status()
|
||||
|
||||
page.get_by_test_id("apply-plan").click()
|
||||
|
||||
expect(page.get_by_test_id("conflict")).to_be_visible()
|
||||
# The refusal is not a second run: the page now shows the server's real state.
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
expect(page.get_by_test_id("apply-plan")).to_be_disabled()
|
||||
|
||||
|
||||
def test_the_applied_plan_survives_a_reload(page, server):
|
||||
_open_plan(page, server)
|
||||
page.get_by_test_id("apply-plan").click()
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
|
||||
page.reload()
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
expect(page.get_by_test_id("operation-row").first.get_by_test_id("op-state")).to_have_text(
|
||||
"complete"
|
||||
)
|
||||
# The result banner is tab state, not server state, so it correctly does not return.
|
||||
expect(page.get_by_test_id("apply-result")).to_have_count(0)
|
||||
|
||||
|
||||
def test_the_view_matches_the_journal_after_a_server_restart(page, server):
|
||||
_open_plan(page, server)
|
||||
page.get_by_test_id("apply-plan").click()
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
|
||||
server.stop()
|
||||
server.start() # same port, so the page reconnects to a genuinely fresh process
|
||||
|
||||
page.reload()
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
expect(page.get_by_test_id("operation-row").first.get_by_test_id("op-state")).to_have_text(
|
||||
"complete"
|
||||
)
|
||||
expect(page.get_by_test_id("op-verified")).to_be_visible()
|
||||
|
||||
|
||||
def test_apply_can_be_confirmed_from_the_keyboard(page, server):
|
||||
_open_plan(page, server)
|
||||
page.get_by_test_id("apply-plan").focus()
|
||||
page.keyboard.press("Enter")
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
assert (server.seeded.lib / APPROVED).is_dir()
|
||||
|
||||
|
||||
def test_a_finished_rename_offers_no_rollback(page, server):
|
||||
"""Rollback is recovery, not undo: once an operation is `complete` the journal
|
||||
treats it as terminal, so the view must not suggest it can be reversed."""
|
||||
_open_plan(page, server)
|
||||
page.get_by_test_id("apply-plan").click()
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("applied")
|
||||
expect(page.get_by_test_id("rollback-plan")).to_have_count(0)
|
||||
|
||||
|
||||
def test_rollback_returns_an_unfinished_move_to_its_source(page, server):
|
||||
_approve(server.base)
|
||||
plan = _build(server.base)
|
||||
# A crash after the move but before the database caught up: the content sits at
|
||||
# the destination and the operation is still reversible.
|
||||
_interrupt(server.seeded, plan, state="moving")
|
||||
(server.seeded.lib / "rome").rename(server.seeded.lib / APPROVED)
|
||||
with session_factory(server.seeded) as sf:
|
||||
from photo_pipeline.services.rename_journal import RenameJournal
|
||||
|
||||
journal = RenameJournal(sf)
|
||||
journal.transition(journal.operations(plan["id"])[0]["id"], "moved", fencing_token=1)
|
||||
|
||||
page.goto(f"{server.base}/app/#/renames")
|
||||
page.get_by_test_id("rollback-plan").click()
|
||||
|
||||
expect(page.get_by_test_id("plan-state")).to_have_text("rolled_back")
|
||||
assert (server.seeded.lib / "rome" / "a.jpg").exists()
|
||||
assert not (server.seeded.lib / APPROVED).exists()
|
||||
|
||||
|
||||
# ── recovery ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_an_interrupted_rename_is_shown_and_offers_the_safe_action(page, server):
|
||||
_approve(server.base)
|
||||
plan = _build(server.base)
|
||||
_interrupt(server.seeded, plan) # source intact, destination free → resumable
|
||||
|
||||
page.goto(f"{server.base}/app/#/renames")
|
||||
banner = page.get_by_test_id("recovery-banner")
|
||||
banner.wait_for()
|
||||
expect(page.get_by_test_id("recovery-item")).to_have_attribute(
|
||||
"data-classification", "resumable"
|
||||
)
|
||||
# While it is unresolved nothing else may mutate the library.
|
||||
expect(page.get_by_test_id("build-plan")).to_be_disabled()
|
||||
expect(page.get_by_test_id("apply-plan")).to_be_disabled()
|
||||
|
||||
page.get_by_test_id("resolve-recovery").click()
|
||||
expect(banner).to_have_count(0)
|
||||
expect(page.get_by_test_id("build-plan")).to_be_enabled()
|
||||
|
||||
|
||||
def test_ambiguous_evidence_offers_no_automatic_action(page, server):
|
||||
_approve(server.base)
|
||||
plan = _build(server.base)
|
||||
# Both paths exist: nothing can tell which side holds the truth.
|
||||
_interrupt(server.seeded, plan, make_destination=True)
|
||||
|
||||
page.goto(f"{server.base}/app/#/renames")
|
||||
page.get_by_test_id("recovery-banner").wait_for()
|
||||
expect(page.get_by_test_id("recovery-item")).to_have_attribute("data-classification", "manual")
|
||||
expect(page.get_by_test_id("manual-only")).to_be_visible()
|
||||
# No button is offered for work the server would refuse to do.
|
||||
expect(page.get_by_test_id("resolve-recovery")).to_have_count(0)
|
||||
|
||||
|
||||
def test_an_unresolved_rename_blocks_unrelated_name_changes(page, server):
|
||||
_approve(server.base)
|
||||
plan = _build(server.base)
|
||||
_interrupt(server.seeded, plan, make_destination=True)
|
||||
|
||||
page.goto(f"{server.base}/app/#/albums?album=rome")
|
||||
expect(page.get_by_test_id("rename-blocked")).to_be_visible()
|
||||
expect(page.get_by_test_id("approve")).to_be_disabled()
|
||||
expect(page.get_by_test_id("generate-proposals")).to_be_disabled()
|
||||
|
||||
# The disabled buttons are not the only guard: the API refuses too.
|
||||
current = httpx.get(f"{server.base}/api/v1/albums/proposals/rome", timeout=TIMEOUT).json()
|
||||
refused = httpx.post(
|
||||
f"{server.base}/api/v1/albums/proposals/rome/approve",
|
||||
json={"expected_version": current["version"]},
|
||||
timeout=TIMEOUT,
|
||||
)
|
||||
assert refused.status_code == 409
|
||||
assert refused.json()["error"]["code"] == "rename_recovery_required"
|
||||
@@ -12,12 +12,19 @@ from pathlib import Path
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
MAP = json.loads((REPO / "tests" / "story_traceability.json").read_text())["stories"]
|
||||
PHASE_A_STORIES = {f"US01-0{n}" for n in range(1, 8)}
|
||||
PHASE_D_STORIES = {f"US04-0{n}" for n in range(1, 7)}
|
||||
|
||||
|
||||
def test_all_phase_a_stories_are_mapped():
|
||||
assert PHASE_A_STORIES <= set(MAP)
|
||||
|
||||
|
||||
def test_all_phase_d_stories_are_mapped():
|
||||
"""US04-06 acceptance: every guarded-rename story, plan through browser, is tied
|
||||
to automated tests — renaming is the first thing that mutates the real library."""
|
||||
assert PHASE_D_STORIES <= set(MAP)
|
||||
|
||||
|
||||
def test_every_mapped_test_file_exists_and_is_nonempty():
|
||||
for story, files in MAP.items():
|
||||
assert files, f"{story} maps to no tests"
|
||||
|
||||
304
tests/integration/test_rename_apply.py
Normal file
304
tests/integration/test_rename_apply.py
Normal file
@@ -0,0 +1,304 @@
|
||||
"""Applying and verifying guarded renames (US04-03).
|
||||
|
||||
Everything here runs against a real temporary library: files really move, and the
|
||||
tests assert the bytes survived, the stable asset IDs kept their identity, the path
|
||||
history was recorded, and nothing unexpected was ever overwritten.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from photo_pipeline.api.app import create_app
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import AlbumProposal, Asset, AssetPath
|
||||
from photo_pipeline.services import hashing
|
||||
from photo_pipeline.services.rename_apply import (
|
||||
ApplyConflict,
|
||||
ApplyError,
|
||||
RenameApplyService,
|
||||
)
|
||||
from photo_pipeline.services.rename_journal import JournalState, RenameJournal
|
||||
from photo_pipeline.services.renames import RenameService
|
||||
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _factory(tmp_path):
|
||||
(tmp_path / "data").mkdir()
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir()
|
||||
config = Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
||||
}
|
||||
)
|
||||
run_migrations(config.database_url)
|
||||
return config, create_session_factory(create_db_engine(config.database_url)), lib
|
||||
|
||||
|
||||
def _album(sf, lib, album, *, approved_name, names=("a.jpg", "b.jpg")):
|
||||
folder = lib / album
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
with sf() as session:
|
||||
for index, name in enumerate(names):
|
||||
path = folder / name
|
||||
path.write_bytes(f"content-{album}-{index}".encode())
|
||||
session.add(
|
||||
Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path=str(path),
|
||||
current_path=str(path),
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=path.stat().st_size,
|
||||
current_sha256=hashing.sha256_file(path),
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
AlbumProposal(
|
||||
id=str(uuid.uuid4()),
|
||||
album=album,
|
||||
proposed_name=approved_name,
|
||||
final_name=approved_name,
|
||||
status="approved",
|
||||
version=2,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
return folder
|
||||
|
||||
|
||||
def _plan(sf, lib):
|
||||
return RenameService(sf, library_roots=(lib,)).build_plan()
|
||||
|
||||
|
||||
def _apply(sf, lib, plan, **kwargs):
|
||||
service = RenameApplyService(sf, library_roots=(lib,))
|
||||
return service.apply(plan["id"], expected_version=plan["version"], **kwargs)
|
||||
|
||||
|
||||
def _paths(sf):
|
||||
with sf() as session:
|
||||
return {a.id: a.current_path for a in session.scalars(select(Asset))}
|
||||
|
||||
|
||||
def _bytes_under(root):
|
||||
return {
|
||||
str(p.relative_to(root)): p.read_bytes() for p in sorted(root.rglob("*")) if p.is_file()
|
||||
}
|
||||
|
||||
|
||||
# ── normal apply ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_apply_moves_the_folder_and_preserves_bytes_and_identity(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
before_ids = set(_paths(sf))
|
||||
before_bytes = _bytes_under(lib)
|
||||
|
||||
plan = _plan(sf, lib)
|
||||
result = _apply(sf, lib, plan)
|
||||
|
||||
assert result["applied"] == 1 and result["failed"] == 0
|
||||
assert result["state"] == "applied"
|
||||
assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists()
|
||||
# Same asset IDs, new paths, identical bytes.
|
||||
after = _paths(sf)
|
||||
assert set(after) == before_ids, "renaming must not change asset identity"
|
||||
assert all("2019 Rome" in path for path in after.values())
|
||||
assert _bytes_under(lib) == {
|
||||
key.replace("rome/", "2019 Rome/"): value for key, value in before_bytes.items()
|
||||
}
|
||||
|
||||
|
||||
def test_apply_records_the_path_history(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
|
||||
plan = _plan(sf, lib)
|
||||
_apply(sf, lib, plan)
|
||||
|
||||
with sf() as session:
|
||||
rows = list(session.scalars(select(AssetPath).order_by(AssetPath.valid_from)))
|
||||
open_rows = [r for r in rows if r.valid_until is None]
|
||||
assert len(open_rows) == 1 and "2019 Rome" in open_rows[0].path
|
||||
assert open_rows[0].reason == "rename"
|
||||
|
||||
|
||||
def test_every_operation_reaches_complete_in_the_journal(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
|
||||
plan = _plan(sf, lib)
|
||||
_apply(sf, lib, plan)
|
||||
|
||||
states = {op["journal_state"] for op in RenameJournal(sf).operations(plan["id"])}
|
||||
assert states == {JournalState.COMPLETE}
|
||||
|
||||
|
||||
# ── confirmation and leases ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_apply_rejects_a_stale_version_without_touching_disk(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
before = _bytes_under(lib)
|
||||
|
||||
service = RenameApplyService(sf, library_roots=(lib,))
|
||||
with pytest.raises(ApplyConflict):
|
||||
service.apply(plan["id"], expected_version=plan["version"] + 5)
|
||||
assert _bytes_under(lib) == before and (lib / "rome").exists()
|
||||
|
||||
|
||||
def test_apply_rejects_a_changed_checksum(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
service = RenameApplyService(sf, library_roots=(lib,))
|
||||
with pytest.raises(ApplyConflict):
|
||||
service.apply(
|
||||
plan["id"], expected_version=plan["version"], expected_checksum="not-the-checksum"
|
||||
)
|
||||
|
||||
|
||||
def test_an_invalid_plan_can_never_be_applied(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
(lib / "2019 Rome").mkdir() # destination occupied → plan is invalid
|
||||
plan = _plan(sf, lib)
|
||||
assert plan["state"] == "invalid"
|
||||
|
||||
service = RenameApplyService(sf, library_roots=(lib,))
|
||||
with pytest.raises(ApplyError):
|
||||
service.apply(plan["id"], expected_version=plan["version"])
|
||||
|
||||
|
||||
def test_repeated_apply_is_a_no_op(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_apply(sf, lib, plan)
|
||||
after_first = _bytes_under(lib)
|
||||
|
||||
service = RenameApplyService(sf, library_roots=(lib,))
|
||||
fresh_version = RenameService(sf, library_roots=(lib,)).get(plan["id"])["version"]
|
||||
second = service.apply(plan["id"], expected_version=fresh_version)
|
||||
assert second["applied"] == 0 and second["skipped"] == 1
|
||||
assert _bytes_under(lib) == after_first
|
||||
|
||||
|
||||
# ── preconditions rechecked at mutation time ─────────────────────────────────
|
||||
|
||||
|
||||
def test_a_source_changed_after_planning_is_refused(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
folder = _album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
# The file changes between preview and confirmation.
|
||||
(folder / "a.jpg").write_bytes(b"tampered")
|
||||
|
||||
result = _apply(sf, lib, plan)
|
||||
assert result["failed"] == 1 and result["applied"] == 0
|
||||
assert (lib / "rome").exists(), "a failed precondition must leave the source alone"
|
||||
operation = RenameJournal(sf).operations(plan["id"])[0]
|
||||
assert operation["journal_state"] == JournalState.FAILED
|
||||
assert operation["error_code"] == "source_changed"
|
||||
|
||||
|
||||
def test_an_occupant_appearing_after_planning_is_never_overwritten(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
# Someone creates the destination between validation and apply.
|
||||
(lib / "2019 Rome").mkdir()
|
||||
(lib / "2019 Rome" / "precious.jpg").write_bytes(b"do not lose me")
|
||||
|
||||
result = _apply(sf, lib, plan)
|
||||
assert result["failed"] == 1
|
||||
assert (lib / "2019 Rome" / "precious.jpg").read_bytes() == b"do not lose me"
|
||||
assert (lib / "rome").exists()
|
||||
|
||||
|
||||
# ── staged procedures ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_case_only_rename_uses_the_staged_procedure(tmp_path):
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="Rome", names=("a.jpg",))
|
||||
plan = _plan(sf, lib)
|
||||
assert plan["operations"][0]["case_only"] is True
|
||||
|
||||
result = _apply(sf, lib, plan)
|
||||
assert result["applied"] == 1
|
||||
# The directory entry now carries the new casing, and no staging dir survives.
|
||||
entries = {p.name for p in lib.iterdir()}
|
||||
assert "Rome" in entries
|
||||
assert not any(name.startswith(".rename-") for name in entries)
|
||||
assert list(_paths(sf).values())[0].endswith("Rome/a.jpg")
|
||||
|
||||
|
||||
def test_cross_filesystem_move_copies_verifies_then_deletes(tmp_path):
|
||||
"""The cross-filesystem branch is driven directly: a real second filesystem is
|
||||
not portable in CI, but the copy-verify-delete procedure is what matters."""
|
||||
_, sf, lib = _factory(tmp_path)
|
||||
source = lib / "rome"
|
||||
source.mkdir()
|
||||
(source / "a.jpg").write_bytes(b"payload")
|
||||
destination = lib / "2019 Rome"
|
||||
|
||||
RenameApplyService(sf, library_roots=(lib,))._copy_verify_delete(source, destination)
|
||||
|
||||
assert not source.exists()
|
||||
assert (destination / "a.jpg").read_bytes() == b"payload"
|
||||
assert not any(p.name.startswith(".rename-") for p in lib.iterdir())
|
||||
|
||||
|
||||
# ── restart durability ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_applied_state_survives_a_restart(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
|
||||
plan = _plan(sf, lib)
|
||||
_apply(sf, lib, plan)
|
||||
expected = _paths(sf)
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
assert _paths(reopened) == expected
|
||||
states = {op["journal_state"] for op in RenameJournal(reopened).operations(plan["id"])}
|
||||
assert states == {JournalState.COMPLETE}
|
||||
|
||||
|
||||
# ── API ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_api_apply_requires_the_current_version(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome", names=("a.jpg",))
|
||||
|
||||
with TestClient(create_app(config)) as client:
|
||||
plan = client.post("/api/v1/rename-plans").json()
|
||||
|
||||
stale = client.post(
|
||||
f"/api/v1/rename-plans/{plan['id']}/apply",
|
||||
json={"expected_version": plan["version"] + 9},
|
||||
)
|
||||
assert stale.status_code == 409
|
||||
assert stale.json()["error"]["code"] == "version_conflict"
|
||||
assert (lib / "rome").exists()
|
||||
|
||||
applied = client.post(
|
||||
f"/api/v1/rename-plans/{plan['id']}/apply",
|
||||
json={"expected_version": plan["version"], "expected_checksum": plan["checksum"]},
|
||||
).json()
|
||||
assert applied["applied"] == 1 and applied["state"] == "applied"
|
||||
assert (lib / "2019 Rome").is_dir()
|
||||
|
||||
recovery = client.get("/api/v1/rename-recovery").json()
|
||||
assert recovery["blocks_mutation"] is False and recovery["items"] == []
|
||||
354
tests/integration/test_rename_recovery.py
Normal file
354
tests/integration/test_rename_recovery.py
Normal file
@@ -0,0 +1,354 @@
|
||||
"""Recovering or rolling back interrupted renames (US04-04).
|
||||
|
||||
The crash tests are real: a child process applies a plan and is killed by the
|
||||
``PHOTO_PIPELINE_FAULT_AFTER`` barrier at each persisted journal transition in turn.
|
||||
The parent then reopens the database, classifies the wreckage from journal + disk
|
||||
evidence, and asserts recovery converges without losing or overwriting anything.
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import AlbumProposal, Asset
|
||||
from photo_pipeline.services import hashing
|
||||
from photo_pipeline.services.rename_apply import ApplyError, RenameApplyService
|
||||
from photo_pipeline.services.rename_journal import (
|
||||
MANUAL,
|
||||
RESUMABLE,
|
||||
ROLLBACK_SAFE,
|
||||
JournalState,
|
||||
)
|
||||
from photo_pipeline.services.renames import RenameService
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
# A child process that applies the plan and dies at the configured barrier.
|
||||
APPLY_SCRIPT = """
|
||||
import sys
|
||||
sys.path.insert(0, {repo!r})
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory
|
||||
from photo_pipeline.services.rename_apply import RenameApplyService
|
||||
|
||||
db_url, lib, plan_id, version = sys.argv[1], sys.argv[2], sys.argv[3], int(sys.argv[4])
|
||||
sf = create_session_factory(create_db_engine(db_url))
|
||||
RenameApplyService(sf, library_roots=(lib,)).apply(plan_id, expected_version=version)
|
||||
"""
|
||||
|
||||
|
||||
def _factory(tmp_path):
|
||||
(tmp_path / "data").mkdir()
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir()
|
||||
config = Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
||||
}
|
||||
)
|
||||
run_migrations(config.database_url)
|
||||
return config, create_session_factory(create_db_engine(config.database_url)), lib
|
||||
|
||||
|
||||
def _album(sf, lib, album, *, approved_name, names=("a.jpg",)):
|
||||
folder = lib / album
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
with sf() as session:
|
||||
for index, name in enumerate(names):
|
||||
path = folder / name
|
||||
path.write_bytes(f"content-{album}-{index}".encode())
|
||||
session.add(
|
||||
Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path=str(path),
|
||||
current_path=str(path),
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=path.stat().st_size,
|
||||
current_sha256=hashing.sha256_file(path),
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
AlbumProposal(
|
||||
id=str(uuid.uuid4()),
|
||||
album=album,
|
||||
proposed_name=approved_name,
|
||||
final_name=approved_name,
|
||||
status="approved",
|
||||
version=2,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
return folder
|
||||
|
||||
|
||||
def _plan(sf, lib):
|
||||
return RenameService(sf, library_roots=(lib,)).build_plan()
|
||||
|
||||
|
||||
def _crash_during_apply(config, lib, plan, barrier, tmp_path):
|
||||
"""Run apply in a child that dies right after ``barrier`` is persisted."""
|
||||
script = tmp_path / f"apply_{barrier}.py"
|
||||
script.write_text(APPLY_SCRIPT.format(repo=str(REPO)))
|
||||
env = dict(os.environ)
|
||||
env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
str(script),
|
||||
config.database_url,
|
||||
str(lib),
|
||||
plan["id"],
|
||||
str(plan["version"]),
|
||||
],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
)
|
||||
assert result.returncode == -9 or result.returncode == 9, (
|
||||
f"child should have been killed at {barrier}, got {result.returncode}: "
|
||||
f"{result.stderr.decode(errors='replace')[-400:]}"
|
||||
)
|
||||
|
||||
|
||||
def _assets(sf):
|
||||
with sf() as session:
|
||||
return {a.id: a.current_path for a in session.scalars(select(Asset))}
|
||||
|
||||
|
||||
def _files(lib):
|
||||
return {p.read_bytes() for p in lib.rglob("*") if p.is_file() and ".rename-" not in str(p)}
|
||||
|
||||
|
||||
# ── fault injection at every persisted transition ────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"barrier",
|
||||
[
|
||||
JournalState.MOVING,
|
||||
JournalState.MOVED,
|
||||
JournalState.DATABASE_UPDATED,
|
||||
JournalState.VERIFIED,
|
||||
],
|
||||
)
|
||||
def test_crash_at_each_transition_recovers_without_losing_content(barrier, tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
content_before = _files(lib)
|
||||
|
||||
_crash_during_apply(config, lib, plan, barrier, tmp_path)
|
||||
|
||||
# Fresh process: the journal plus the disk decide what happened.
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
verdicts = service.journal.classify_all()
|
||||
assert verdicts, f"a crash at {barrier} must leave recoverable work"
|
||||
assert all(v["classification"] in {RESUMABLE, ROLLBACK_SAFE} for v in verdicts), verdicts
|
||||
|
||||
service.recover()
|
||||
|
||||
# No content was lost anywhere in the library.
|
||||
assert _files(lib) == content_before
|
||||
# And nothing is left half-done.
|
||||
assert service.journal.blocks_mutation() is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"barrier", [JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED]
|
||||
)
|
||||
def test_recovery_is_idempotent_across_repeated_restarts(barrier, tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, barrier, tmp_path)
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
service.recover()
|
||||
settled_paths = _assets(reopened)
|
||||
settled_files = _files(lib)
|
||||
|
||||
# Repeated recovery passes converge — the second and third change nothing.
|
||||
for _ in range(2):
|
||||
service.recover()
|
||||
assert _assets(reopened) == settled_paths
|
||||
assert _files(lib) == settled_files
|
||||
|
||||
|
||||
def test_crash_before_the_move_leaves_the_source_intact_and_resumable(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
verdict = service.journal.classify_all()[0]
|
||||
assert verdict["classification"] == RESUMABLE
|
||||
assert (lib / "rome").exists() and not (lib / "2019 Rome").exists()
|
||||
|
||||
service.recover()
|
||||
# Resumable work returns to `planned`, so a fresh apply can finish the job.
|
||||
operation = service.journal.operations(plan["id"])[0]
|
||||
assert operation["journal_state"] == JournalState.PLANNED
|
||||
|
||||
fresh = RenameService(reopened, library_roots=(lib,)).get(plan["id"])
|
||||
service.apply(plan["id"], expected_version=fresh["version"])
|
||||
assert (lib / "2019 Rome").is_dir() and not (lib / "rome").exists()
|
||||
|
||||
|
||||
def test_crash_after_the_move_completes_the_bookkeeping(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
assert service.journal.classify_all()[0]["classification"] == ROLLBACK_SAFE
|
||||
|
||||
service.recover()
|
||||
operation = service.journal.operations(plan["id"])[0]
|
||||
assert operation["journal_state"] == JournalState.COMPLETE
|
||||
# The database caught up with the filesystem.
|
||||
assert all("2019 Rome" in path for path in _assets(reopened).values())
|
||||
|
||||
|
||||
# ── ambiguous evidence ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_an_unexpected_destination_occupant_forces_manual_recovery(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)
|
||||
|
||||
# Someone puts something at the destination while the app was down.
|
||||
(lib / "2019 Rome").mkdir()
|
||||
(lib / "2019 Rome" / "stranger.jpg").write_bytes(b"not ours")
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
verdict = service.journal.classify_all()[0]
|
||||
assert verdict["classification"] == MANUAL
|
||||
assert verdict["reason"]
|
||||
|
||||
result = service.recover()
|
||||
assert result["manual"] == 1
|
||||
# Nothing was touched, and the ambiguity keeps blocking other mutations.
|
||||
assert (lib / "2019 Rome" / "stranger.jpg").read_bytes() == b"not ours"
|
||||
assert (lib / "rome").exists()
|
||||
assert service.journal.blocks_mutation() is True
|
||||
|
||||
|
||||
def test_manual_work_blocks_a_different_plan_from_applying(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVING, tmp_path)
|
||||
(lib / "2019 Rome").mkdir() # ambiguous
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
# The user parks the stuck album (un-approves it) and tries a different one, so
|
||||
# the new plan is valid on its own — only the unresolved journal entry stands in
|
||||
# the way.
|
||||
with reopened() as session:
|
||||
stuck = session.scalar(select(AlbumProposal).where(AlbumProposal.album == "rome"))
|
||||
stuck.status = "edited"
|
||||
session.commit()
|
||||
_album(reopened, lib, "paris", approved_name="2020 Paris")
|
||||
other = RenameService(reopened, library_roots=(lib,)).build_plan()
|
||||
assert other["state"] == "validated", "the unrelated plan itself must be applyable"
|
||||
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
with pytest.raises(ApplyError, match="another rename is unresolved"):
|
||||
service.apply(other["id"], expected_version=other["version"])
|
||||
# Nothing moved for the unrelated album either.
|
||||
assert (lib / "paris").exists() and not (lib / "2020 Paris").exists()
|
||||
|
||||
|
||||
# ── rollback ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_rollback_returns_the_content_to_its_source(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
before = _files(lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
operation = service.journal.operations(plan["id"])[0]
|
||||
|
||||
service.rollback_operation(operation["id"])
|
||||
|
||||
assert (lib / "rome").is_dir() and not (lib / "2019 Rome").exists()
|
||||
assert _files(lib) == before
|
||||
assert service.journal.get(operation["id"])["journal_state"] == JournalState.ROLLED_BACK
|
||||
assert all("rome/" in path for path in _assets(reopened).values())
|
||||
|
||||
|
||||
def test_rollback_refuses_when_the_source_is_occupied(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
|
||||
|
||||
# Something new appears at the old location while the app was down.
|
||||
(lib / "rome").mkdir()
|
||||
(lib / "rome" / "new.jpg").write_bytes(b"newcomer")
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
operation = service.journal.operations(plan["id"])[0]
|
||||
|
||||
with pytest.raises(ApplyError, match="occupied"):
|
||||
service.rollback_operation(operation["id"])
|
||||
assert (lib / "rome" / "new.jpg").read_bytes() == b"newcomer"
|
||||
|
||||
|
||||
def test_rollback_refuses_content_whose_bytes_changed(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
|
||||
|
||||
# The moved content is edited before rollback is attempted.
|
||||
(lib / "2019 Rome" / "a.jpg").write_bytes(b"edited after the move")
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
operation = service.journal.operations(plan["id"])[0]
|
||||
|
||||
with pytest.raises(ApplyError, match="manual recovery"):
|
||||
service.rollback_operation(operation["id"])
|
||||
assert (lib / "2019 Rome" / "a.jpg").read_bytes() == b"edited after the move"
|
||||
|
||||
|
||||
def test_rollback_is_idempotent(tmp_path):
|
||||
config, sf, lib = _factory(tmp_path)
|
||||
_album(sf, lib, "rome", approved_name="2019 Rome")
|
||||
plan = _plan(sf, lib)
|
||||
_crash_during_apply(config, lib, plan, JournalState.MOVED, tmp_path)
|
||||
|
||||
reopened = create_session_factory(create_db_engine(config.database_url))
|
||||
service = RenameApplyService(reopened, library_roots=(lib,))
|
||||
operation = service.journal.operations(plan["id"])[0]
|
||||
service.rollback_operation(operation["id"])
|
||||
settled = _files(lib)
|
||||
|
||||
# A second rollback has nothing left to do and refuses rather than moving again.
|
||||
with pytest.raises(ApplyError):
|
||||
service.rollback_operation(operation["id"])
|
||||
assert _files(lib) == settled
|
||||
@@ -91,6 +91,18 @@
|
||||
"US04-02": [
|
||||
"tests/unit/test_rename_journal_states.py",
|
||||
"tests/integration/test_rename_journal.py"
|
||||
],
|
||||
"US04-03": [
|
||||
"tests/integration/test_rename_apply.py"
|
||||
],
|
||||
"US04-04": [
|
||||
"tests/integration/test_rename_recovery.py"
|
||||
],
|
||||
"US04-05": [
|
||||
"tests/e2e/test_renames_ui.py"
|
||||
],
|
||||
"US04-06": [
|
||||
"tests/e2e/test_phase_d_pipeline.py"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user