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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
from __future__ import annotations
import json
import os
from helpers import World
import txn as txnmod
class TxnWorld(World):
def setUp(self) -> None:
super().setUp()
self.cdn_server = self.start_cdn()
def crash_at(self, state: str, entry: str = "a-main") -> dict:
os.environ[txnmod.CRASH_ENV] = state
try:
with self.assertRaises(txnmod.InjectedCrash):
self.run_pipeline("--only", entry, "--upload")
finally:
del os.environ[txnmod.CRASH_ENV]
return self.txn_data()
def txn_dirs(self) -> list:
root = self.cache / "txn"
return sorted(p for p in root.iterdir() if p.is_dir()) if root.exists() else []
def txn_data(self, index: int = -1) -> dict:
return json.loads((self.txn_dirs()[index] / "txn.json").read_text(encoding="utf-8"))
def a_url(self) -> str:
return "https://img.example.org/i/up1.webp"
class CrashRecoveryTest(TxnWorld):
def test_prepared_becomes_abandoned(self) -> None:
before = self.doc("a.md")
self.assertEqual(self.crash_at("prepared")["state"], "prepared")
code, out, _ = self.recover()
self.assertEqual(code, 0, out)
self.assertEqual(self.txn_data()["state"], "abandoned")
self.assertEqual(self.cdn_server.requests, [])
self.assertEqual(self.doc("a.md"), before)
self.assertEqual(self.read_lock(), {})
# The entry is still stale and the next run picks it up.
code, _, _ = self.run_pipeline("--only", "a-main", "--upload")
self.assertEqual(code, 0)
self.assertEqual(self.statuses()["a-main"], "updated")
def test_uploading_becomes_unknown_then_resolve_url(self) -> None:
data = self.crash_at("uploading")
code, out, _ = self.recover()
self.assertEqual(code, 3)
self.assertEqual(self.txn_data()["state"], "upload-unknown")
# Never retried automatically.
self.assertEqual(self.recover()[0], 3)
self.assertEqual(self.cdn_server.requests, [])
manual_url = "https://img.example.org/i/found-in-admin.webp"
code, out, err = self.recover("--resolve", data["txn_id"], "--url", manual_url)
self.assertEqual(code, 0, out + err)
final = self.txn_data()
self.assertEqual(final["state"], "committed")
self.assertEqual(final["upload"]["del"], txnmod.DEL_UNKNOWN)
self.assertIn(manual_url, self.doc("a.md").decode())
self.assertEqual(self.read_lock()["a-main"]["url"], manual_url)
self.assertTrue((self.cache / "baseline" / "a-main.png").exists())
def test_uploading_resolve_retry_upload(self) -> None:
data = self.crash_at("uploading")
self.recover()
code, out, err = self.recover("--resolve", data["txn_id"], "--retry-upload")
self.assertEqual(code, 0, out + err)
self.assertEqual(len(self.cdn_server.requests), 1)
self.assertEqual(self.txn_data()["state"], "committed")
self.assertIn(self.a_url(), self.doc("a.md").decode())
def test_resolve_validation(self) -> None:
data = self.crash_at("uploaded")
self.assertEqual(self.recover("--resolve", data["txn_id"], "--url", "https://x/y")[0], 2)
self.assertEqual(self.recover("--resolve", "nope", "--url", "https://x/y")[0], 2)
self.assertEqual(self.recover("--resolve", data["txn_id"])[0], 2)
def test_uploaded_completes_doc_lock_baseline(self) -> None:
self.crash_at("uploaded")
self.assertNotIn(self.a_url(), self.doc("a.md").decode())
code, out, _ = self.recover()
self.assertEqual(code, 0, out)
self.assertEqual(self.txn_data()["state"], "committed")
self.assertIn(self.a_url(), self.doc("a.md").decode())
self.assertEqual(self.read_lock()["a-main"]["url"], self.a_url())
self.assertTrue((self.cache / "baseline" / "a-main.png").exists())
self.assertEqual(len(self.cdn_server.requests), 1)
def test_doc_applied_completes_lock_only(self) -> None:
self.crash_at("doc-applied")
doc_after_crash = self.doc("a.md")
self.assertIn(self.a_url(), doc_after_crash.decode())
self.assertEqual(self.read_lock(), {})
self.recover()
self.assertEqual(self.doc("a.md"), doc_after_crash)
self.assertEqual(self.read_lock()["a-main"]["url"], self.a_url())
self.assertEqual(self.txn_data()["state"], "committed")
def test_lock_applied_completes_baseline_only(self) -> None:
self.crash_at("lock-applied")
lock = self.read_lock()
self.assertFalse((self.cache / "baseline" / "a-main.png").exists())
self.recover()
self.assertEqual(self.read_lock(), lock)
self.assertTrue((self.cache / "baseline" / "a-main.png").exists())
self.assertEqual(self.txn_data()["state"], "committed")
def test_tampered_webp_is_artifact_corrupt(self) -> None:
self.crash_at("uploaded")
before_doc, before_lock = self.doc("a.md"), self.read_lock()
webp = self.txn_dirs()[-1] / "final.webp"
webp.write_bytes(webp.read_bytes() + b"x")
code, _, _ = self.recover()
self.assertEqual(code, 3)
self.assertEqual(self.txn_data()["state"], "artifact-corrupt")
self.assertEqual(self.doc("a.md"), before_doc)
self.assertEqual(self.read_lock(), before_lock)
def test_lock_changed_after_prepare_is_conflict(self) -> None:
self.crash_at("uploaded")
foreign = {"a-main": {"url": "https://other/x.webp", "note": "newer data"}}
self.lock.write_text(json.dumps(foreign), encoding="utf-8")
before_doc = self.doc("a.md")
code, out, _ = self.recover()
self.assertEqual(code, 3)
self.assertEqual(self.txn_data()["state"], "conflict")
self.assertEqual(self.read_lock(), foreign)
self.assertEqual(self.doc("a.md"), before_doc)
self.assertIn("del:", out)
class DocEditTest(TxnWorld):
def test_concurrent_unrelated_edit_is_kept(self) -> None:
self.crash_at("uploaded")
path = self.manual / "src/content/docs/guides/a.md"
path.write_bytes(self.doc("a.md").replace(b"Some steps.", b"Some edited steps."))
self.assertEqual(self.recover()[0], 0)
text = self.doc("a.md").decode()
self.assertIn("Some edited steps.", text)
self.assertIn(f"[img-1]: {self.a_url()}", text)
def test_edit_to_same_definition_is_conflict(self) -> None:
self.crash_at("uploaded")
path = self.manual / "src/content/docs/guides/a.md"
edited = self.doc("a.md").replace(b"https://cdn.example.org/i/a1.webp", b"https://elsewhere/x.webp")
path.write_bytes(edited)
self.assertEqual(self.recover()[0], 3)
self.assertEqual(self.txn_data()["state"], "conflict")
self.assertEqual(self.doc("a.md"), edited)
self.assertEqual(self.read_lock(), {})
def test_crlf_doc_keeps_crlf(self) -> None:
code, _, _ = self.run_pipeline("--only", "b-main", "--upload")
self.assertEqual(code, 0)
raw = self.doc("b.md")
self.assertIn(b"https://img.example.org/i/up1.webp\r\n", raw)
self.assertEqual(raw.count(b"\n"), raw.count(b"\r\n"))
def test_del_never_in_committed_files(self) -> None:
code, _, _ = self.run_pipeline("--upload")
self.assertEqual(code, 0)
for path in (self.lock, self.manual / "src/content/docs/guides/a.md",
self.manual / "src/content/docs/guides/b.md", self.cache / "last-run.json"):
text = path.read_text(encoding="utf-8")
self.assertNotIn("/del/", text, path)
self.assertNotIn('"del"', text, path)
# It is kept in the ignored transaction cache.
self.assertIn("/del/secret", (self.txn_dirs()[0] / "txn.json").read_text(encoding="utf-8"))
def test_conflict_stops_remaining_transactions(self) -> None:
import apply
original = apply.apply_doc
def conflicting(*args, **kwargs):
raise apply.Conflict("simulated concurrent edit")
txnmod.apply_doc = conflicting
try:
code, out, _ = self.run_pipeline("--upload")
finally:
txnmod.apply_doc = original
self.assertEqual(code, 3)
st = self.statuses()
self.assertEqual(st["a-main"], "conflict")
self.assertEqual(st["a-dialog"], "failed:stopped-after-conflict")
self.assertEqual(len(self.cdn_server.requests), 1)
self.assertIn("del: https://img.example.org/del/secret1", out)
|
