- Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathundo.py
More file actions
Latest commit
98 lines (89 loc) · 3.45 KB
/
Copy pathundo.py
File metadata and controls
98 lines (89 loc) · 3.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
"""Preconditioned reversal of completed rename-run manifests."""
from __future__ importannotations
importos
fromexecutionimportapply_plan
fromrename_planimportPlanIssue, RenameOperation, RenamePlan, create_plan, validate_plan
fromrun_manifestimportRunManifest, file_sha256, read_manifest
defcreate_undo_plan(manifest: RunManifest) ->tuple[RenamePlan, tuple[PlanIssue, ...]]:
"""Build a reversible plan only when completed targets still match their hashes."""
issues: list[PlanIssue] = []
operations: list[RenameOperation] = []
ifmanifest.action!="apply"ormanifest.state!="applied":
issue=PlanIssue(
"",
"",
"",
"not_applied_manifest",
"only completed apply manifests can be undone",
)
returncreate_plan(()), (issue,)
foroperationinmanifest.operations:
ifoperation.result=="noop":
continue
reverse=RenameOperation(operation.scene_id, operation.destination, operation.source)
ifoperation.result!="applied":
issues.append(
PlanIssue(
reverse.scene_id,
reverse.source,
reverse.destination,
"operation_not_applied",
"manifest does not record this operation as applied",
)
)
continue
ifnotoperation.sha256:
issues.append(
PlanIssue(
reverse.scene_id,
reverse.source,
reverse.destination,
"missing_fingerprint",
"manifest has no post-apply SHA-256 for this operation",
)
)
continue
ifnotos.path.isfile(reverse.source):
issues.append(
PlanIssue(
reverse.scene_id,
reverse.source,
reverse.destination,
"missing_applied_destination",
"applied destination no longer exists as a regular file",
)
)
continue
iffile_sha256(reverse.source) !=operation.sha256:
issues.append(
PlanIssue(
reverse.scene_id,
reverse.source,
reverse.destination,
"changed_applied_destination",
"applied destination SHA-256 differs from the completed run",
)
)
continue
ifos.path.exists(reverse.destination):
issues.append(
PlanIssue(
reverse.scene_id,
reverse.source,
reverse.destination,
"occupied_original_source",
"original source path is occupied and will not be replaced",
)
)
continue
operations.append(reverse)
plan=create_plan(operations)
issues.extend(validate_plan(plan))
returnplan, tuple(issues)
defundo_manifest(manifest_path: str) ->tuple[RunManifest, RenamePlan, tuple[PlanIssue, ...]]:
"""Apply a validated reverse plan from one completed manifest, or return blockers."""
manifest=read_manifest(manifest_path)
plan, issues=create_undo_plan(manifest)
ifissues:
returnmanifest, plan, issues
returnmanifest, plan, apply_plan(plan)