from __future__ import annotations
|
|
import hashlib
|
import json
|
import os
|
import shutil
|
import subprocess
|
import sys
|
import tempfile
|
import unittest
|
from unittest import mock
|
from datetime import datetime, timezone
|
from pathlib import Path
|
|
|
PROJECT_DEV = Path(__file__).resolve().parents[1]
|
PROJECT_ROOT = PROJECT_DEV.parents[1]
|
sys.path.insert(0, str(PROJECT_DEV))
|
|
import bili_half_hour_pipeline as product
|
|
|
NOW = datetime(2026, 8, 29, 8, 0, tzinfo=timezone.utc)
|
UID = "1420210197"
|
BVID = "BV1Q541167Qg"
|
|
|
def canonical(value: object) -> bytes:
|
return (json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n").encode("ascii")
|
|
|
class PipelineFixture:
|
def __init__(self, root: Path) -> None:
|
self.root = root
|
self.archive = root / "ana-data" / "news-fixture"
|
self.state = root / "dev" / "tmp" / "pipeline"
|
self.archive.mkdir(parents=True)
|
self.formal = self.archive / "manifest.jsonl"
|
self.handoffs = self.archive / "video-processing-handoffs.jsonl"
|
self.formal.write_bytes(canonical({
|
"schema_version": 1, "stable_id": "OLD", "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": "old.txt", "bytes": 3, "sha256": hashlib.sha256(b"old").hexdigest().upper()
|
}))
|
(self.archive / "old.txt").write_bytes(b"old")
|
self.handoffs.write_bytes(canonical({
|
"schema_version": 1, "type": "media-processing-handoff", "status": "READY",
|
"handoff_id": "HANDOFF-OLD", "queue_job_id": "0" * 64, "creator_uid": UID,
|
"bvid": "BV1Q541167Qf", "source_url": "https://www.bilibili.com/video/BV1Q541167Qf",
|
"media_path": "F:/video/old.mkv", "mapping_path": "F:/video/old.download.json",
|
"bytes": 1, "sha256": "A" * 64, "duration_seconds": 1.0,
|
"video_codec": "av1", "audio_codec": "aac", "created_at": NOW.isoformat()
|
}))
|
self.config_path = root / "pipeline.json"
|
self.config_path.write_bytes(canonical({
|
"schema_version": 1,
|
"task_id": product.TASK_ID,
|
"interval_minutes": 30,
|
"creator": {"uid": UID, "name": "fixture", "dynamic_url": f"https://space.bilibili.com/{UID}/dynamic"},
|
"paths": {
|
"project_root": str(root), "archive_root": str(self.archive),
|
"formal_manifest": str(self.formal), "processing_handoffs": str(self.handoffs),
|
"state_dir": str(self.state), "video_root": str(root / "external-video")
|
},
|
"downstream": {
|
"video_downloader_thread_id": "019fcc5d-798f-7ea1-8325-3a4d1f2dc5a5",
|
"media_processor_thread_id": "019fb7a4-bdfd-79f2-bd6b-e67e2b7d8efd",
|
"minutes_thread_id": "019fae88-ef98-7f83-b964-dcd4c5f842ef",
|
"reply_thread_id": "019fbcbb-bed7-7c90-83ab-f50610f80d3a"
|
},
|
"git": {
|
"remote": "origin",
|
"branch": "main",
|
"allowed_extensions": [".txt", ".md", ".json", ".jsonl", ".srt", ".pdf", ".png", ".jpg"],
|
"allowed_docs": ["ana-data/news-fixture/目录导读.md"],
|
}
|
}))
|
(self.archive / "目录导读.md").write_bytes(b"# fixture\n")
|
self.config = product.load_config(self.config_path)
|
|
def append_formal(self, value: dict[str, object]) -> None:
|
with self.formal.open("ab") as stream:
|
stream.write(canonical(value))
|
|
def append_handoff(self, value: dict[str, object]) -> None:
|
with self.handoffs.open("ab") as stream:
|
stream.write(canonical(value))
|
|
|
def prepare_migration_fixture(root: Path) -> tuple[PipelineFixture, dict[str, Path], dict[str, bytes], Path]:
|
fixture = PipelineFixture(root)
|
(fixture.root / "external-video").mkdir()
|
fixture.append_formal({
|
"schema_version": 1,
|
"stable_id": BVID,
|
"item_type": "video",
|
"status": product.VIDEO_COMPLETE,
|
"creator": "fixture",
|
"title": "A / canonical: title?",
|
"published_at": "2026-08-30T12:34:56+08:00",
|
})
|
transcript_dir = fixture.archive / f"{BVID}.transcript"
|
minutes_dir = fixture.archive / f"{BVID}.minutes"
|
transcript_dir.mkdir()
|
minutes_dir.mkdir()
|
payloads = {
|
"transcript_txt": b"transcript\n",
|
"transcript_srt": b"1\n00:00:00,000 --> 00:00:01,000\ntext\n",
|
"transcript_json": b'{"segments":[]}\n',
|
"minutes_md": b"# minutes\n",
|
"minutes_pdf": b"%PDF-fixture\n",
|
}
|
legacy_paths = {
|
"transcript_txt": transcript_dir / f"{BVID}.txt",
|
"transcript_srt": transcript_dir / f"{BVID}.srt",
|
"transcript_json": transcript_dir / f"{BVID}.json",
|
"minutes_md": minutes_dir / f"{BVID}.md",
|
"minutes_pdf": minutes_dir / f"{BVID}.pdf",
|
}
|
for kind, path in legacy_paths.items():
|
path.write_bytes(payloads[kind])
|
flac = transcript_dir / f"{BVID}.audio.flac"
|
flac.write_bytes(b"fLaC-fixture")
|
subprocess.run(["git", "init", "-b", "main"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.name", "Fixture"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.email", "fixture@example.invalid"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "add", "."], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "commit", "-m", "legacy baseline"], cwd=fixture.root, check=True, capture_output=True)
|
payloads["minutes_pdf"] = b"%PDF-modified-worktree-fixture\n"
|
legacy_paths["minutes_pdf"].write_bytes(payloads["minutes_pdf"])
|
(fixture.archive / "目录导读.md").write_bytes(b"# canonical naming guide\n")
|
product.initialize(fixture.config, NOW)
|
return fixture, legacy_paths, payloads, flac
|
|
|
class HalfHourPipelineTests(unittest.TestCase):
|
@staticmethod
|
def _new_handoff(fixture: PipelineFixture, *, source_url: str | None = None) -> None:
|
fixture.append_handoff({
|
"schema_version": 1, "type": "media-processing-handoff", "status": "READY",
|
"handoff_id": "HANDOFF-NEW", "queue_job_id": "1" * 64, "creator_uid": UID, "bvid": BVID,
|
"source_url": source_url or f"https://www.bilibili.com/video/{BVID}", "media_path": "F:/video/new.mkv",
|
"mapping_path": "F:/video/new.download.json", "bytes": 100, "sha256": "B" * 64,
|
"duration_seconds": 60.0, "video_codec": "av1", "audio_codec": "aac", "created_at": NOW.isoformat(),
|
})
|
|
def test_external_schema_integers_are_strict_and_fail_before_mutation(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
original = json.loads(fixture.config_path.read_text(encoding="ascii"))
|
for field, bad_values in {
|
"schema_version": [True, 1.0, "1", None],
|
"interval_minutes": [True, 30.0, "30", None],
|
}.items():
|
for bad in bad_values:
|
candidate = dict(original)
|
candidate[field] = bad
|
fixture.config_path.write_bytes(canonical(candidate))
|
with self.assertRaises(product.PipelineError) as rejected:
|
product.load_config(fixture.config_path)
|
self.assertEqual("E_CONFIG", rejected.exception.code)
|
self.assertFalse(fixture.state.exists())
|
fixture.config_path.write_bytes(canonical(original))
|
config = product.load_config(fixture.config_path)
|
product.initialize(config, NOW)
|
state_before = config.state_path.read_bytes()
|
state = json.loads(state_before)
|
state["schema_version"] = True
|
config.state_path.write_bytes(canonical(state))
|
with self.assertRaises(product.PipelineError) as rejected:
|
product.begin(config, NOW)
|
self.assertEqual("E_STATE", rejected.exception.code)
|
self.assertFalse(config.runs_path.exists())
|
|
def test_processing_handoff_accepts_exact_schema_aliases_and_rejects_schema_drift(self) -> None:
|
def handoff(schema_key: str | None, schema_value: object = 1) -> dict[str, object]:
|
value: dict[str, object] = {
|
"type": "media-processing-handoff", "status": "READY",
|
"handoff_id": "HANDOFF-SCHEMA", "queue_job_id": "1" * 64,
|
"creator_uid": UID, "bvid": BVID, "source_url": f"https://www.bilibili.com/video/{BVID}",
|
"media_path": "F:/video/schema.mkv", "mapping_path": "F:/video/schema.download.json",
|
"bytes": 100, "sha256": "B" * 64, "duration_seconds": 60.0,
|
"video_codec": "av1", "audio_codec": "aac", "created_at": NOW.isoformat(),
|
}
|
if schema_key is not None:
|
value[schema_key] = schema_value
|
return value
|
|
positive_sha = (
|
("schema", "b" * 64),
|
("schema_version", "B" * 64),
|
("schema_version", "Aa" * 32),
|
)
|
for schema_key, sha256 in positive_sha:
|
with self.subTest(positive=(schema_key, sha256[:2])), tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
row = handoff(schema_key)
|
row["sha256"] = sha256
|
fixture.append_handoff(row)
|
created = product.reconcile(fixture.config, NOW)["created_outbox_ids"]
|
self.assertEqual(1, len(created))
|
item = product.pending(fixture.config)["items"][0]
|
self.assertEqual(("VIDEO_TRANSCRIPTION_READY", BVID), (item["kind"], item["payload"]["bvid"]))
|
self.assertEqual(sha256.upper(), item["payload"]["sha256"])
|
|
invalid: list[tuple[str, dict[str, object]]] = [
|
("bool", handoff("schema", True)),
|
("float", handoff("schema", 1.0)),
|
("string", handoff("schema", "1")),
|
("null", handoff("schema", None)),
|
("wrong_integer", handoff("schema", 2)),
|
("missing", handoff(None)),
|
]
|
invalid_sha = handoff("schema")
|
invalid_sha["sha256"] = "G" * 64
|
invalid.append(("invalid_sha", invalid_sha))
|
unicode_expansion_sha = handoff("schema")
|
unicode_expansion_sha["sha256"] = "0" * 62 + "\ufb00"
|
invalid.append(("unicode_expansion_sha", unicode_expansion_sha))
|
dual = handoff("schema")
|
dual["schema_version"] = 1
|
invalid.append(("dual", dual))
|
extra = handoff("schema")
|
extra["schema_extra"] = 1
|
invalid.append(("extra", extra))
|
for label, row in invalid:
|
with self.subTest(negative=label), tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
fixture.append_handoff(row)
|
state_before = fixture.config.state_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as rejected:
|
product.reconcile(fixture.config, NOW)
|
self.assertEqual("E_SOURCE_BINDING", rejected.exception.code)
|
self.assertEqual(state_before, fixture.config.state_path.read_bytes())
|
self.assertFalse(fixture.config.outbox_path.exists())
|
|
def test_outbox_history_schema_tamper_fails_without_append(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
fixture.append_handoff({
|
"schema_version": 1, "type": "media-processing-handoff", "status": "READY",
|
"handoff_id": "HANDOFF-NEW", "queue_job_id": "1" * 64, "creator_uid": UID, "bvid": BVID,
|
"source_url": f"https://www.bilibili.com/video/{BVID}", "media_path": "F:/video/new.mkv",
|
"mapping_path": "F:/video/new.download.json", "bytes": 100, "sha256": "B" * 64,
|
"duration_seconds": 60.0, "video_codec": "av1", "audio_codec": "aac", "created_at": NOW.isoformat(),
|
})
|
product.reconcile(fixture.config, NOW)
|
rows = [json.loads(line) for line in fixture.config.outbox_path.read_text(encoding="ascii").splitlines()]
|
rows[0]["schema_version"] = True
|
fixture.config.outbox_path.write_bytes(b"".join(canonical(row) for row in rows))
|
before = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as rejected:
|
product.pending(fixture.config)
|
self.assertEqual("E_OUTBOX", rejected.exception.code)
|
self.assertEqual(before, fixture.config.outbox_path.read_bytes())
|
|
def test_baseline_nonoverlap_reconcile_and_exact_once_outbox(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
initialized = product.initialize(fixture.config, NOW)
|
self.assertEqual(("INITIALIZED", 1, 1), (
|
initialized["status"], initialized["baseline"]["formal"]["lines"], initialized["baseline"]["handoff"]["lines"]
|
))
|
self.assertEqual("RUN_STARTED", product.begin(fixture.config, NOW)["status"])
|
self.assertEqual("RUN_RESUMED", product.begin(fixture.config, NOW)["status"])
|
with self.assertRaises(product.PipelineError) as active:
|
product.begin(fixture.config, datetime(2026, 8, 29, 8, 30, tzinfo=timezone.utc))
|
self.assertEqual("E_RUN_ACTIVE", active.exception.code)
|
|
body = b"new text\n"
|
image = b"\x89PNG\r\nfixture"
|
(fixture.archive / "new.txt").write_bytes(body)
|
(fixture.archive / "new.png").write_bytes(image)
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "123456", "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": "new.txt", "bytes": len(body), "sha256": hashlib.sha256(body).hexdigest().upper(),
|
"image_path": "new.png", "image_bytes": len(image), "image_sha256": hashlib.sha256(image).hexdigest().upper()
|
})
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": BVID, "item_type": "video", "status": "VIDEO_DOWNLOAD_PENDING_EXTENSION",
|
"creator_uid": UID, "source_url": f"https://www.bilibili.com/video/{BVID}", "title": "fixture video",
|
"published_at": NOW.isoformat(), "expected_duration_seconds": 60
|
})
|
fixture.append_handoff({
|
"schema_version": 1, "type": "media-processing-handoff", "status": "READY",
|
"handoff_id": "HANDOFF-NEW", "queue_job_id": "1" * 64, "creator_uid": UID, "bvid": BVID,
|
"source_url": f"https://www.bilibili.com/video/{BVID}", "media_path": "F:/video/new.mkv",
|
"mapping_path": "F:/video/new.download.json", "bytes": 100, "sha256": "B" * 64,
|
"duration_seconds": 60.0, "video_codec": "av1", "audio_codec": "aac", "created_at": NOW.isoformat()
|
})
|
result = product.reconcile(fixture.config, NOW)
|
self.assertEqual(3, len(result["created_outbox_ids"]))
|
self.assertEqual(0, len(product.reconcile(fixture.config, NOW)["created_outbox_ids"]))
|
pending = product.pending(fixture.config)
|
self.assertEqual(3, pending["count"])
|
self.assertEqual(
|
{"GIT_DELIVERY_READY", "VIDEO_DOWNLOAD_READY", "VIDEO_TRANSCRIPTION_READY"},
|
{item["kind"] for item in pending["items"]},
|
)
|
video_item = next(item for item in pending["items"] if item["kind"] == "VIDEO_DOWNLOAD_READY")
|
self.assertEqual(
|
fixture.config.video_downloader_thread_id,
|
product.dispatch_intent(fixture.config, video_item["outbox_id"], NOW)["target_thread_id"],
|
)
|
serialized = fixture.config.outbox_path.read_text(encoding="ascii").lower()
|
for secret in ("cookie", "sessdata", "localstorage", "profile"):
|
self.assertNotIn(secret, serialized)
|
|
def test_dispatch_intent_is_restart_safe_and_observed_exact_once(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
fixture.append_handoff({
|
"schema_version": 1, "type": "media-processing-handoff", "status": "READY",
|
"handoff_id": "HANDOFF-NEW", "queue_job_id": "1" * 64, "creator_uid": UID, "bvid": BVID,
|
"source_url": f"https://www.bilibili.com/video/{BVID}", "media_path": "F:/video/new.mkv",
|
"mapping_path": "F:/video/new.download.json", "bytes": 100, "sha256": "B" * 64,
|
"duration_seconds": 60.0, "video_codec": "av1", "audio_codec": "aac", "created_at": NOW.isoformat()
|
})
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
intent = product.dispatch_intent(fixture.config, outbox_id, NOW)
|
self.assertEqual(("DISPATCH_INTENT_DURABLE", fixture.config.media_thread_id), (intent["status"], intent["target_thread_id"]))
|
self.assertEqual(outbox_id, intent["envelope"]["outbox_id"])
|
self.assertEqual("DISPATCH_INTENT", product.pending(fixture.config)["items"][0]["delivery_state"])
|
resumed = product.dispatch_intent(fixture.config, outbox_id, NOW)
|
self.assertEqual(("DISPATCH_INTENT_RESUMED", intent["envelope"]), (resumed["status"], resumed["envelope"]))
|
observed = product.observe_dispatch(fixture.config, outbox_id, "DELIVERY-1", NOW)
|
self.assertEqual("DISPATCH_OBSERVED", observed["status"])
|
self.assertEqual("DISPATCH_ALREADY_OBSERVED", product.observe_dispatch(fixture.config, outbox_id, "DELIVERY-1", NOW)["status"])
|
self.assertEqual(0, product.pending(fixture.config)["count"])
|
with self.assertRaises(product.PipelineError) as conflict:
|
product.observe_dispatch(fixture.config, outbox_id, "DELIVERY-2", NOW)
|
self.assertEqual("E_DISPATCH_RECEIPT", conflict.exception.code)
|
|
def test_transcript_and_minutes_receipts_create_restart_safe_downstream(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
fixture.append_handoff({
|
"schema_version": 1, "type": "media-processing-handoff", "status": "READY",
|
"handoff_id": "HANDOFF-NEW", "queue_job_id": "1" * 64, "creator_uid": UID, "bvid": BVID,
|
"source_url": f"https://www.bilibili.com/video/{BVID}", "media_path": "F:/video/new.mkv",
|
"mapping_path": "F:/video/new.download.json", "bytes": 100, "sha256": "B" * 64,
|
"duration_seconds": 60.0, "video_codec": "av1", "audio_codec": "aac", "created_at": NOW.isoformat()
|
})
|
source_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
product.dispatch_intent(fixture.config, source_id, NOW)
|
product.observe_dispatch(fixture.config, source_id, "TRANSCRIPT-DISPATCH-1", NOW)
|
transcript = fixture.archive / "new.transcript.txt"
|
transcript_srt = fixture.archive / "new.transcript.srt"
|
transcript_json = fixture.archive / "new.transcript.json"
|
transcript.write_bytes(b"transcript\n")
|
transcript_srt.write_bytes(b"1\n00:00:00,000 --> 00:00:01,000\ntranscript\n")
|
transcript_json.write_bytes(b'{"segments":[]}\n')
|
receipt = fixture.root / "transcript-receipt.json"
|
receipt.write_bytes(canonical({
|
"schema_version": 1, "type": "TRANSCRIPTION_COMPLETE", "outbox_id": source_id,
|
"stable_id": BVID, "terminal_id": "TRANSCRIPT-1", "status": "COMPLETE",
|
"files": [
|
{"path": str(transcript), "bytes": transcript.stat().st_size,
|
"sha256": hashlib.sha256(transcript.read_bytes()).hexdigest().upper(), "kind": "transcript_txt"},
|
{"path": str(transcript_srt), "bytes": transcript_srt.stat().st_size,
|
"sha256": hashlib.sha256(transcript_srt.read_bytes()).hexdigest().upper(), "kind": "transcript_srt"},
|
{"path": str(transcript_json), "bytes": transcript_json.stat().st_size,
|
"sha256": hashlib.sha256(transcript_json.read_bytes()).hexdigest().upper(), "kind": "transcript_json"},
|
],
|
"created_at": NOW.isoformat()
|
}))
|
committed = product.ingest_receipt(fixture.config, receipt, NOW)
|
self.assertEqual(("RECEIPT_COMMITTED", 2), (committed["status"], len(committed["created_outbox_ids"])))
|
replay = product.ingest_receipt(fixture.config, receipt, NOW)
|
self.assertEqual("RECEIPT_ALREADY_COMMITTED", replay["status"])
|
terminal_before = fixture.config.terminals_path.read_bytes()
|
outbox_before = fixture.config.outbox_path.read_bytes()
|
conflicting = json.loads(receipt.read_text(encoding="ascii"))
|
conflicting["terminal_id"] = "TRANSCRIPT-CONFLICT"
|
receipt.write_bytes(canonical(conflicting))
|
with self.assertRaises(product.PipelineError) as conflict:
|
product.ingest_receipt(fixture.config, receipt, NOW)
|
self.assertEqual("E_RECEIPT_CONFLICT", conflict.exception.code)
|
self.assertEqual(terminal_before, fixture.config.terminals_path.read_bytes())
|
self.assertEqual(outbox_before, fixture.config.outbox_path.read_bytes())
|
items = product.pending(fixture.config)["items"]
|
minutes_item = next(item for item in items if item["kind"] == "MINUTES_READY")
|
product.dispatch_intent(fixture.config, minutes_item["outbox_id"], NOW)
|
product.observe_dispatch(fixture.config, minutes_item["outbox_id"], "MINUTES-DISPATCH-1", NOW)
|
minutes = fixture.archive / "new.minutes.md"
|
minutes_pdf = fixture.archive / "new.minutes.pdf"
|
minutes.write_bytes(b"# minutes\n")
|
minutes_pdf.write_bytes(b"%PDF-fixture\n")
|
minutes_receipt = fixture.root / "minutes-receipt.json"
|
minutes_receipt.write_bytes(canonical({
|
"schema_version": 1, "type": "MINUTES_COMPLETE", "outbox_id": minutes_item["outbox_id"],
|
"stable_id": BVID, "terminal_id": "MINUTES-1", "status": "COMPLETE",
|
"files": [
|
{"path": str(minutes), "bytes": minutes.stat().st_size,
|
"sha256": hashlib.sha256(minutes.read_bytes()).hexdigest().upper(), "kind": "minutes_md"},
|
{"path": str(minutes_pdf), "bytes": minutes_pdf.stat().st_size,
|
"sha256": hashlib.sha256(minutes_pdf.read_bytes()).hexdigest().upper(), "kind": "minutes_pdf"},
|
],
|
"created_at": NOW.isoformat()
|
}))
|
result = product.ingest_receipt(fixture.config, minutes_receipt, NOW)
|
self.assertEqual(1, len(result["created_outbox_ids"]))
|
git_items = [
|
row for row in product._outbox_rows(fixture.config)
|
if row.get("event") == "CREATED" and row.get("kind") == "GIT_DELIVERY_READY"
|
]
|
self.assertEqual(2, len(git_items))
|
self.assertEqual([3, 2], [len(product._git_artifacts(fixture.config, row["payload"]["files"])[0]) for row in git_items])
|
mismatched = dict(git_items[0]["payload"]["files"][0])
|
mismatched["kind"] = "minutes_pdf"
|
with self.assertRaises(product.PipelineError) as kind_drift:
|
product._git_artifacts(fixture.config, [mismatched])
|
self.assertEqual("E_GIT_SCOPE", kind_drift.exception.code)
|
|
def test_git_delivery_uses_exact_paths_and_nonforce_push(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
(fixture.root / ".git").mkdir()
|
(fixture.root / ".git" / "index").write_bytes(b"USER-INDEX")
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
artifact = fixture.archive / "new.txt"
|
artifact.write_bytes(b"content\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "123", "item_type": "text", "status": "SAVED", "creator_uid": UID,
|
"path": "new.txt", "bytes": artifact.stat().st_size, "sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper()
|
})
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
commands: list[list[str]] = []
|
rel = artifact.relative_to(fixture.root).as_posix()
|
|
def runner(command: list[str], **_: object) -> subprocess.CompletedProcess[str]:
|
commands.append(command)
|
if command[:5] == ["git", "diff", "--cached", "--name-only", "-z"] and command[5] != "--":
|
return subprocess.CompletedProcess(command, 0, b"", b"")
|
if command[:6] == ["git", "diff", "--cached", "--name-only", "-z", "--"]:
|
return subprocess.CompletedProcess(command, 0, rel.encode("utf-8") + b"\0", b"")
|
if command[:4] == ["git", "diff", "--cached", "--name-only"]:
|
return subprocess.CompletedProcess(command, 0, "", "")
|
if command[:3] == ["git", "rev-parse", "HEAD"]:
|
return subprocess.CompletedProcess(command, 0, "a" * 40 + "\n", "")
|
if command[:2] == ["git", "hash-object"]:
|
return subprocess.CompletedProcess(command, 0, "d" * 40 + "\n", "")
|
if command[:2] == ["git", "write-tree"]:
|
return subprocess.CompletedProcess(command, 0, "b" * 40 + "\n", "")
|
if command[:2] == ["git", "commit-tree"]:
|
return subprocess.CompletedProcess(command, 0, "c" * 40 + "\n", "")
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
result = product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)
|
self.assertEqual("GIT_PUSHED", result["status"])
|
self.assertIn(["git", "update-index", "--add", "--cacheinfo", f"100644,{'d' * 40},{rel}"], commands)
|
self.assertIn(["git", "push", "origin", "c" * 40 + ":refs/heads/main"], commands)
|
flattened = "\n".join(" ".join(command) for command in commands)
|
self.assertNotIn("git add .", flattened)
|
self.assertNotIn("--force", flattened)
|
|
def test_git_delivery_accepts_exact_published_head_without_git_writes(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
for command in (
|
["git", "init", "-b", "main"],
|
["git", "config", "user.name", "Fixture"],
|
["git", "config", "user.email", "fixture@example.invalid"],
|
["git", "add", "."],
|
["git", "commit", "-m", "fixture baseline"],
|
):
|
subprocess.run(command, cwd=fixture.root, check=True, capture_output=True)
|
remote = fixture.root / "remote.git"
|
subprocess.run(["git", "init", "--bare", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "push", "-u", "origin", "main"], cwd=fixture.root, check=True, capture_output=True)
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
artifact = fixture.archive / "已交付.txt"
|
artifact.write_bytes(b"already published\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "published", "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": artifact.name, "bytes": artifact.stat().st_size,
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
subprocess.run(
|
["git", "add", "--", artifact.relative_to(fixture.root).as_posix(), fixture.formal.relative_to(fixture.root).as_posix()],
|
cwd=fixture.root, check=True, capture_output=True,
|
)
|
subprocess.run(["git", "commit", "-m", "accepted data delivery"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "push", "origin", "main"], cwd=fixture.root, check=True, capture_output=True)
|
head_before = subprocess.run(
|
["git", "rev-parse", "HEAD"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip()
|
shared_index = fixture.root / ".git" / "index"
|
index_before = shared_index.read_bytes()
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
commands: list[list[str]] = []
|
|
def runner(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[Any]:
|
commands.append(command)
|
return subprocess.run(command, **kwargs)
|
|
result = product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)
|
self.assertEqual(("GIT_NO_CHANGES", head_before), (result["status"], result["commit_sha"]))
|
self.assertEqual(index_before, shared_index.read_bytes())
|
self.assertEqual(head_before, subprocess.run(
|
["git", "rev-parse", "HEAD"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip())
|
self.assertEqual(head_before, subprocess.run(
|
["git", "rev-parse", "refs/remotes/origin/main"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip())
|
forbidden = {"hash-object", "update-index", "commit-tree", "update-ref", "push"}
|
self.assertFalse(any(len(command) > 1 and command[1] in forbidden for command in commands))
|
self.assertEqual(0, len(list(fixture.config.state_dir.glob(f"git-index-{outbox_id}*"))))
|
events = [json.loads(line) for line in fixture.config.outbox_path.read_text(encoding="ascii").splitlines()]
|
complete = [row for row in events if row.get("outbox_id") == outbox_id and row.get("event") == "COMPLETE"]
|
self.assertEqual(["NO_CHANGES"], [row["result"] for row in complete])
|
|
local_only = fixture.archive / "local-only.txt"
|
local_only.write_bytes(b"not published\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "local-only", "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": local_only.name, "bytes": local_only.stat().st_size,
|
"sha256": hashlib.sha256(local_only.read_bytes()).hexdigest().upper(),
|
})
|
subprocess.run(
|
["git", "add", "--", local_only.relative_to(fixture.root).as_posix(), fixture.formal.relative_to(fixture.root).as_posix()],
|
cwd=fixture.root, check=True, capture_output=True,
|
)
|
subprocess.run(["git", "commit", "-m", "local only"], cwd=fixture.root, check=True, capture_output=True)
|
later = datetime(2026, 8, 29, 8, 30, tzinfo=timezone.utc)
|
local_outbox = product.reconcile(fixture.config, later)["created_outbox_ids"][0]
|
before = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as unpublished:
|
product.git_deliver(fixture.config, local_outbox, later)
|
self.assertEqual("E_GIT_PUSH", unpublished.exception.code)
|
self.assertEqual(before, fixture.config.outbox_path.read_bytes())
|
|
def test_git_no_changes_rejects_terminal_boundary_artifact_and_ref_drift(self) -> None:
|
for mode, expected_code in (
|
("artifact_after_read", "E_GIT_SCOPE"),
|
("head_after_capture", "E_GIT_PUSH"),
|
("remote_after_read", "E_GIT_PUSH"),
|
):
|
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
for command in (
|
["git", "init", "-b", "main"],
|
["git", "config", "user.name", "Fixture"],
|
["git", "config", "user.email", "fixture@example.invalid"],
|
["git", "add", "."],
|
["git", "commit", "-m", "fixture baseline"],
|
):
|
subprocess.run(command, cwd=fixture.root, check=True, capture_output=True)
|
remote = fixture.root / "remote.git"
|
subprocess.run(["git", "init", "--bare", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "push", "-u", "origin", "main"], cwd=fixture.root, check=True, capture_output=True)
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
artifact = fixture.archive / "终态竞态.txt"
|
artifact.write_bytes(b"published artifact\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": f"race-{mode}", "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": artifact.name, "bytes": artifact.stat().st_size,
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
subprocess.run(
|
["git", "add", "--", artifact.relative_to(fixture.root).as_posix(), fixture.formal.relative_to(fixture.root).as_posix()],
|
cwd=fixture.root, check=True, capture_output=True,
|
)
|
subprocess.run(["git", "commit", "-m", "accepted published artifact"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "push", "origin", "main"], cwd=fixture.root, check=True, capture_output=True)
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
outbox_before = fixture.config.outbox_path.read_bytes()
|
injected = False
|
|
def runner(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[Any]:
|
nonlocal injected
|
result = subprocess.run(command, **kwargs)
|
if not injected and mode in {"artifact_after_read", "head_after_capture"} and command[:2] == ["git", "show"]:
|
injected = True
|
if mode == "artifact_after_read":
|
identity = artifact.stat()
|
payload = artifact.read_bytes()
|
artifact.write_bytes(b"X" + payload[1:])
|
os.utime(artifact, ns=(identity.st_atime_ns, identity.st_mtime_ns))
|
else:
|
old_head = subprocess.run(
|
["git", "rev-parse", "HEAD"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip()
|
tree = subprocess.run(
|
["git", "rev-parse", "HEAD^{tree}"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip()
|
env = dict(os.environ)
|
env.update({
|
"GIT_AUTHOR_NAME": "Race", "GIT_COMMITTER_NAME": "Race",
|
"GIT_AUTHOR_EMAIL": "race@example.invalid", "GIT_COMMITTER_EMAIL": "race@example.invalid",
|
})
|
new_head = subprocess.run(
|
["git", "commit-tree", tree, "-p", old_head, "-m", "same tree local race"],
|
cwd=fixture.root, check=True, text=True, capture_output=True, env=env,
|
).stdout.strip()
|
subprocess.run(["git", "update-ref", "HEAD", new_head, old_head], cwd=fixture.root, check=True, capture_output=True)
|
elif (
|
not injected
|
and mode == "remote_after_read"
|
and command[:3] == ["git", "rev-parse", "refs/remotes/origin/main"]
|
):
|
injected = True
|
old_remote = result.stdout.strip()
|
parent = subprocess.run(
|
["git", "rev-parse", "HEAD^"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip()
|
subprocess.run(
|
["git", "update-ref", "refs/remotes/origin/main", parent, old_remote],
|
cwd=fixture.root, check=True, capture_output=True,
|
)
|
return result
|
|
with self.assertRaises(product.PipelineError) as drift:
|
product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)
|
self.assertTrue(injected)
|
self.assertEqual(expected_code, drift.exception.code)
|
self.assertEqual(outbox_before, fixture.config.outbox_path.read_bytes())
|
self.assertEqual(0, len(list(fixture.config.state_dir.glob(f"git-index-{outbox_id}*"))))
|
self.assertEqual(0, len(list((fixture.root / ".git").rglob("*.lock"))))
|
|
def test_append_only_rewrite_and_unsafe_artifact_fail_closed(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
fixture.formal.write_bytes(b"")
|
with self.assertRaises(product.PipelineError) as rewritten:
|
product.reconcile(fixture.config, NOW)
|
self.assertEqual("E_HISTORY_REWRITE", rewritten.exception.code)
|
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
original = fixture.formal.read_bytes()
|
replacement = original.replace(b'"stable_id":"OLD"', b'"stable_id":"NEW"')
|
self.assertEqual(len(original), len(replacement))
|
fixture.formal.write_bytes(replacement)
|
with self.assertRaises(product.PipelineError) as prefix:
|
product.reconcile(fixture.config, NOW)
|
self.assertEqual("E_HISTORY_REWRITE", prefix.exception.code)
|
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
outside = fixture.root / "outside.txt"
|
outside.write_bytes(b"outside")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "123", "item_type": "text", "status": "SAVED", "creator_uid": UID,
|
"path": str(outside), "bytes": outside.stat().st_size, "sha256": hashlib.sha256(outside.read_bytes()).hexdigest().upper()
|
})
|
with self.assertRaises(product.PipelineError) as escaped:
|
product.reconcile(fixture.config, NOW)
|
self.assertEqual("E_ARTIFACT", escaped.exception.code)
|
|
def test_git_commit_journal_recovers_push_without_second_commit(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
(fixture.root / ".git").mkdir()
|
(fixture.root / ".git" / "index").write_bytes(b"USER-INDEX")
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
artifact = fixture.archive / "recover.txt"
|
artifact.write_bytes(b"recover\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "recover", "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": "recover.txt", "bytes": artifact.stat().st_size,
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
rel = artifact.relative_to(fixture.root).as_posix()
|
commands: list[list[str]] = []
|
push_count = 0
|
|
def runner(command: list[str], **_: object) -> subprocess.CompletedProcess[str]:
|
nonlocal push_count
|
commands.append(command)
|
if command[:5] == ["git", "diff", "--cached", "--name-only", "-z"] and command[5] != "--":
|
return subprocess.CompletedProcess(command, 0, b"", b"")
|
if command[:6] == ["git", "diff", "--cached", "--name-only", "-z", "--"]:
|
return subprocess.CompletedProcess(command, 0, rel.encode("utf-8") + b"\0", b"")
|
if command[:4] == ["git", "diff", "--cached", "--name-only"]:
|
return subprocess.CompletedProcess(command, 0, "", "")
|
if command[:3] == ["git", "rev-parse", "HEAD"]:
|
return subprocess.CompletedProcess(command, 0, ("a" if push_count == 0 else "c") * 40 + "\n", "")
|
if command[:2] == ["git", "hash-object"]:
|
return subprocess.CompletedProcess(command, 0, "d" * 40 + "\n", "")
|
if command[:2] == ["git", "write-tree"]:
|
return subprocess.CompletedProcess(command, 0, "b" * 40 + "\n", "")
|
if command[:2] == ["git", "commit-tree"]:
|
return subprocess.CompletedProcess(command, 0, "c" * 40 + "\n", "")
|
if command[:2] == ["git", "push"]:
|
push_count += 1
|
return subprocess.CompletedProcess(command, 1 if push_count == 1 else 0, "", "")
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
with self.assertRaises(product.PipelineError) as first:
|
product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)
|
self.assertEqual("E_GIT_PUSH", first.exception.code)
|
self.assertEqual("GIT_PUSHED", product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)["status"])
|
self.assertEqual(1, sum(1 for command in commands if command[:2] == ["git", "commit-tree"]))
|
self.assertEqual(1, sum(1 for command in commands if command[:2] == ["git", "update-index"]))
|
|
def test_payload_source_rebinding_and_signed_url_reject_before_dispatch(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": BVID, "item_type": "video", "status": "VIDEO_DOWNLOAD_PENDING_EXTENSION",
|
"creator_uid": UID, "source_url": f"https://www.bilibili.com/video/{BVID}?token=SYNTHETIC_SECRET",
|
"title": "fixture", "published_at": NOW.isoformat(), "expected_duration_seconds": 60,
|
})
|
with self.assertRaises(product.PipelineError) as secret:
|
product.reconcile(fixture.config, NOW)
|
self.assertEqual("E_SECRET_FIELD", secret.exception.code)
|
self.assertFalse(fixture.config.outbox_path.exists())
|
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
self._new_handoff(fixture)
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
rows = [json.loads(line) for line in fixture.config.outbox_path.read_text(encoding="ascii").splitlines()]
|
rows[0]["payload"]["bvid"] = "BV1Q541167Qf"
|
fixture.config.outbox_path.write_bytes(b"".join(canonical(row) for row in rows))
|
before = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as tampered:
|
product.dispatch_intent(fixture.config, outbox_id, NOW)
|
self.assertEqual("E_OUTBOX", tampered.exception.code)
|
self.assertEqual(before, fixture.config.outbox_path.read_bytes())
|
|
def test_legacy_audit_handoff_identity_key_is_narrowly_projected(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": BVID, "item_type": "video",
|
"status": "VIDEO_DOWNLOAD_PENDING_EXTENSION", "creator_uid": UID,
|
"source_url": f"https://www.bilibili.com/video/{BVID}",
|
"title": "fixture", "published_at": NOW.isoformat(),
|
"expected_duration_seconds": 60,
|
"runtime_authorization_handoff_id": "HANDOFF-LEGACY-AUDIT-IDENTITY",
|
})
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
pending = product.pending(fixture.config)
|
self.assertEqual((1, outbox_id), (pending["count"], pending["items"][0]["outbox_id"]))
|
serialized = fixture.config.outbox_path.read_text(encoding="ascii").lower()
|
self.assertNotIn("authorization", serialized)
|
self.assertEqual(
|
"DISPATCH_INTENT_DURABLE",
|
product.dispatch_intent(fixture.config, outbox_id, NOW)["status"],
|
)
|
|
def test_receipt_requires_observed_and_late_observed_is_mutation_zero(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
self._new_handoff(fixture)
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
product.dispatch_intent(fixture.config, outbox_id, NOW)
|
transcript = fixture.archive / "ordered.transcript.txt"
|
transcript.write_bytes(b"ordered\n")
|
receipt = fixture.root / "receipt.json"
|
receipt.write_bytes(canonical({
|
"schema_version": 1, "type": "TRANSCRIPTION_COMPLETE", "outbox_id": outbox_id,
|
"stable_id": BVID, "terminal_id": "TRANSCRIPT-ORDER", "status": "COMPLETE",
|
"files": [{"path": str(transcript), "bytes": transcript.stat().st_size,
|
"sha256": hashlib.sha256(transcript.read_bytes()).hexdigest().upper(), "kind": "transcript"}],
|
"created_at": NOW.isoformat(),
|
}))
|
outbox_before = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as early:
|
product.ingest_receipt(fixture.config, receipt, NOW)
|
self.assertEqual("E_RECEIPT", early.exception.code)
|
self.assertEqual(outbox_before, fixture.config.outbox_path.read_bytes())
|
self.assertFalse(fixture.config.terminals_path.exists())
|
product._append(fixture.config.outbox_path, {
|
"schema_version": 1, "event": "COMPLETE", "outbox_id": outbox_id,
|
"result": "TRANSCRIPTION_COMPLETE", "terminal_id": "TRANSCRIPT-ORDER",
|
"receipt_sha256": hashlib.sha256(receipt.read_bytes()).hexdigest().upper(), "completed_at": NOW.isoformat(),
|
})
|
malformed = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as late:
|
product.observe_dispatch(fixture.config, outbox_id, "DELIVERY-LATE", NOW)
|
self.assertEqual("E_OUTBOX", late.exception.code)
|
self.assertEqual(malformed, fixture.config.outbox_path.read_bytes())
|
|
def test_git_commit_failure_uses_only_task_index_and_retry_is_deterministic(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
(fixture.root / ".git").mkdir()
|
shared_index = fixture.root / ".git" / "index"
|
shared_index.write_bytes(b"USER-INDEX")
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
artifact = fixture.archive / "failure.txt"
|
artifact.write_bytes(b"failure\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "failure", "item_type": "text", "status": "SAVED", "creator_uid": UID,
|
"path": "failure.txt", "bytes": artifact.stat().st_size, "sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
rel = artifact.relative_to(fixture.root).as_posix()
|
calls: list[tuple[list[str], dict[str, object]]] = []
|
commit_attempt = 0
|
|
def runner(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
|
nonlocal commit_attempt
|
calls.append((command, kwargs))
|
if command[:5] == ["git", "diff", "--cached", "--name-only", "-z"] and command[5] != "--":
|
return subprocess.CompletedProcess(command, 0, b"", b"")
|
if command[:6] == ["git", "diff", "--cached", "--name-only", "-z", "--"]:
|
return subprocess.CompletedProcess(command, 0, rel.encode("utf-8") + b"\0", b"")
|
if command[:4] == ["git", "diff", "--cached", "--name-only"]:
|
return subprocess.CompletedProcess(command, 0, "", "")
|
if command[:3] == ["git", "rev-parse", "HEAD"]:
|
return subprocess.CompletedProcess(command, 0, "a" * 40 + "\n", "")
|
if command[:2] == ["git", "hash-object"]:
|
return subprocess.CompletedProcess(command, 0, "d" * 40 + "\n", "")
|
if command[:2] == ["git", "write-tree"]:
|
return subprocess.CompletedProcess(command, 0, "b" * 40 + "\n", "")
|
if command[:2] == ["git", "commit-tree"]:
|
commit_attempt += 1
|
return subprocess.CompletedProcess(command, 1 if commit_attempt == 1 else 0, "" if commit_attempt == 1 else "c" * 40 + "\n", "")
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
with self.assertRaises(product.PipelineError) as failed:
|
product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)
|
self.assertEqual("E_GIT_COMMIT", failed.exception.code)
|
self.assertEqual(b"USER-INDEX", shared_index.read_bytes())
|
self.assertEqual(1, fixture.config.outbox_path.read_text(encoding="ascii").count('"event":"GIT_COMMIT_INTENT"'))
|
self.assertEqual("GIT_PUSHED", product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)["status"])
|
self.assertEqual(b"USER-INDEX", shared_index.read_bytes())
|
self.assertEqual(1, sum(command[:2] == ["git", "update-index"] for command, _ in calls))
|
task_envs = [kwargs["env"] for command, kwargs in calls if command[:2] == ["git", "update-index"]]
|
self.assertTrue(all(str(env["GIT_INDEX_FILE"]).startswith(str(fixture.config.state_dir)) for env in task_envs))
|
|
def test_real_git_commit_failure_preserves_shared_index_and_exact_retry_pushes(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
for command in (
|
["git", "init", "-b", "main"],
|
["git", "config", "user.name", "Fixture"],
|
["git", "config", "user.email", "fixture@example.invalid"],
|
["git", "add", "."],
|
["git", "commit", "-m", "fixture baseline"],
|
):
|
subprocess.run(command, cwd=fixture.root, check=True, capture_output=True)
|
remote = fixture.root / "remote.git"
|
subprocess.run(["git", "init", "--bare", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
artifact = fixture.archive / "真实-交付.txt"
|
artifact.write_bytes(b"real failure\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "real-failure", "item_type": "text", "status": "SAVED", "creator_uid": UID,
|
"path": artifact.name, "bytes": artifact.stat().st_size,
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
shared_index = fixture.root / ".git" / "index"
|
index_before = shared_index.read_bytes()
|
failed_once = False
|
|
def runner(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
|
nonlocal failed_once
|
if command[:2] == ["git", "commit-tree"] and not failed_once:
|
failed_once = True
|
return subprocess.CompletedProcess(command, 1, "", "injected")
|
return subprocess.run(command, **kwargs)
|
|
with self.assertRaises(product.PipelineError) as failure:
|
product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)
|
self.assertEqual("E_GIT_COMMIT", failure.exception.code)
|
self.assertEqual(index_before, shared_index.read_bytes())
|
staged = subprocess.run(["git", "diff", "--cached", "--name-only"], cwd=fixture.root, check=True, text=True, capture_output=True)
|
self.assertEqual("", staged.stdout)
|
result = product.git_deliver(fixture.config, outbox_id, NOW, runner=runner)
|
self.assertEqual("GIT_PUSHED", result["status"])
|
self.assertEqual(index_before, shared_index.read_bytes())
|
remote_head = subprocess.run(
|
["git", "--git-dir", str(remote), "rev-parse", "refs/heads/main"], check=True, text=True, capture_output=True,
|
).stdout.strip()
|
self.assertEqual(result["commit_sha"], remote_head)
|
|
def test_real_git_batch_freezes_shared_index_across_moving_head_and_rejects_drift(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
for command in (
|
["git", "init", "-b", "main"],
|
["git", "config", "user.name", "Fixture"],
|
["git", "config", "user.email", "fixture@example.invalid"],
|
["git", "add", "."],
|
["git", "commit", "-m", "fixture baseline"],
|
):
|
subprocess.run(command, cwd=fixture.root, check=True, capture_output=True)
|
remote = fixture.root / "remote.git"
|
subprocess.run(["git", "init", "--bare", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
user_file = fixture.root / "user-staged.txt"
|
user_file.write_bytes(b"user staged preimage\n")
|
subprocess.run(["git", "add", user_file.name], cwd=fixture.root, check=True, capture_output=True)
|
shared_index = fixture.root / ".git" / "index"
|
index_before = shared_index.read_bytes()
|
baseline_head = subprocess.run(
|
["git", "rev-parse", "HEAD"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip()
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
for stable_id in ("batch-one", "batch-two"):
|
artifact = fixture.archive / f"{stable_id}.txt"
|
artifact.write_bytes((stable_id + "\n").encode("ascii"))
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": stable_id, "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": artifact.name, "bytes": artifact.stat().st_size,
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
outbox_ids = product.reconcile(fixture.config, NOW)["created_outbox_ids"]
|
self.assertEqual(2, len(outbox_ids))
|
first = product.git_deliver(fixture.config, outbox_ids[0], NOW)
|
self.assertEqual(index_before, shared_index.read_bytes())
|
interpreted_after_head_move = subprocess.run(
|
["git", "diff", "--cached", "--name-only"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.splitlines()
|
self.assertIn("user-staged.txt", interpreted_after_head_move)
|
guards = list(fixture.config.state_dir.glob("git-shared-index-guard-*.json"))
|
self.assertEqual(1, len(guards))
|
guards[0].unlink()
|
outbox_before_recovery = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as wrong_frozen_bytes:
|
product.recover_git_index_guard(
|
fixture.config,
|
outbox_ids[1],
|
baseline_head,
|
len(index_before),
|
"0" * 64,
|
)
|
self.assertEqual("E_GIT_INDEX_DIRTY", wrong_frozen_bytes.exception.code)
|
self.assertEqual(outbox_before_recovery, fixture.config.outbox_path.read_bytes())
|
self.assertEqual([], list(fixture.config.state_dir.glob("git-shared-index-guard-*.json")))
|
recovered = product.recover_git_index_guard(
|
fixture.config,
|
outbox_ids[1],
|
baseline_head,
|
len(index_before),
|
hashlib.sha256(index_before).hexdigest().upper(),
|
)
|
self.assertEqual("GIT_INDEX_GUARD_RECOVERED", recovered["status"])
|
second = product.git_deliver(fixture.config, outbox_ids[1], NOW)
|
self.assertEqual(index_before, shared_index.read_bytes())
|
self.assertNotEqual(first["commit_sha"], second["commit_sha"])
|
self.assertEqual([], list(fixture.config.state_dir.glob("git-index-*")))
|
remote_head = subprocess.run(
|
["git", "--git-dir", str(remote), "rev-parse", "refs/heads/main"], check=True, text=True, capture_output=True,
|
).stdout.strip()
|
self.assertEqual(second["commit_sha"], remote_head)
|
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
for command in (
|
["git", "init", "-b", "main"],
|
["git", "config", "user.name", "Fixture"],
|
["git", "config", "user.email", "fixture@example.invalid"],
|
["git", "add", "."],
|
["git", "commit", "-m", "fixture baseline"],
|
):
|
subprocess.run(command, cwd=fixture.root, check=True, capture_output=True)
|
remote = fixture.root / "remote.git"
|
subprocess.run(["git", "init", "--bare", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
for stable_id in ("drift-one", "drift-two"):
|
artifact = fixture.archive / f"{stable_id}.txt"
|
artifact.write_bytes((stable_id + "\n").encode("ascii"))
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": stable_id, "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": artifact.name, "bytes": artifact.stat().st_size,
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
outbox_ids = product.reconcile(fixture.config, NOW)["created_outbox_ids"]
|
first = product.git_deliver(fixture.config, outbox_ids[0], NOW)
|
user_file = fixture.root / "late-user-stage.txt"
|
user_file.write_bytes(b"late user stage\n")
|
subprocess.run(["git", "add", user_file.name], cwd=fixture.root, check=True, capture_output=True)
|
outbox_before = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as drift:
|
product.git_deliver(fixture.config, outbox_ids[1], NOW)
|
self.assertEqual("E_GIT_INDEX_DIRTY", drift.exception.code)
|
self.assertEqual(outbox_before, fixture.config.outbox_path.read_bytes())
|
self.assertEqual(first["commit_sha"], subprocess.run(
|
["git", "rev-parse", "HEAD"], cwd=fixture.root, check=True, text=True, capture_output=True,
|
).stdout.strip())
|
self.assertEqual([], list(fixture.config.state_dir.glob("git-index-*")))
|
|
def test_existing_git_guard_recovery_revalidates_guard_and_shared_index(self) -> None:
|
baseline_head = "a" * 40
|
|
def prepare(raw: str) -> tuple[PipelineFixture, Path, Path, str, bytes, dict[str, bytes]]:
|
fixture = PipelineFixture(Path(raw))
|
(fixture.root / ".git").mkdir()
|
shared_index = fixture.root / ".git" / "index"
|
index_bytes = b"INDEX-A"
|
shared_index.write_bytes(index_bytes)
|
product.initialize(fixture.config, NOW)
|
product.begin(fixture.config, NOW)
|
artifact = fixture.archive / "guarded.txt"
|
artifact.write_bytes(b"guarded\n")
|
fixture.append_formal({
|
"schema_version": 1, "stable_id": "guarded", "item_type": "text", "status": "SAVED",
|
"creator_uid": UID, "path": artifact.name, "bytes": artifact.stat().st_size,
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest().upper(),
|
})
|
outbox_id = product.reconcile(fixture.config, NOW)["created_outbox_ids"][0]
|
staged = {"value": b"user-staged.txt\0"}
|
|
def runner(command: list[str], **_: object) -> subprocess.CompletedProcess[object]:
|
if command[:3] == ["git", "rev-parse", "HEAD"]:
|
return subprocess.CompletedProcess(command, 0, baseline_head + "\n", "")
|
if command[:5] == ["git", "diff", "--cached", "--name-only", "-z"]:
|
return subprocess.CompletedProcess(command, 0, staged["value"], b"")
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
rows = product._outbox_rows(fixture.config)
|
item = next(row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == outbox_id)
|
guard = product._ensure_git_index_guard(fixture.config, rows, item, runner)
|
guard_path = fixture.config.git_index_guard_path(guard["batch_id"])
|
fixture.guard_runner = runner # type: ignore[attr-defined]
|
return fixture, shared_index, guard_path, outbox_id, index_bytes, staged
|
|
with tempfile.TemporaryDirectory() as raw:
|
fixture, _, _, outbox_id, index_bytes, _ = prepare(raw)
|
result = product.recover_git_index_guard(
|
fixture.config, outbox_id, baseline_head, len(index_bytes), hashlib.sha256(index_bytes).hexdigest().upper(),
|
runner=fixture.guard_runner, # type: ignore[attr-defined]
|
)
|
self.assertEqual("GIT_INDEX_GUARD_ALREADY_DURABLE", result["status"])
|
|
for case in ("bytes", "staged_paths", "file_identity", "guard_field"):
|
with self.subTest(case=case), tempfile.TemporaryDirectory() as raw:
|
fixture, shared_index, guard_path, outbox_id, index_bytes, staged = prepare(raw)
|
if case == "bytes":
|
shared_index.write_bytes(b"INDEX-B")
|
elif case == "staged_paths":
|
staged["value"] = b"other-user-staged.txt\0"
|
elif case == "file_identity":
|
replacement = shared_index.with_name("index-replacement")
|
replacement.write_bytes(index_bytes)
|
os.replace(replacement, shared_index)
|
else:
|
guard = json.loads(guard_path.read_text(encoding="ascii"))
|
guard["outbox_ids"] = ["F" * 64]
|
guard_path.write_bytes(canonical(guard))
|
outbox_before = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as rejected:
|
product.recover_git_index_guard(
|
fixture.config, outbox_id, baseline_head, len(index_bytes),
|
hashlib.sha256(index_bytes).hexdigest().upper(),
|
runner=fixture.guard_runner, # type: ignore[attr-defined]
|
)
|
self.assertEqual("E_GIT_INDEX_DIRTY", rejected.exception.code)
|
self.assertEqual(outbox_before, fixture.config.outbox_path.read_bytes())
|
|
for broken in (False, True):
|
with self.subTest(symlink_broken=broken), tempfile.TemporaryDirectory() as raw:
|
fixture, _, guard_path, outbox_id, index_bytes, _ = prepare(raw)
|
target = guard_path.with_name("guard-link-target.json")
|
if not broken:
|
target.write_bytes(guard_path.read_bytes())
|
guard_path.unlink()
|
try:
|
os.symlink(target, guard_path)
|
except OSError:
|
continue
|
outbox_before = fixture.config.outbox_path.read_bytes()
|
with self.assertRaises(product.PipelineError) as rejected:
|
product.recover_git_index_guard(
|
fixture.config, outbox_id, baseline_head, len(index_bytes),
|
hashlib.sha256(index_bytes).hexdigest().upper(),
|
runner=fixture.guard_runner, # type: ignore[attr-defined]
|
)
|
self.assertEqual("E_GIT_INDEX_DIRTY", rejected.exception.code)
|
self.assertEqual(outbox_before, fixture.config.outbox_path.read_bytes())
|
|
def test_full_chain_reparse_and_identity_drift_fail_closed(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
artifact = fixture.archive / "stable.txt"
|
artifact.write_bytes(b"stable\n")
|
digest = hashlib.sha256(artifact.read_bytes()).hexdigest().upper()
|
original = product._strict_chain
|
calls = 0
|
|
def drifting(root: Path, target: Path, *, final_file: bool) -> tuple[tuple[str, tuple[int, ...]], ...]:
|
nonlocal calls
|
calls += 1
|
value = original(root, target, final_file=final_file)
|
if calls == 2:
|
mutated = list(value)
|
name, identity = mutated[-2]
|
mutated[-2] = (name, identity[:-1] + (identity[-1] + 1,))
|
return tuple(mutated)
|
return value
|
|
with mock.patch.object(product, "_strict_chain", side_effect=drifting):
|
with self.assertRaises(product.PipelineError) as drift:
|
product._artifact("stable.txt", artifact.stat().st_size, digest, fixture.config)
|
self.assertEqual("E_ARTIFACT", drift.exception.code)
|
outside = fixture.root / "outside.txt"
|
outside.write_bytes(b"outside")
|
link = fixture.archive / "linked.txt"
|
try:
|
os.symlink(outside, link)
|
except OSError:
|
return
|
with self.assertRaises(product.PipelineError) as reparse:
|
product._artifact("linked.txt", outside.stat().st_size, hashlib.sha256(outside.read_bytes()).hexdigest().upper(), fixture.config)
|
self.assertEqual("E_ARTIFACT", reparse.exception.code)
|
|
def test_finish_allows_next_slot_and_cli_is_ascii_only(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
product.initialize(fixture.config, NOW)
|
first = product.begin(fixture.config, NOW)
|
finished = product.finish(fixture.config, "COMPLETE", NOW)
|
self.assertEqual(first["run"]["run_id"], finished["run_id"])
|
second = product.begin(fixture.config, datetime(2026, 8, 29, 8, 30, tzinfo=timezone.utc))
|
self.assertNotEqual(first["run"]["run_id"], second["run"]["run_id"])
|
code, result = product.run(["--config", str(fixture.config_path), "pending"])
|
self.assertEqual((0, "PENDING"), (code, result["status"]))
|
product._canonical(result).decode("ascii")
|
|
def test_git_preflight_detects_index_regression_without_mutating_shared_index(self) -> None:
|
class Interrupted(BaseException):
|
pass
|
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
subprocess.run(["git", "init", "-b", "main"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.name", "Fixture"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.email", "fixture@example.invalid"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "add", "."], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "commit", "-m", "baseline"], cwd=fixture.root, check=True, capture_output=True)
|
relative = (fixture.archive / "old.txt").relative_to(fixture.root).as_posix()
|
subprocess.run(["git", "rm", "--cached", "--", relative], cwd=fixture.root, check=True, capture_output=True)
|
index_path = fixture.root / ".git" / "index"
|
lock_path = fixture.root / ".git" / "index.lock"
|
lock_path.write_bytes(b"")
|
before = index_path.read_bytes()
|
before_stat = os.stat(index_path)
|
result = product.git_preflight(fixture.config)
|
self.assertFalse(result["index_matches_head"])
|
self.assertEqual([relative], result["archive_staged_delete_present"])
|
self.assertEqual((0, hashlib.sha256(b"").hexdigest().upper()), (
|
result["stale_index_lock"]["bytes"], result["stale_index_lock"]["sha256"]
|
))
|
self.assertEqual(before, index_path.read_bytes())
|
self.assertEqual((before_stat.st_dev, before_stat.st_ino), (os.stat(index_path).st_dev, os.stat(index_path).st_ino))
|
|
def fail_runner(*_args: object, **_kwargs: object) -> object:
|
raise RuntimeError("synthetic read failure")
|
|
with self.assertRaises(RuntimeError):
|
product.git_preflight(fixture.config, runner=fail_runner) # type: ignore[arg-type]
|
self.assertEqual(before, index_path.read_bytes())
|
|
def interrupt_runner(*_args: object, **_kwargs: object) -> object:
|
raise Interrupted()
|
|
with self.assertRaises(Interrupted):
|
product.git_preflight(fixture.config, runner=interrupt_runner) # type: ignore[arg-type]
|
self.assertEqual(before, index_path.read_bytes())
|
|
def test_git_preflight_uses_one_held_index_during_transient_path_swap(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
subprocess.run(["git", "init", "-b", "main"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.name", "Fixture"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.email", "fixture@example.invalid"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "add", "."], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "commit", "-m", "baseline"], cwd=fixture.root, check=True, capture_output=True)
|
index_path = fixture.root / ".git" / "index"
|
alternate = fixture.root / ".git" / "alternate-index"
|
parked = fixture.root / ".git" / "original-index-held-test"
|
shutil.copyfile(index_path, alternate)
|
relative = (fixture.archive / "old.txt").relative_to(fixture.root).as_posix()
|
alternate_env = {**os.environ, "GIT_INDEX_FILE": str(alternate)}
|
subprocess.run(
|
["git", "rm", "--cached", "--", relative],
|
cwd=fixture.root,
|
env=alternate_env,
|
check=True,
|
capture_output=True,
|
)
|
alternate_staged = subprocess.run(
|
["git", "diff", "--cached", "--name-only", "HEAD", "--"],
|
cwd=fixture.root,
|
env=alternate_env,
|
check=True,
|
text=True,
|
capture_output=True,
|
).stdout.splitlines()
|
self.assertEqual([relative], alternate_staged)
|
before = index_path.read_bytes()
|
before_stat = os.stat(index_path)
|
race = {"attempted": False, "performed": False, "denied": False}
|
|
def swap_runner(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[object]:
|
if not race["attempted"] and command[1:2] == ["diff"]:
|
race["attempted"] = True
|
try:
|
os.replace(index_path, parked)
|
os.replace(alternate, index_path)
|
race["performed"] = True
|
except OSError:
|
race["denied"] = True
|
try:
|
return subprocess.run(command, **kwargs) # type: ignore[arg-type]
|
finally:
|
if race["performed"]:
|
os.replace(index_path, alternate)
|
os.replace(parked, index_path)
|
return subprocess.run(command, **kwargs) # type: ignore[arg-type]
|
|
result = product.git_preflight(fixture.config, runner=swap_runner) # type: ignore[arg-type]
|
self.assertTrue(race["attempted"])
|
self.assertTrue(race["performed"] or race["denied"])
|
self.assertTrue(result["index_matches_head"])
|
self.assertEqual([], result["staged_paths"])
|
self.assertEqual([], result["archive_staged_delete_present"])
|
self.assertEqual(before, index_path.read_bytes())
|
self.assertEqual(
|
(before_stat.st_dev, before_stat.st_ino),
|
(os.stat(index_path).st_dev, os.stat(index_path).st_ino),
|
)
|
|
def test_migration_restarts_from_one_durable_report_at_every_write_boundary(self) -> None:
|
class Interrupted(BaseException):
|
pass
|
|
boundaries = [
|
*(f"move-{index}" for index in range(0, 7)),
|
"rmdir-1",
|
"rmdir-2",
|
"outbox",
|
]
|
for boundary in boundaries:
|
with self.subTest(boundary=boundary), tempfile.TemporaryDirectory() as raw:
|
fixture, legacy_paths, payloads, flac = prepare_migration_fixture(Path(raw))
|
initial_plan = product.plan_video_artifact_migration(fixture.config, NOW)
|
original_move = product._move_relocation_file
|
original_rmdir = Path.rmdir
|
original_append = product._append_outbox
|
move_calls = 0
|
rmdir_calls = 0
|
|
def interrupted_move(*args: object, **kwargs: object) -> None:
|
nonlocal move_calls
|
if boundary == "move-0" and move_calls == 0:
|
raise Interrupted()
|
original_move(*args, **kwargs)
|
move_calls += 1
|
if boundary == f"move-{move_calls}":
|
raise Interrupted()
|
|
def interrupted_rmdir(path: Path) -> None:
|
nonlocal rmdir_calls
|
original_rmdir(path)
|
if path.name in {f"{BVID}.transcript", f"{BVID}.minutes"}:
|
rmdir_calls += 1
|
if boundary == f"rmdir-{rmdir_calls}":
|
raise Interrupted()
|
|
def interrupted_append(*args: object, **kwargs: object) -> str:
|
value = original_append(*args, **kwargs)
|
if boundary == "outbox":
|
raise Interrupted()
|
return value
|
|
with (
|
mock.patch.object(product, "_move_relocation_file", side_effect=interrupted_move),
|
mock.patch.object(Path, "rmdir", new=interrupted_rmdir),
|
mock.patch.object(product, "_append_outbox", side_effect=interrupted_append),
|
):
|
with self.assertRaises(Interrupted):
|
product.migrate_video_artifacts(fixture.config, NOW)
|
reports = list(fixture.config.relocation_root.glob("*.json"))
|
self.assertEqual(1, len(reports))
|
self.assertEqual(initial_plan, product._read_relocation_report(fixture.config, reports[0]))
|
|
completed = product.migrate_video_artifacts(
|
fixture.config,
|
datetime(2026, 8, 29, 8, 1, tzinfo=timezone.utc),
|
)
|
self.assertEqual(initial_plan["batch_id"], completed["batch_id"])
|
self.assertEqual(initial_plan["created_at"], product._read_relocation_report(fixture.config, reports[0])["created_at"])
|
rows = product._outbox_rows(fixture.config)
|
created = [
|
row for row in rows
|
if row.get("event") == "CREATED" and row.get("outbox_id") == completed["git_outbox_id"]
|
]
|
self.assertEqual(1, len(created))
|
for alias in initial_plan["items"][0]["aliases"]:
|
self.assertFalse((fixture.root / alias["old_path"]).exists())
|
self.assertEqual(payloads[alias["kind"]], (fixture.root / alias["new_path"]).read_bytes())
|
target_flac = fixture.root / "external-video" / "intermediate" / "transcription" / f"{BVID}.audio.flac"
|
self.assertFalse(flac.exists())
|
self.assertEqual(b"fLaC-fixture", target_flac.read_bytes())
|
self.assertFalse((fixture.archive / f"{BVID}.transcript").exists())
|
self.assertFalse((fixture.archive / f"{BVID}.minutes").exists())
|
self.assertTrue(all(not path.exists() for path in legacy_paths.values()))
|
|
def test_relocation_source_replacement_cannot_delete_replacement_bytes(self) -> None:
|
source_tmp_root = PROJECT_ROOT / "dev" / "tmp"
|
source_tmp_root.mkdir(parents=True, exist_ok=True)
|
target_parents = [source_tmp_root]
|
if Path("F:/").is_dir():
|
target_parents.append(Path("F:/"))
|
for target_parent in target_parents:
|
with (
|
self.subTest(target_parent=str(target_parent)),
|
tempfile.TemporaryDirectory(dir=source_tmp_root) as source_raw,
|
tempfile.TemporaryDirectory(dir=target_parent) as target_raw,
|
):
|
source_root = Path(source_raw)
|
target_root = Path(target_raw)
|
source = source_root / "source.bin"
|
target = target_root / "target.bin"
|
parked = source_root / "parked-original.bin"
|
source.write_bytes(b"AAAA")
|
original_delete = product._delete_held_relocation_source
|
race = {"swapped": False, "denied": False}
|
|
def replace_before_delete(*args: object, **kwargs: object) -> None:
|
try:
|
os.replace(source, parked)
|
source.write_bytes(b"BBBB")
|
race["swapped"] = True
|
except OSError:
|
race["denied"] = True
|
original_delete(*args, **kwargs)
|
|
if os.name == "nt":
|
with mock.patch.object(product, "_delete_held_relocation_source", side_effect=replace_before_delete):
|
product._move_relocation_file(
|
source_root,
|
target_root,
|
source,
|
target,
|
4,
|
hashlib.sha256(b"AAAA").hexdigest().upper(),
|
)
|
self.assertTrue(race["denied"])
|
self.assertFalse(race["swapped"])
|
self.assertFalse(source.exists())
|
else:
|
with (
|
mock.patch.object(product, "_delete_held_relocation_source", side_effect=replace_before_delete),
|
self.assertRaises(product.PipelineError) as rejected,
|
):
|
product._move_relocation_file(
|
source_root,
|
target_root,
|
source,
|
target,
|
4,
|
hashlib.sha256(b"AAAA").hexdigest().upper(),
|
)
|
self.assertEqual("E_RELOCATION", rejected.exception.code)
|
self.assertTrue(race["swapped"])
|
self.assertEqual(b"BBBB", source.read_bytes())
|
self.assertEqual(b"AAAA", parked.read_bytes())
|
self.assertEqual(b"AAAA", target.read_bytes())
|
|
def test_relocation_report_is_rebound_to_the_exact_external_plan(self) -> None:
|
mutations = (
|
"missing_intermediate", "missing_remove", "cross_path", "extra_alias",
|
"omitted_legacy_id", "omitted_canonical_id",
|
)
|
for mutation in mutations:
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as raw:
|
fixture, legacy_paths, payloads, flac = prepare_migration_fixture(Path(raw))
|
report = json.loads(json.dumps(product.plan_video_artifact_migration(fixture.config, NOW)))
|
if mutation == "missing_intermediate":
|
report["items"][0]["intermediates"] = []
|
elif mutation == "missing_remove":
|
report["remove_paths"] = report["remove_paths"][1:]
|
elif mutation == "cross_path":
|
report["items"][0]["aliases"][0]["old_path"] = fixture.formal.relative_to(fixture.root).as_posix()
|
elif mutation == "extra_alias":
|
report["items"][0]["aliases"].append(dict(report["items"][0]["aliases"][0]))
|
else:
|
omitted = "BV1Q541167Qf"
|
omitted_title = "Omitted legacy item"
|
omitted_published = "2026-08-31T01:02:03+08:00"
|
fixture.append_formal({
|
"schema_version": 1,
|
"stable_id": omitted,
|
"item_type": "video",
|
"status": product.VIDEO_COMPLETE,
|
"creator": "fixture",
|
"title": omitted_title,
|
"published_at": omitted_published,
|
})
|
omitted_base = omitted if mutation == "omitted_legacy_id" else product._canonical_video_base(
|
omitted,
|
omitted_title,
|
omitted_published,
|
)
|
omitted_dir = fixture.archive / f"{omitted_base}.transcript"
|
omitted_dir.mkdir()
|
(omitted_dir / f"{omitted_base}.txt").write_bytes(b"omitted\n")
|
(omitted_dir / f"{omitted_base}.srt").write_bytes(b"1\n00:00:00,000 --> 00:00:01,000\nomitted\n")
|
(omitted_dir / f"{omitted_base}.json").write_bytes(b'{"segments":[]}\n')
|
report["batch_id"] = product._relocation_batch_id(report)
|
fixture.config.relocation_root.mkdir(parents=True)
|
forged_path = fixture.config.relocation_root / f"{report['batch_id']}.json"
|
forged_path.write_bytes(canonical(report))
|
state_before = fixture.config.state_path.read_bytes()
|
formal_before = fixture.formal.read_bytes()
|
with self.assertRaises(product.PipelineError) as rejected:
|
product.migrate_video_artifacts(fixture.config, NOW)
|
self.assertEqual("E_RELOCATION", rejected.exception.code)
|
self.assertEqual(state_before, fixture.config.state_path.read_bytes())
|
self.assertEqual(formal_before, fixture.formal.read_bytes())
|
self.assertFalse(fixture.config.outbox_path.exists())
|
self.assertEqual(b"fLaC-fixture", flac.read_bytes())
|
for kind, path in legacy_paths.items():
|
self.assertEqual(payloads[kind], path.read_bytes())
|
canonical_base = product._canonical_video_base(BVID, "A / canonical: title?", "2026-08-30T12:34:56+08:00")
|
self.assertFalse((fixture.archive / f"{canonical_base}.transcript").exists())
|
self.assertFalse((fixture.archive / f"{canonical_base}.minutes").exists())
|
|
def test_video_artifact_migration_is_byte_exact_alias_resolvable_and_git_scoped(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
fixture = PipelineFixture(Path(raw))
|
(fixture.root / "external-video").mkdir()
|
stable_id = BVID
|
title = "A / canonical: title?"
|
published_at = "2026-08-30T12:34:56+08:00"
|
fixture.append_formal({
|
"schema_version": 1,
|
"stable_id": stable_id,
|
"item_type": "video",
|
"status": product.VIDEO_COMPLETE,
|
"creator_uid": UID,
|
"title": title,
|
"published_at": published_at,
|
})
|
transcript_dir = fixture.archive / f"{stable_id}.transcript"
|
minutes_dir = fixture.archive / f"{stable_id}.minutes"
|
transcript_dir.mkdir()
|
minutes_dir.mkdir()
|
payloads = {
|
"transcript_txt": b"transcript\n",
|
"transcript_srt": b"1\n00:00:00,000 --> 00:00:01,000\ntext\n",
|
"transcript_json": b'{"segments":[]}\n',
|
"minutes_md": b"# minutes\n",
|
"minutes_pdf": b"%PDF-fixture\n",
|
}
|
legacy_paths = {
|
"transcript_txt": transcript_dir / f"{stable_id}.txt",
|
"transcript_srt": transcript_dir / f"{stable_id}.srt",
|
"transcript_json": transcript_dir / f"{stable_id}.json",
|
"minutes_md": minutes_dir / f"{stable_id}.md",
|
"minutes_pdf": minutes_dir / f"{stable_id}.pdf",
|
}
|
for kind, path in legacy_paths.items():
|
path.write_bytes(payloads[kind])
|
flac = transcript_dir / f"{stable_id}.audio.flac"
|
flac.write_bytes(b"fLaC-fixture")
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.name", "Fixture"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "config", "user.email", "fixture@example.invalid"], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "add", "."], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "commit", "-m", "legacy baseline"], cwd=fixture.root, check=True, capture_output=True)
|
remote = fixture.root / "remote.git"
|
subprocess.run(["git", "init", "--bare", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=fixture.root, check=True, capture_output=True)
|
subprocess.run(["git", "push", "-u", "origin", "main"], cwd=fixture.root, check=True, capture_output=True)
|
(fixture.archive / "目录导读.md").write_bytes(b"# canonical naming guide\n")
|
product.initialize(fixture.config, NOW)
|
index_path = fixture.root / ".git" / "index"
|
index_before = index_path.read_bytes()
|
|
plan = product.plan_video_artifact_migration(fixture.config, NOW)
|
expected_base = "20260830-123456_video_A _ canonical_ title__BV1Q541167Qg"
|
self.assertEqual(expected_base, plan["items"][0]["canonical_base"])
|
result = product.migrate_video_artifacts(fixture.config, NOW)
|
self.assertEqual(("VIDEO_ARTIFACTS_MIGRATED", 1, 5, 1), (
|
result["status"], result["item_count"], result["public_file_count"], result["intermediate_count"]
|
))
|
self.assertEqual(index_before, index_path.read_bytes())
|
for alias in plan["items"][0]["aliases"]:
|
old_path = fixture.root / alias["old_path"]
|
new_path = fixture.root / alias["new_path"]
|
self.assertFalse(old_path.exists())
|
self.assertEqual(payloads[alias["kind"]], new_path.read_bytes())
|
rebound = product._artifact(alias["old_path"], alias["bytes"], alias["sha256"], fixture.config)
|
self.assertEqual(alias["old_path"], rebound["path"])
|
external_flac = fixture.root / "external-video" / "intermediate" / "transcription" / f"{stable_id}.audio.flac"
|
self.assertEqual(b"fLaC-fixture", external_flac.read_bytes())
|
self.assertFalse(flac.exists())
|
report_path = fixture.root / result["report"]["path"]
|
self.assertTrue(report_path.is_file())
|
rerun = product.migrate_video_artifacts(fixture.config, NOW)
|
self.assertEqual(result["git_outbox_id"], rerun["git_outbox_id"])
|
|
delivered = product.git_deliver(fixture.config, result["git_outbox_id"], NOW)
|
self.assertEqual("GIT_PUSHED", delivered["status"])
|
self.assertEqual(index_before, index_path.read_bytes())
|
changed = subprocess.run(
|
["git", "diff-tree", "--no-commit-id", "--name-status", "-r", "HEAD"],
|
cwd=fixture.root,
|
check=True,
|
text=True,
|
capture_output=True,
|
).stdout
|
for alias in plan["items"][0]["aliases"]:
|
self.assertIn(alias["old_path"], changed)
|
self.assertIn(alias["new_path"], changed)
|
self.assertNotIn(".flac", changed.lower())
|
local_head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=fixture.root, check=True, text=True, capture_output=True).stdout.strip()
|
remote_head = subprocess.run(
|
["git", "--git-dir", str(remote), "rev-parse", "refs/heads/main"],
|
check=True,
|
text=True,
|
capture_output=True,
|
).stdout.strip()
|
self.assertEqual(local_head, remote_head)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
