| 1 | import json |
| 2 | import tempfile |
| 3 | import unittest |
| 4 | from pathlib import Path |
| 5 | |
| 6 | from scripts import publish_safety |
| 7 | |
| 8 | |
| 9 | class PublishSafetyTests(unittest.TestCase): |
| 10 | def write_manifest(self, root: Path) -> Path: |
| 11 | manifest = root / "data/candidates/2026-W23/99/publish-manifest.json" |
| 12 | manifest.parent.mkdir(parents=True, exist_ok=True) |
| 13 | manifest.write_text( |
| 14 | json.dumps( |
| 15 | { |
| 16 | "schema_version": "publish_eligibility_v1", |
| 17 | "candidate": {"summary_sha256": "candidate-sha"}, |
| 18 | "source_artifacts": [{"path": "data/raw/2026-W23.json", "sha256": "raw-sha"}], |
| 19 | "analysis": {"source": "copilot-cli"}, |
| 20 | } |
| 21 | ), |
| 22 | encoding="utf-8", |
| 23 | ) |
| 24 | return manifest |
| 25 | |
| 26 | def test_backup_existing_is_immutable_and_restorable_with_provenance(self) -> None: |
| 27 | tests_root = Path(__file__).resolve().parent |
| 28 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 29 | root = Path(tmpdir) |
| 30 | target = root / "content/weekly/2026/W23.md" |
| 31 | target.parent.mkdir(parents=True, exist_ok=True) |
| 32 | target.write_text("known good article\n", encoding="utf-8") |
| 33 | transaction_manifest = root / "data/published/2026-W23/promotion-manifest.json" |
| 34 | transaction_manifest.parent.mkdir(parents=True, exist_ok=True) |
| 35 | transaction_manifest.write_text( |
| 36 | '{"schema_version":"promotion_transaction_v1","transaction_id":"old"}\n', |
| 37 | encoding="utf-8", |
| 38 | ) |
| 39 | source = root / "data/raw/2026-W23.json" |
| 40 | source.parent.mkdir(parents=True, exist_ok=True) |
| 41 | source.write_text('{"week":"2026-W23"}\n', encoding="utf-8") |
| 42 | self.write_manifest(root) |
| 43 | |
| 44 | exit_code = publish_safety.main( |
| 45 | [ |
| 46 | "backup-existing", |
| 47 | "--root", |
| 48 | str(root), |
| 49 | "--week", |
| 50 | "2026-W23", |
| 51 | "--run-id", |
| 52 | "99", |
| 53 | "--kind", |
| 54 | "content", |
| 55 | "--manifest", |
| 56 | "data/candidates/2026-W23/99/publish-manifest.json", |
| 57 | "--expected-publish-ref", |
| 58 | "abc", |
| 59 | "--actual-publish-ref", |
| 60 | "abc", |
| 61 | "--path", |
| 62 | "content/weekly/2026/W23.md", |
| 63 | "--path", |
| 64 | "data/published/2026-W23/promotion-manifest.json", |
| 65 | ] |
| 66 | ) |
| 67 | |
| 68 | self.assertEqual(exit_code, 0) |
| 69 | backup_manifest = root / "data/backups/2026-W23/99/content/manifest.json" |
| 70 | payload = json.loads(backup_manifest.read_text(encoding="utf-8")) |
| 71 | self.assertEqual(payload["schema_version"], "publish_backup_v1") |
| 72 | self.assertEqual(payload["publish_ref"]["expected"], "abc") |
| 73 | self.assertEqual(payload["source_manifest"]["source_artifacts"][0]["sha256"], "raw-sha") |
| 74 | self.assertRegex(payload["files"][0]["sha256"], r"^[0-9a-f]{64}$") |
| 75 | backed_up_paths = {entry["path"] for entry in payload["files"]} |
| 76 | self.assertIn("content/weekly/2026/W23.md", backed_up_paths) |
| 77 | self.assertIn("data/published/2026-W23/promotion-manifest.json", backed_up_paths) |
| 78 | |
| 79 | with self.assertRaises(SystemExit): |
| 80 | publish_safety.main( |
| 81 | [ |
| 82 | "backup-existing", |
| 83 | "--root", |
| 84 | str(root), |
| 85 | "--week", |
| 86 | "2026-W23", |
| 87 | "--run-id", |
| 88 | "99", |
| 89 | "--kind", |
| 90 | "content", |
| 91 | "--manifest", |
| 92 | "data/candidates/2026-W23/99/publish-manifest.json", |
| 93 | "--path", |
| 94 | "content/weekly/2026/W23.md", |
| 95 | ] |
| 96 | ) |
| 97 | |
| 98 | target.write_text("bad replacement\n", encoding="utf-8") |
| 99 | transaction_manifest.write_text( |
| 100 | '{"schema_version":"promotion_transaction_v1","transaction_id":"new"}\n', |
| 101 | encoding="utf-8", |
| 102 | ) |
| 103 | self.assertEqual( |
| 104 | publish_safety.main( |
| 105 | [ |
| 106 | "restore-backup", |
| 107 | "--root", |
| 108 | str(root), |
| 109 | "--backup-manifest", |
| 110 | str(backup_manifest), |
| 111 | ] |
| 112 | ), |
| 113 | 0, |
| 114 | ) |
| 115 | self.assertEqual(target.read_text(encoding="utf-8"), "known good article\n") |
| 116 | restored_transaction = json.loads(transaction_manifest.read_text(encoding="utf-8")) |
| 117 | self.assertEqual(restored_transaction["transaction_id"], "old") |
| 118 | |
| 119 | def test_backup_existing_requires_loadable_publish_manifest(self) -> None: |
| 120 | tests_root = Path(__file__).resolve().parent |
| 121 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 122 | root = Path(tmpdir) |
| 123 | target = root / "content/weekly/2026/W23.md" |
| 124 | target.parent.mkdir(parents=True, exist_ok=True) |
| 125 | target.write_text("known good article\n", encoding="utf-8") |
| 126 | |
| 127 | with self.assertRaises(SystemExit): |
| 128 | publish_safety.main( |
| 129 | [ |
| 130 | "backup-existing", |
| 131 | "--root", |
| 132 | str(root), |
| 133 | "--week", |
| 134 | "2026-W23", |
| 135 | "--run-id", |
| 136 | "99", |
| 137 | "--kind", |
| 138 | "content", |
| 139 | "--manifest", |
| 140 | "data/candidates/2026-W23/99/publish-manifest.json", |
| 141 | "--path", |
| 142 | "content/weekly/2026/W23.md", |
| 143 | ] |
| 144 | ) |
| 145 | self.assertFalse((root / "data/backups").exists()) |
| 146 | |
| 147 | malformed_manifest = root / "data/candidates/2026-W23/99/publish-manifest.json" |
| 148 | malformed_manifest.parent.mkdir(parents=True, exist_ok=True) |
| 149 | malformed_manifest.write_text("{not json", encoding="utf-8") |
| 150 | |
| 151 | with self.assertRaises(SystemExit): |
| 152 | publish_safety.main( |
| 153 | [ |
| 154 | "backup-existing", |
| 155 | "--root", |
| 156 | str(root), |
| 157 | "--week", |
| 158 | "2026-W23", |
| 159 | "--run-id", |
| 160 | "99", |
| 161 | "--kind", |
| 162 | "content", |
| 163 | "--manifest", |
| 164 | "data/candidates/2026-W23/99/publish-manifest.json", |
| 165 | "--path", |
| 166 | "content/weekly/2026/W23.md", |
| 167 | ] |
| 168 | ) |
| 169 | self.assertFalse((root / "data/backups").exists()) |
| 170 | |
| 171 | def test_backup_existing_rejects_non_file_targets(self) -> None: |
| 172 | tests_root = Path(__file__).resolve().parent |
| 173 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 174 | root = Path(tmpdir) |
| 175 | self.write_manifest(root) |
| 176 | target_dir = root / "content/weekly/2026" |
| 177 | target_dir.mkdir(parents=True, exist_ok=True) |
| 178 | |
| 179 | with self.assertRaises(SystemExit): |
| 180 | publish_safety.main( |
| 181 | [ |
| 182 | "backup-existing", |
| 183 | "--root", |
| 184 | str(root), |
| 185 | "--week", |
| 186 | "2026-W23", |
| 187 | "--run-id", |
| 188 | "99", |
| 189 | "--kind", |
| 190 | "content", |
| 191 | "--manifest", |
| 192 | "data/candidates/2026-W23/99/publish-manifest.json", |
| 193 | "--path", |
| 194 | "content/weekly/2026", |
| 195 | ] |
| 196 | ) |
| 197 | self.assertFalse((root / "data/backups").exists()) |
| 198 | |
| 199 | def test_restore_backup_rejects_manifest_outside_root(self) -> None: |
| 200 | tests_root = Path(__file__).resolve().parent |
| 201 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 202 | root = Path(tmpdir) |
| 203 | outside_manifest = root.parent / f"{root.name}-outside-manifest.json" |
| 204 | outside_manifest.write_text( |
| 205 | '{"schema_version":"publish_backup_v1","files":[]}\n', encoding="utf-8" |
| 206 | ) |
| 207 | try: |
| 208 | with self.assertRaises(SystemExit): |
| 209 | publish_safety.main( |
| 210 | [ |
| 211 | "restore-backup", |
| 212 | "--root", |
| 213 | str(root), |
| 214 | "--backup-manifest", |
| 215 | str(outside_manifest), |
| 216 | ] |
| 217 | ) |
| 218 | with self.assertRaises(SystemExit): |
| 219 | publish_safety.main( |
| 220 | [ |
| 221 | "restore-backup", |
| 222 | "--root", |
| 223 | str(root), |
| 224 | "--backup-manifest", |
| 225 | "../outside-manifest.json", |
| 226 | ] |
| 227 | ) |
| 228 | finally: |
| 229 | outside_manifest.unlink(missing_ok=True) |
| 230 | |
| 231 | def test_raw_store_is_immutable_and_restore_is_hash_verified(self) -> None: |
| 232 | tests_root = Path(__file__).resolve().parent |
| 233 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 234 | root = Path(tmpdir) |
| 235 | raw = root / "data/raw/2026-W23.json" |
| 236 | raw.parent.mkdir(parents=True, exist_ok=True) |
| 237 | original_raw = b'{"week":"2026-W23","evidence":"original"}\n' |
| 238 | raw.write_bytes(original_raw) |
| 239 | |
| 240 | self.assertEqual( |
| 241 | publish_safety.main( |
| 242 | [ |
| 243 | "store-raw", |
| 244 | "--root", |
| 245 | str(root), |
| 246 | "--week", |
| 247 | "2026-W23", |
| 248 | "--source-run-id", |
| 249 | "26753498571", |
| 250 | "--source-artifact-id", |
| 251 | "7330965888", |
| 252 | "--source-head-sha", |
| 253 | "abc123", |
| 254 | "--path", |
| 255 | "data/raw/2026-W23.json", |
| 256 | ] |
| 257 | ), |
| 258 | 0, |
| 259 | ) |
| 260 | store = root / "data/raw-store/2026-W23/26753498571" |
| 261 | manifest = json.loads((store / "manifest.json").read_text(encoding="utf-8")) |
| 262 | self.assertEqual(manifest["schema_version"], "raw_store_v1") |
| 263 | self.assertEqual(manifest["source_run_id"], "26753498571") |
| 264 | self.assertEqual(manifest["source_artifact"]["id"], "7330965888") |
| 265 | self.assertEqual(manifest["source_artifact"]["retention_days"], 90) |
| 266 | self.assertEqual(manifest["files"][0]["original_path"], "data/raw/2026-W23.json") |
| 267 | original_store_bytes = { |
| 268 | path.relative_to(store): path.read_bytes() |
| 269 | for path in store.rglob("*") |
| 270 | if path.is_file() |
| 271 | } |
| 272 | |
| 273 | raw.write_bytes(b"replacement that must not enter immutable storage\n") |
| 274 | with self.assertRaisesRegex(SystemExit, "Refusing to overwrite immutable raw store"): |
| 275 | publish_safety.main( |
| 276 | [ |
| 277 | "store-raw", |
| 278 | "--root", |
| 279 | str(root), |
| 280 | "--week", |
| 281 | "2026-W23", |
| 282 | "--source-run-id", |
| 283 | "26753498571", |
| 284 | "--source-artifact-id", |
| 285 | "7330965888", |
| 286 | "--source-head-sha", |
| 287 | "abc123", |
| 288 | "--path", |
| 289 | "data/raw/2026-W23.json", |
| 290 | ] |
| 291 | ) |
| 292 | self.assertEqual( |
| 293 | original_store_bytes, |
| 294 | { |
| 295 | path.relative_to(store): path.read_bytes() |
| 296 | for path in store.rglob("*") |
| 297 | if path.is_file() |
| 298 | }, |
| 299 | ) |
| 300 | |
| 301 | stale_optional = raw.parent / "2026-W23-external-news.json" |
| 302 | stale_optional.write_bytes(b"stale evidence from a different source run\n") |
| 303 | self.assertEqual( |
| 304 | publish_safety.main( |
| 305 | [ |
| 306 | "restore-raw", |
| 307 | "--root", |
| 308 | str(root), |
| 309 | "--week", |
| 310 | "2026-W23", |
| 311 | "--source-run-id", |
| 312 | "26753498571", |
| 313 | ] |
| 314 | ), |
| 315 | 0, |
| 316 | ) |
| 317 | self.assertEqual(raw.read_bytes(), original_raw) |
| 318 | self.assertFalse(stale_optional.exists()) |
| 319 | |
| 320 | stored_raw = store / "files/data/raw/2026-W23.json" |
| 321 | stored_raw.write_bytes(b"X" * len(original_raw)) |
| 322 | raw.write_bytes(b"accepted input must remain unchanged\n") |
| 323 | stale_on_failure = raw.parent / "2026-W23-techcrunch.json" |
| 324 | stale_on_failure.write_bytes(b"also unchanged when stored verification fails\n") |
| 325 | with self.assertRaisesRegex(SystemExit, "checksum mismatch"): |
| 326 | publish_safety.main( |
| 327 | [ |
| 328 | "restore-raw", |
| 329 | "--root", |
| 330 | str(root), |
| 331 | "--week", |
| 332 | "2026-W23", |
| 333 | "--source-run-id", |
| 334 | "26753498571", |
| 335 | ] |
| 336 | ) |
| 337 | self.assertEqual(raw.read_bytes(), b"accepted input must remain unchanged\n") |
| 338 | self.assertTrue(stale_on_failure.exists()) |
| 339 | |
| 340 | def test_restore_accepts_custom_source_artifact_name(self) -> None: |
| 341 | """A non-default --source-artifact-name is still valid provenance and restorable.""" |
| 342 | tests_root = Path(__file__).resolve().parent |
| 343 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 344 | root = Path(tmpdir) |
| 345 | raw = root / "data/raw/2026-W24.json" |
| 346 | raw.parent.mkdir(parents=True, exist_ok=True) |
| 347 | original_raw = b'{"week":"2026-W24","evidence":"original"}\n' |
| 348 | raw.write_bytes(original_raw) |
| 349 | |
| 350 | self.assertEqual( |
| 351 | publish_safety.main( |
| 352 | [ |
| 353 | "store-raw", |
| 354 | "--root", |
| 355 | str(root), |
| 356 | "--week", |
| 357 | "2026-W24", |
| 358 | "--source-run-id", |
| 359 | "26753498572", |
| 360 | "--source-artifact-id", |
| 361 | "7330965999", |
| 362 | "--source-artifact-name", |
| 363 | "raw-evidence-bundle", |
| 364 | "--source-head-sha", |
| 365 | "def456", |
| 366 | "--path", |
| 367 | "data/raw/2026-W24.json", |
| 368 | ] |
| 369 | ), |
| 370 | 0, |
| 371 | ) |
| 372 | store = root / "data/raw-store/2026-W24/26753498572" |
| 373 | manifest = json.loads((store / "manifest.json").read_text(encoding="utf-8")) |
| 374 | self.assertEqual(manifest["source_artifact"]["name"], "raw-evidence-bundle") |
| 375 | |
| 376 | raw.write_bytes(b"drift that restore must overwrite\n") |
| 377 | self.assertEqual( |
| 378 | publish_safety.main( |
| 379 | [ |
| 380 | "restore-raw", |
| 381 | "--root", |
| 382 | str(root), |
| 383 | "--week", |
| 384 | "2026-W24", |
| 385 | "--source-run-id", |
| 386 | "26753498572", |
| 387 | ] |
| 388 | ), |
| 389 | 0, |
| 390 | ) |
| 391 | self.assertEqual(raw.read_bytes(), original_raw) |
| 392 | |
| 393 | |
| 394 | if __name__ == "__main__": |
| 395 | unittest.main() |