| 1 | import json |
| 2 | import os |
| 3 | import tempfile |
| 4 | import unittest |
| 5 | from pathlib import Path |
| 6 | |
| 7 | import scripts.publish_manifest as publish_manifest |
| 8 | from scripts import promotion_guard |
| 9 | |
| 10 | WEEK = "2026-W23" |
| 11 | RUN_STARTED_AT = "2026-06-05T21:16:49Z" |
| 12 | |
| 13 | |
| 14 | GOOD_SUMMARY = """--- |
| 15 | title: "Good AI Article" |
| 16 | date: 2026-06-05T21:16:49Z |
| 17 | week: "2026-W23" |
| 18 | year: 2026 |
| 19 | tags: [ai] |
| 20 | categories: [weekly] |
| 21 | repos_featured: 1 |
| 22 | stars_tracked: 100 |
| 23 | top_repo: "owner/good" |
| 24 | quality_score: 90 |
| 25 | summary: "A good AI-authored weekly summary." |
| 26 | --- |
| 27 | |
| 28 | ## This Week's Trends |
| 29 | |
| 30 | Canonical good analysis. |
| 31 | """ |
| 32 | |
| 33 | GOOD_CONTENT = """--- |
| 34 | title: "Good AI Article" |
| 35 | week: "2026-W23" |
| 36 | draft: false |
| 37 | --- |
| 38 | |
| 39 | Canonical good rendered content. |
| 40 | """ |
| 41 | |
| 42 | VALID_REPLACEMENT_SUMMARY = GOOD_SUMMARY.replace("Good AI Article", "Better AI Article").replace( |
| 43 | "Canonical good analysis.", "Better candidate analysis." |
| 44 | ) |
| 45 | VALID_REPLACEMENT_CONTENT = GOOD_CONTENT.replace("Good AI Article", "Better AI Article").replace( |
| 46 | "Canonical good rendered content.", "Better candidate rendered content." |
| 47 | ) |
| 48 | |
| 49 | |
| 50 | def write_file(root: Path, relative_path: str, content: str) -> Path: |
| 51 | path = root / relative_path |
| 52 | path.parent.mkdir(parents=True, exist_ok=True) |
| 53 | path.write_text(content, encoding="utf-8") |
| 54 | return path |
| 55 | |
| 56 | |
| 57 | def install_existing_good_article(root: Path) -> tuple[Path, Path]: |
| 58 | summary = write_file(root, "data/analyzed/2026-W23-summary.md", GOOD_SUMMARY) |
| 59 | content = write_file(root, "content/weekly/2026/W23.md", GOOD_CONTENT) |
| 60 | return summary, content |
| 61 | |
| 62 | |
| 63 | def write_candidate(root: Path, name: str, summary: str, content: str) -> tuple[Path, Path]: |
| 64 | summary_path = write_file(root, f"data/staging/{WEEK}/{name}/summary.md", summary) |
| 65 | content_path = write_file(root, f"data/staging/{WEEK}/{name}/content.md", content) |
| 66 | return summary_path, content_path |
| 67 | |
| 68 | |
| 69 | def write_source_artifact(root: Path, name: str = "raw") -> Path: |
| 70 | return write_file(root, f"data/raw/{WEEK}-{name}.json", json.dumps({"week": WEEK}) + "\n") |
| 71 | |
| 72 | |
| 73 | def write_publish_raw(root: Path) -> Path: |
| 74 | return write_file( |
| 75 | root, |
| 76 | f"data/raw/{WEEK}.json", |
| 77 | json.dumps( |
| 78 | { |
| 79 | "week": WEEK, |
| 80 | "crawled_at": RUN_STARTED_AT, |
| 81 | "metadata": {"same_day_reuse": "not_reused"}, |
| 82 | } |
| 83 | ) |
| 84 | + "\n", |
| 85 | ) |
| 86 | |
| 87 | |
| 88 | def write_gate_report(root: Path, path: Path, *, passed: bool = True) -> None: |
| 89 | gates = { |
| 90 | "structural_schema": {"passed": True, "errors": []}, |
| 91 | "ai_provenance": {"passed": True, "errors": []}, |
| 92 | "evidence_citation": {"passed": passed, "errors": [] if passed else ["missing evidence"]}, |
| 93 | "editorial_quality": {"passed": True, "errors": []}, |
| 94 | } |
| 95 | write_file( |
| 96 | root, |
| 97 | path.as_posix(), |
| 98 | json.dumps( |
| 99 | { |
| 100 | "passed": passed, |
| 101 | "source": "copilot-cli", |
| 102 | "model": "copilot-default", |
| 103 | "failure_class": "passed" if passed else "evidence_citation", |
| 104 | "errors_after_repair": [] if passed else ["missing evidence"], |
| 105 | "repair_actions": [], |
| 106 | "gates": gates, |
| 107 | } |
| 108 | ) |
| 109 | + "\n", |
| 110 | ) |
| 111 | |
| 112 | |
| 113 | def write_preflight( |
| 114 | root: Path, path: Path, *, degraded: bool = False, publish_eligible: bool = True |
| 115 | ) -> None: |
| 116 | write_file( |
| 117 | root, |
| 118 | path.as_posix(), |
| 119 | json.dumps( |
| 120 | { |
| 121 | "prompt_token_budget": 90000, |
| 122 | "prompt_tokens": 1200, |
| 123 | "prompt_bytes": 4800, |
| 124 | "prompt_checksum_sha256": "a" * 64, |
| 125 | "prompt_within_budget": True, |
| 126 | "degraded": degraded, |
| 127 | "publish_eligible": publish_eligible, |
| 128 | "promotion_policy": "normal-promotion" |
| 129 | if publish_eligible |
| 130 | else "staged/candidate-only by default", |
| 131 | "degradation_reason": "Prompt was deterministically compacted." |
| 132 | if degraded |
| 133 | else None, |
| 134 | "fallback_policy": "copilot-only", |
| 135 | "components": [], |
| 136 | "deterministic_slices": [], |
| 137 | } |
| 138 | ) |
| 139 | + "\n", |
| 140 | ) |
| 141 | |
| 142 | |
| 143 | def create_publish_manifest( |
| 144 | root: Path, |
| 145 | name: str, |
| 146 | *, |
| 147 | source: str = "copilot-cli", |
| 148 | model: str = "copilot-default", |
| 149 | gate_passed: bool = True, |
| 150 | ) -> Path: |
| 151 | candidate_dir = Path("data/candidates") / WEEK / name |
| 152 | summary_path = candidate_dir / f"{WEEK}-summary.md" |
| 153 | manifest_path = candidate_dir / "publish-manifest.json" |
| 154 | gate_report = candidate_dir / "analysis-gate-report.json" |
| 155 | preflight_report = candidate_dir / "diagnostics" / "analysis-preflight.json" |
| 156 | synthesis_file = candidate_dir / "diagnostics" / "synthesis-narrative.md" |
| 157 | write_file(root, summary_path.as_posix(), VALID_REPLACEMENT_SUMMARY) |
| 158 | write_publish_raw(root) |
| 159 | write_gate_report(root, gate_report, passed=gate_passed) |
| 160 | if source == "copilot-cli": |
| 161 | write_preflight(root, preflight_report) |
| 162 | write_file(root, synthesis_file.as_posix(), "Weekly synthesis narrative.\n") |
| 163 | |
| 164 | previous_cwd = Path.cwd() |
| 165 | try: |
| 166 | os.chdir(root) |
| 167 | args = [ |
| 168 | "create", |
| 169 | "--week", |
| 170 | WEEK, |
| 171 | "--run-id", |
| 172 | name, |
| 173 | "--current-datetime", |
| 174 | RUN_STARTED_AT, |
| 175 | "--summary", |
| 176 | summary_path.as_posix(), |
| 177 | "--published-summary", |
| 178 | f"data/analyzed/{WEEK}-summary.md", |
| 179 | "--raw-json", |
| 180 | f"data/raw/{WEEK}.json", |
| 181 | "--analysis-source", |
| 182 | source, |
| 183 | "--analysis-model", |
| 184 | model, |
| 185 | "--validation-status", |
| 186 | "passed" if gate_passed else "failed", |
| 187 | "--gate-report", |
| 188 | gate_report.as_posix(), |
| 189 | "--output", |
| 190 | manifest_path.as_posix(), |
| 191 | ] |
| 192 | if source == "copilot-cli": |
| 193 | args.extend(["--preflight-report", preflight_report.as_posix()]) |
| 194 | # Simulate a successful upstream synthesis step; fail-closed |
| 195 | # behavior for missing/empty/failed synthesis is covered by |
| 196 | # dedicated tests in tests/test_publish_manifest.py. |
| 197 | args.extend(["--synthesis-status", "available"]) |
| 198 | args.extend(["--synthesis-file", synthesis_file.as_posix()]) |
| 199 | publish_manifest.main(args) |
| 200 | finally: |
| 201 | os.chdir(previous_cwd) |
| 202 | return root / manifest_path |
| 203 | |
| 204 | |
| 205 | def assert_eligible_from_root(root: Path, manifest_path: Path) -> int: |
| 206 | previous_cwd = Path.cwd() |
| 207 | try: |
| 208 | os.chdir(root) |
| 209 | return publish_manifest.main(["assert-eligible", "--manifest", str(manifest_path)]) |
| 210 | finally: |
| 211 | os.chdir(previous_cwd) |
| 212 | |
| 213 | |
| 214 | def manifest_for(root: Path, name: str, **overrides) -> Path: |
| 215 | summary_path, content_path = write_candidate( |
| 216 | root, |
| 217 | name, |
| 218 | overrides.pop("summary", VALID_REPLACEMENT_SUMMARY), |
| 219 | overrides.pop("content", VALID_REPLACEMENT_CONTENT), |
| 220 | ) |
| 221 | source_artifact = write_source_artifact(root, name) |
| 222 | manifest = { |
| 223 | "schema_version": "publish_eligibility_v1", |
| 224 | "week": WEEK, |
| 225 | "run_id": f"{WEEK}-{name}", |
| 226 | "run_started_at": RUN_STARTED_AT, |
| 227 | "candidate_summary_path": summary_path.relative_to(root).as_posix(), |
| 228 | "candidate_content_path": content_path.relative_to(root).as_posix(), |
| 229 | "promotion_eligible": True, |
| 230 | "ai_provenance": { |
| 231 | "source": "copilot-cli", |
| 232 | "model": "copilot-default", |
| 233 | "degraded": False, |
| 234 | }, |
| 235 | "gate_results": { |
| 236 | "structural_schema": True, |
| 237 | "ai_provenance": True, |
| 238 | "evidence_citation": True, |
| 239 | "editorial_quality": True, |
| 240 | }, |
| 241 | "source_artifacts": [ |
| 242 | { |
| 243 | "path": source_artifact.relative_to(root).as_posix(), |
| 244 | "checksum": "sha256:test", |
| 245 | "generated_at": RUN_STARTED_AT, |
| 246 | "reused_same_day": False, |
| 247 | "stale": False, |
| 248 | } |
| 249 | ], |
| 250 | } |
| 251 | for key, value in overrides.items(): |
| 252 | manifest[key] = value |
| 253 | manifest_path = root / "data" / "staging" / WEEK / name / "publish-manifest.json" |
| 254 | manifest_path.write_text( |
| 255 | json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8" |
| 256 | ) |
| 257 | return manifest_path |
| 258 | |
| 259 | |
| 260 | def nested_manifest_for(root: Path, name: str, **overrides) -> Path: |
| 261 | summary_path, content_path = write_candidate( |
| 262 | root, |
| 263 | name, |
| 264 | overrides.pop("summary", VALID_REPLACEMENT_SUMMARY), |
| 265 | overrides.pop("content", VALID_REPLACEMENT_CONTENT), |
| 266 | ) |
| 267 | source_artifact = write_source_artifact(root, name) |
| 268 | manifest = { |
| 269 | "schema_version": "publish_eligibility_v1", |
| 270 | "week": WEEK, |
| 271 | "run_id": f"{WEEK}-{name}", |
| 272 | "run_started_at": RUN_STARTED_AT, |
| 273 | "candidate": { |
| 274 | "summary_path": summary_path.relative_to(root).as_posix(), |
| 275 | "content_path": content_path.relative_to(root).as_posix(), |
| 276 | "summary_sha256": "sha256:test", |
| 277 | }, |
| 278 | "analysis": { |
| 279 | "ai_status": "ai", |
| 280 | "source": "copilot-cli", |
| 281 | "model": "copilot-default", |
| 282 | "model_status": "available", |
| 283 | }, |
| 284 | "validation": { |
| 285 | "gate_report": { |
| 286 | "present": True, |
| 287 | "passed": True, |
| 288 | "gates": { |
| 289 | "structural_schema": {"passed": True, "errors": []}, |
| 290 | "ai_provenance": {"passed": True, "errors": []}, |
| 291 | "evidence_citation": {"passed": True, "errors": []}, |
| 292 | "editorial_quality": {"passed": True, "errors": []}, |
| 293 | }, |
| 294 | }, |
| 295 | }, |
| 296 | "promotion": {"eligible": True, "decision": "promote", "reasons": []}, |
| 297 | "source_artifacts": [ |
| 298 | { |
| 299 | "path": source_artifact.relative_to(root).as_posix(), |
| 300 | "sha256": "test", |
| 301 | "crawled_at": RUN_STARTED_AT, |
| 302 | "freshness": {"status": "fresh", "reasons": []}, |
| 303 | } |
| 304 | ], |
| 305 | } |
| 306 | for key, value in overrides.items(): |
| 307 | manifest[key] = value |
| 308 | manifest_path = root / "data" / "staging" / WEEK / name / "publish-manifest.json" |
| 309 | manifest_path.write_text( |
| 310 | json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8" |
| 311 | ) |
| 312 | return manifest_path |
| 313 | |
| 314 | |
| 315 | def no_ai_manifest_for( |
| 316 | root: Path, name: str, *, policy: dict | None = None, quality_score: int = 70 |
| 317 | ) -> Path: |
| 318 | summary = VALID_REPLACEMENT_SUMMARY.replace( |
| 319 | "quality_score: 90", f"quality_score: {quality_score}" |
| 320 | ).replace( |
| 321 | "Better candidate analysis.", "Automated data-only summary generated without AI assistance." |
| 322 | ) |
| 323 | policy = policy or {"mode": "default"} |
| 324 | return manifest_for( |
| 325 | root, |
| 326 | name, |
| 327 | summary=summary, |
| 328 | ai_provenance={ |
| 329 | "source": "no-ai", |
| 330 | "model": "none", |
| 331 | "degraded": False, |
| 332 | "fallback_reason": "copilot quality gate failed", |
| 333 | "attempted_ai_paths": ["provider=copilot-cli,model=copilot-default,status=failed"], |
| 334 | }, |
| 335 | promotion_policy=policy, |
| 336 | ) |
| 337 | |
| 338 | |
| 339 | class PromotionGuardTests(unittest.TestCase): |
| 340 | def test_failed_degraded_and_no_ai_candidates_do_not_replace_existing_good_article( |
| 341 | self, |
| 342 | ) -> None: |
| 343 | tests_root = Path(__file__).resolve().parent |
| 344 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 345 | root = Path(tmpdir) |
| 346 | canonical_summary, canonical_content = install_existing_good_article(root) |
| 347 | original_summary = canonical_summary.read_text(encoding="utf-8") |
| 348 | original_content = canonical_content.read_text(encoding="utf-8") |
| 349 | |
| 350 | blocked_manifests = [ |
| 351 | manifest_for(root, "failed", promotion_eligible=False), |
| 352 | manifest_for( |
| 353 | root, |
| 354 | "degraded", |
| 355 | ai_provenance={ |
| 356 | "source": "copilot-cli", |
| 357 | "model": "copilot-default", |
| 358 | "degraded": True, |
| 359 | }, |
| 360 | ), |
| 361 | manifest_for( |
| 362 | root, |
| 363 | "no-ai", |
| 364 | ai_provenance={"source": "no-ai", "model": "none", "degraded": False}, |
| 365 | ), |
| 366 | ] |
| 367 | |
| 368 | for manifest_path in blocked_manifests: |
| 369 | with self.assertRaises(promotion_guard.PromotionBlocked): |
| 370 | promotion_guard.promote_candidate(manifest_path, root=root) |
| 371 | self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary) |
| 372 | self.assertEqual(canonical_content.read_text(encoding="utf-8"), original_content) |
| 373 | |
| 374 | self.assertTrue((root / "data/staging/2026-W23/no-ai/summary.md").exists()) |
| 375 | diagnostics = list((root / "data/diagnostics/promotion").glob("*-blocked.json")) |
| 376 | self.assertTrue(diagnostics) |
| 377 | |
| 378 | def test_missing_malformed_and_stale_manifests_block_promotion(self) -> None: |
| 379 | tests_root = Path(__file__).resolve().parent |
| 380 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 381 | root = Path(tmpdir) |
| 382 | canonical_summary, _ = install_existing_good_article(root) |
| 383 | original_summary = canonical_summary.read_text(encoding="utf-8") |
| 384 | |
| 385 | with self.assertRaises(promotion_guard.PromotionBlocked) as missing: |
| 386 | promotion_guard.promote_candidate( |
| 387 | root / "data/staging/2026-W23/missing/publish-manifest.json", root=root |
| 388 | ) |
| 389 | self.assertIn("Missing publish eligibility manifest", missing.exception.reasons[0]) |
| 390 | |
| 391 | malformed = root / "data/staging/2026-W23/malformed/publish-manifest.json" |
| 392 | malformed.parent.mkdir(parents=True, exist_ok=True) |
| 393 | malformed.write_text("{not json", encoding="utf-8") |
| 394 | with self.assertRaises(promotion_guard.PromotionBlocked) as bad_json: |
| 395 | promotion_guard.promote_candidate(malformed, root=root) |
| 396 | self.assertIn("Malformed publish eligibility manifest", bad_json.exception.reasons[0]) |
| 397 | |
| 398 | stale_manifest = manifest_for( |
| 399 | root, |
| 400 | "stale", |
| 401 | source_artifacts=[ |
| 402 | { |
| 403 | "path": write_source_artifact(root, "stale").relative_to(root).as_posix(), |
| 404 | "checksum": "sha256:stale", |
| 405 | "generated_at": "2026-06-04T21:16:49Z", |
| 406 | "reused_same_day": False, |
| 407 | "stale": True, |
| 408 | } |
| 409 | ], |
| 410 | ) |
| 411 | with self.assertRaises(promotion_guard.PromotionBlocked) as stale: |
| 412 | promotion_guard.promote_candidate(stale_manifest, root=root) |
| 413 | self.assertIn("source_artifacts[1] is stale.", stale.exception.reasons) |
| 414 | self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary) |
| 415 | |
| 416 | def test_same_successful_rerun_is_stable_and_does_not_duplicate_content(self) -> None: |
| 417 | tests_root = Path(__file__).resolve().parent |
| 418 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 419 | root = Path(tmpdir) |
| 420 | install_existing_good_article(root) |
| 421 | manifest_path = manifest_for(root, "valid") |
| 422 | |
| 423 | first_summary, first_content = promotion_guard.promote_candidate( |
| 424 | manifest_path, root=root |
| 425 | ) |
| 426 | first_summary_text = first_summary.read_text(encoding="utf-8") |
| 427 | first_content_text = first_content.read_text(encoding="utf-8") |
| 428 | |
| 429 | second_summary, second_content = promotion_guard.promote_candidate( |
| 430 | manifest_path, root=root |
| 431 | ) |
| 432 | |
| 433 | self.assertEqual(second_summary.read_text(encoding="utf-8"), first_summary_text) |
| 434 | self.assertEqual(second_content.read_text(encoding="utf-8"), first_content_text) |
| 435 | self.assertEqual( |
| 436 | second_summary.read_text(encoding="utf-8").count("Better candidate analysis."), 1 |
| 437 | ) |
| 438 | self.assertEqual( |
| 439 | second_content.read_text(encoding="utf-8").count( |
| 440 | "Better candidate rendered content." |
| 441 | ), |
| 442 | 1, |
| 443 | ) |
| 444 | transaction_path = root / "data/published/2026-W23/promotion-manifest.json" |
| 445 | first_transaction = json.loads(transaction_path.read_text(encoding="utf-8")) |
| 446 | promotion_guard.promote_candidate(manifest_path, root=root) |
| 447 | second_transaction = json.loads(transaction_path.read_text(encoding="utf-8")) |
| 448 | self.assertEqual(second_transaction, first_transaction) |
| 449 | self.assertEqual(first_transaction["schema_version"], "promotion_transaction_v1") |
| 450 | self.assertEqual( |
| 451 | first_transaction["source_manifest"]["path"], |
| 452 | "data/staging/2026-W23/valid/publish-manifest.json", |
| 453 | ) |
| 454 | self.assertEqual( |
| 455 | first_transaction["provenance"]["source_artifacts"][0]["path"], |
| 456 | "data/raw/2026-W23-valid.json", |
| 457 | ) |
| 458 | self.assertEqual( |
| 459 | first_transaction["published_artifacts"][0]["path"], |
| 460 | "data/analyzed/2026-W23-summary.md", |
| 461 | ) |
| 462 | self.assertEqual( |
| 463 | first_transaction["published_artifacts"][1]["path"], "content/weekly/2026/W23.md" |
| 464 | ) |
| 465 | |
| 466 | def test_no_ai_first_publish_requires_explicit_policy_and_no_existing_good_article( |
| 467 | self, |
| 468 | ) -> None: |
| 469 | tests_root = Path(__file__).resolve().parent |
| 470 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 471 | root = Path(tmpdir) |
| 472 | default_manifest = no_ai_manifest_for(root, "no-ai-default") |
| 473 | |
| 474 | with self.assertRaises(promotion_guard.PromotionBlocked) as default_block: |
| 475 | promotion_guard.promote_candidate(default_manifest, root=root) |
| 476 | self.assertIn( |
| 477 | "no-AI fallback is ineligible for default promotion.", |
| 478 | default_block.exception.reasons, |
| 479 | ) |
| 480 | |
| 481 | allow_manifest = no_ai_manifest_for( |
| 482 | root, "no-ai-first", policy={"mode": "allow-no-ai-first-publish"} |
| 483 | ) |
| 484 | summary_path, _ = promotion_guard.promote_candidate(allow_manifest, root=root) |
| 485 | |
| 486 | self.assertIn("Automated data-only summary", summary_path.read_text(encoding="utf-8")) |
| 487 | |
| 488 | def test_force_replace_no_ai_requires_audit_and_writes_audit_log(self) -> None: |
| 489 | tests_root = Path(__file__).resolve().parent |
| 490 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 491 | root = Path(tmpdir) |
| 492 | canonical_summary, _ = install_existing_good_article(root) |
| 493 | original_summary = canonical_summary.read_text(encoding="utf-8") |
| 494 | missing_audit = no_ai_manifest_for( |
| 495 | root, "force-missing", policy={"mode": "force-replace"} |
| 496 | ) |
| 497 | |
| 498 | with self.assertRaises(promotion_guard.PromotionBlocked) as blocked: |
| 499 | promotion_guard.promote_candidate(missing_audit, root=root) |
| 500 | self.assertIn("force-replace requires a reason.", blocked.exception.reasons) |
| 501 | self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary) |
| 502 | |
| 503 | force_manifest = no_ai_manifest_for( |
| 504 | root, |
| 505 | "force-ok", |
| 506 | policy={ |
| 507 | "mode": "force-replace", |
| 508 | "reason": "operator approved emergency replace", |
| 509 | "actor": "jmservera", |
| 510 | }, |
| 511 | ) |
| 512 | summary_path, _ = promotion_guard.promote_candidate(force_manifest, root=root) |
| 513 | |
| 514 | self.assertIn("Automated data-only summary", summary_path.read_text(encoding="utf-8")) |
| 515 | audit_path = root / "data/diagnostics/promotion/2026-W23-force-replace-audit.json" |
| 516 | self.assertTrue(audit_path.exists()) |
| 517 | audit = json.loads(audit_path.read_text(encoding="utf-8")) |
| 518 | self.assertEqual(audit["actor"], "jmservera") |
| 519 | |
| 520 | def test_force_replace_ai_candidate_requires_reason_and_actor(self) -> None: |
| 521 | tests_root = Path(__file__).resolve().parent |
| 522 | for name, policy, expected_reason in ( |
| 523 | ( |
| 524 | "ai-force-missing-reason", |
| 525 | {"mode": "force-replace", "actor": "jmservera"}, |
| 526 | "force-replace requires a reason.", |
| 527 | ), |
| 528 | ( |
| 529 | "ai-force-missing-actor", |
| 530 | {"mode": "force-replace", "reason": "operator approved correction"}, |
| 531 | "force-replace requires an actor.", |
| 532 | ), |
| 533 | ): |
| 534 | with self.subTest(name=name): |
| 535 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 536 | root = Path(tmpdir) |
| 537 | manifest_path = manifest_for(root, name, promotion_policy=policy) |
| 538 | |
| 539 | with self.assertRaises(promotion_guard.PromotionBlocked) as blocked: |
| 540 | promotion_guard.promote_candidate(manifest_path, root=root) |
| 541 | |
| 542 | self.assertIn(expected_reason, blocked.exception.reasons) |
| 543 | |
| 544 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 545 | root = Path(tmpdir) |
| 546 | manifest_path = manifest_for( |
| 547 | root, |
| 548 | "ai-force-complete", |
| 549 | promotion_policy={ |
| 550 | "mode": "force-replace", |
| 551 | "reason": "operator approved correction", |
| 552 | "actor": "jmservera", |
| 553 | }, |
| 554 | ) |
| 555 | |
| 556 | summary_path, _ = promotion_guard.promote_candidate(manifest_path, root=root) |
| 557 | |
| 558 | self.assertIn("Better candidate analysis.", summary_path.read_text(encoding="utf-8")) |
| 559 | audit_path = root / "data/diagnostics/promotion/2026-W23-force-replace-audit.json" |
| 560 | audit = json.loads(audit_path.read_text(encoding="utf-8")) |
| 561 | self.assertEqual(audit["reason"], "operator approved correction") |
| 562 | self.assertEqual(audit["actor"], "jmservera") |
| 563 | |
| 564 | def test_same_day_reused_source_candidate_can_promote_when_manifest_is_fresh(self) -> None: |
| 565 | tests_root = Path(__file__).resolve().parent |
| 566 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 567 | root = Path(tmpdir) |
| 568 | install_existing_good_article(root) |
| 569 | source_artifact = write_source_artifact(root, "same-day-reuse") |
| 570 | manifest_path = manifest_for( |
| 571 | root, |
| 572 | "same-day-reuse", |
| 573 | source_artifacts=[ |
| 574 | { |
| 575 | "path": source_artifact.relative_to(root).as_posix(), |
| 576 | "checksum": "sha256:same-day", |
| 577 | "generated_at": "2026-06-05T08:00:00Z", |
| 578 | "reused_same_day": True, |
| 579 | "stale": False, |
| 580 | } |
| 581 | ], |
| 582 | ) |
| 583 | |
| 584 | summary_path, content_path = promotion_guard.promote_candidate(manifest_path, root=root) |
| 585 | |
| 586 | self.assertIn("Better AI Article", summary_path.read_text(encoding="utf-8")) |
| 587 | self.assertIn("Better AI Article", content_path.read_text(encoding="utf-8")) |
| 588 | |
| 589 | def test_candidate_paths_cannot_traverse_outside_repository_root(self) -> None: |
| 590 | tests_root = Path(__file__).resolve().parent |
| 591 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 592 | root = Path(tmpdir) |
| 593 | canonical_summary, canonical_content = install_existing_good_article(root) |
| 594 | original_summary = canonical_summary.read_text(encoding="utf-8") |
| 595 | original_content = canonical_content.read_text(encoding="utf-8") |
| 596 | |
| 597 | outside_summary = root.parent / f"{root.name}-outside-summary.md" |
| 598 | outside_content = root.parent / f"{root.name}-outside-content.md" |
| 599 | try: |
| 600 | outside_summary.write_text(VALID_REPLACEMENT_SUMMARY, encoding="utf-8") |
| 601 | outside_content.write_text(VALID_REPLACEMENT_CONTENT, encoding="utf-8") |
| 602 | manifest_path = manifest_for( |
| 603 | root, |
| 604 | "traversal", |
| 605 | candidate_summary_path=f"../{outside_summary.name}", |
| 606 | candidate_content_path=f"../{outside_content.name}", |
| 607 | ) |
| 608 | |
| 609 | with self.assertRaises(promotion_guard.PromotionBlocked) as blocked: |
| 610 | promotion_guard.promote_candidate(manifest_path, root=root) |
| 611 | |
| 612 | self.assertIn( |
| 613 | "candidate_summary_path must stay under the repository root.", |
| 614 | blocked.exception.reasons, |
| 615 | ) |
| 616 | self.assertIn( |
| 617 | "candidate_content_path must stay under the repository root.", |
| 618 | blocked.exception.reasons, |
| 619 | ) |
| 620 | self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary) |
| 621 | self.assertEqual(canonical_content.read_text(encoding="utf-8"), original_content) |
| 622 | finally: |
| 623 | outside_summary.unlink(missing_ok=True) |
| 624 | outside_content.unlink(missing_ok=True) |
| 625 | |
| 626 | def test_manifest_path_must_be_under_allowed_data_manifest_roots(self) -> None: |
| 627 | tests_root = Path(__file__).resolve().parent |
| 628 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 629 | root = Path(tmpdir) |
| 630 | install_existing_good_article(root) |
| 631 | valid_manifest = manifest_for(root, "misplaced") |
| 632 | misplaced_manifest = root / "other" / "data" / "staging" / "publish-manifest.json" |
| 633 | misplaced_manifest.parent.mkdir(parents=True, exist_ok=True) |
| 634 | misplaced_manifest.write_text( |
| 635 | valid_manifest.read_text(encoding="utf-8"), encoding="utf-8" |
| 636 | ) |
| 637 | |
| 638 | with self.assertRaises(promotion_guard.PromotionBlocked) as blocked: |
| 639 | promotion_guard.promote_candidate(misplaced_manifest, root=root) |
| 640 | |
| 641 | self.assertIn( |
| 642 | "Publish manifest must live under data/staging/ or data/candidates/.", |
| 643 | blocked.exception.reasons, |
| 644 | ) |
| 645 | |
| 646 | def test_publish_manifest_outside_allowed_roots_is_rejected_by_both_gates(self) -> None: |
| 647 | tests_root = Path(__file__).resolve().parent |
| 648 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 649 | root = Path(tmpdir) |
| 650 | install_existing_good_article(root) |
| 651 | valid_manifest = create_publish_manifest(root, "outside-root") |
| 652 | misplaced_manifest = ( |
| 653 | root |
| 654 | / "other" |
| 655 | / "data" |
| 656 | / "candidates" |
| 657 | / WEEK |
| 658 | / "outside-root" |
| 659 | / "publish-manifest.json" |
| 660 | ) |
| 661 | misplaced_manifest.parent.mkdir(parents=True, exist_ok=True) |
| 662 | misplaced_manifest.write_text( |
| 663 | valid_manifest.read_text(encoding="utf-8"), encoding="utf-8" |
| 664 | ) |
| 665 | |
| 666 | with self.assertRaises(SystemExit) as assert_blocked: |
| 667 | assert_eligible_from_root(root, misplaced_manifest) |
| 668 | with self.assertRaises(promotion_guard.PromotionBlocked) as promote_blocked: |
| 669 | promotion_guard.promote_candidate(misplaced_manifest, root=root) |
| 670 | |
| 671 | self.assertEqual( |
| 672 | str(assert_blocked.exception), |
| 673 | "Publish manifest must live under data/staging/ or data/candidates/.", |
| 674 | ) |
| 675 | self.assertIn( |
| 676 | "Publish manifest must live under data/staging/ or data/candidates/.", |
| 677 | promote_blocked.exception.reasons, |
| 678 | ) |
| 679 | |
| 680 | def test_publish_manifest_created_candidate_is_accepted_by_promotion_guard(self) -> None: |
| 681 | tests_root = Path(__file__).resolve().parent |
| 682 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 683 | root = Path(tmpdir) |
| 684 | install_existing_good_article(root) |
| 685 | manifest_path = create_publish_manifest(root, "publish-compatible") |
| 686 | |
| 687 | self.assertEqual(assert_eligible_from_root(root, manifest_path), 0) |
| 688 | summary_path, content_path = promotion_guard.promote_candidate(manifest_path, root=root) |
| 689 | |
| 690 | self.assertIn("Better AI Article", summary_path.read_text(encoding="utf-8")) |
| 691 | self.assertIn("Better AI Article", content_path.read_text(encoding="utf-8")) |
| 692 | |
| 693 | def test_publish_manifest_rejected_candidate_is_rejected_by_both_gates(self) -> None: |
| 694 | tests_root = Path(__file__).resolve().parent |
| 695 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 696 | root = Path(tmpdir) |
| 697 | canonical_summary, canonical_content = install_existing_good_article(root) |
| 698 | original_summary = canonical_summary.read_text(encoding="utf-8") |
| 699 | original_content = canonical_content.read_text(encoding="utf-8") |
| 700 | manifest_path = create_publish_manifest( |
| 701 | root, "publish-rejected", source="no-ai", model="none" |
| 702 | ) |
| 703 | |
| 704 | with self.assertRaises(SystemExit): |
| 705 | assert_eligible_from_root(root, manifest_path) |
| 706 | with self.assertRaises(promotion_guard.PromotionBlocked): |
| 707 | promotion_guard.promote_candidate(manifest_path, root=root) |
| 708 | |
| 709 | self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary) |
| 710 | self.assertEqual(canonical_content.read_text(encoding="utf-8"), original_content) |
| 711 | |
| 712 | def test_nested_manifest_gate_decisions_are_consumed(self) -> None: |
| 713 | tests_root = Path(__file__).resolve().parent |
| 714 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 715 | root = Path(tmpdir) |
| 716 | install_existing_good_article(root) |
| 717 | blocked = nested_manifest_for( |
| 718 | root, |
| 719 | "nested-blocked", |
| 720 | validation={ |
| 721 | "gate_report": { |
| 722 | "present": True, |
| 723 | "passed": False, |
| 724 | "gates": { |
| 725 | "structural_schema": {"passed": True, "errors": []}, |
| 726 | "ai_provenance": {"passed": True, "errors": []}, |
| 727 | "evidence_citation": {"passed": False, "errors": ["missing evidence"]}, |
| 728 | "editorial_quality": {"passed": True, "errors": []}, |
| 729 | }, |
| 730 | } |
| 731 | }, |
| 732 | promotion={"eligible": False, "decision": "block", "reasons": ["missing evidence"]}, |
| 733 | ) |
| 734 | |
| 735 | with self.assertRaises(promotion_guard.PromotionBlocked) as raised: |
| 736 | promotion_guard.promote_candidate(blocked, root=root) |
| 737 | |
| 738 | self.assertIn("promotion_eligible must be true.", raised.exception.reasons) |
| 739 | self.assertIn("validation.gate_report.passed must be true.", raised.exception.reasons) |
| 740 | self.assertIn("evidence_citation must pass.", raised.exception.reasons) |
| 741 | |
| 742 | def test_nested_manifest_missing_required_gate_family_blocks_promotion(self) -> None: |
| 743 | tests_root = Path(__file__).resolve().parent |
| 744 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 745 | root = Path(tmpdir) |
| 746 | install_existing_good_article(root) |
| 747 | manifest_path = nested_manifest_for( |
| 748 | root, |
| 749 | "missing-gate-family", |
| 750 | validation={ |
| 751 | "gate_report": { |
| 752 | "present": True, |
| 753 | "passed": True, |
| 754 | "gates": { |
| 755 | "structural_schema": {"passed": True, "errors": []}, |
| 756 | "ai_provenance": {"passed": True, "errors": []}, |
| 757 | "editorial_quality": {"passed": True, "errors": []}, |
| 758 | }, |
| 759 | } |
| 760 | }, |
| 761 | ) |
| 762 | |
| 763 | with self.assertRaises(promotion_guard.PromotionBlocked) as raised: |
| 764 | promotion_guard.promote_candidate(manifest_path, root=root) |
| 765 | |
| 766 | self.assertIn( |
| 767 | "gate_results must include passing evidence_citation.", raised.exception.reasons |
| 768 | ) |
| 769 | |
| 770 | |
| 771 | if __name__ == "__main__": |
| 772 | unittest.main() |