from __future__ import annotations
|
|
import hashlib
|
import importlib.util
|
import io
|
import json
|
import os
|
import sys
|
import tempfile
|
import unittest
|
from contextlib import redirect_stdout
|
from datetime import datetime
|
from pathlib import Path
|
from unittest import mock
|
|
|
MODULE_PATH = Path(__file__).resolve().parents[1] / "bili_dynamic_collector.py"
|
SPEC = importlib.util.spec_from_file_location("bili_dynamic_collector", MODULE_PATH)
|
assert SPEC and SPEC.loader
|
collector = importlib.util.module_from_spec(SPEC)
|
sys.modules[SPEC.name] = collector
|
SPEC.loader.exec_module(collector)
|
|
FIXTURE = Path(__file__).resolve().parent / "fixtures" / "bili_dynamic_collector" / "dynamic_items.json"
|
NOW = "2026-08-04T12:00:00+08:00"
|
|
|
class BiliDynamicCollectorTests(unittest.TestCase):
|
def setUp(self) -> None:
|
self.temp = tempfile.TemporaryDirectory()
|
self.root = Path(self.temp.name)
|
self.config_path = self.root / "config.json"
|
self.state_dir = self.root / "state"
|
self.download_dir = self.root / "downloads"
|
self.video_dir = self.root / "videos"
|
self.download_dir.mkdir()
|
self.video_dir.mkdir()
|
self.write_json(
|
self.config_path,
|
{
|
"schema_version": 1,
|
"creator": {
|
"name": "青枫浦上Q",
|
"dynamic_url": "https://space.bilibili.com/1420210197/dynamic",
|
},
|
"timezone": "Asia/Shanghai",
|
"window_hours": 72,
|
"minimum_complete_age_seconds": 0,
|
"title_max_length": 48,
|
"allowed_source_hosts": [
|
"space.bilibili.com",
|
"www.bilibili.com",
|
"t.bilibili.com",
|
"b23.tv",
|
],
|
"allowed_video_extensions": [".mp4", ".mkv", ".mov", ".webm"],
|
"native_handoff": {
|
"project_id": "project-info",
|
"source_ai_id": "video-downloader",
|
"source_thread_id": "019fcc5d-798f-7ea1-8325-3a4d1f2dc5a5",
|
"source_role_instance_id": "case_analysis.video_downloader",
|
"target_ai_id": "media-processor",
|
"target_thread_id": "019fb7a4-bdfd-79f2-bd6b-e67e2b7d8efd",
|
"target_role_instance_id": "case_analysis.media_processor",
|
"reply_thread_id": "019fcc5d-798f-7ea1-8325-3a4d1f2dc5a5",
|
},
|
"paths": {
|
"state_dir": "state",
|
"download_dir": "downloads",
|
"video_dir": "videos",
|
},
|
},
|
)
|
|
def tearDown(self) -> None:
|
self.temp.cleanup()
|
|
@staticmethod
|
def write_json(path: Path, value: object) -> None:
|
path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
|
|
def run_check(self) -> tuple[int, dict[str, object]]:
|
return collector.run(
|
["--config", str(self.config_path), "check", "--input", str(FIXTURE), "--now", NOW]
|
)
|
|
def run_main_json(self, argv: list[str]) -> tuple[int, dict[str, object], str]:
|
stream = io.StringIO()
|
with redirect_stdout(stream):
|
code = collector.main(argv)
|
text = stream.getvalue()
|
return code, json.loads(text), text
|
|
def assert_lock_quiescent(self) -> None:
|
lock = self.state_dir / ".collector.lock"
|
self.assertTrue(lock.is_file())
|
self.assertEqual(b"\0", lock.read_bytes())
|
self.assertFalse((self.state_dir / ".collector.lock.owner.json").exists())
|
|
def queue_video(self) -> dict[str, object]:
|
code, result = self.run_check()
|
self.assertEqual(0, code)
|
self.assertEqual("TODO_GENERATED", result["status"])
|
manifest = collector.load_manifest(self.state_dir / "manifest.jsonl")
|
return next(event for event in manifest if event["content_type"] == "video")
|
|
def test_windows_name_is_safe_and_preserves_timestamp(self) -> None:
|
self.assertEqual("_CON", collector.sanitize_windows_component("CON", 48))
|
self.assertEqual("a_b_c", collector.sanitize_windows_component('a<b>c. ', 48))
|
config = collector.load_config(self.config_path)
|
item = collector.normalize_item(
|
{
|
"dynamic_id": "d1",
|
"content_type": "text",
|
"published_at": "2026-08-04T10:00:00+08:00",
|
"title": "午盘: A/B?",
|
"source_url": "https://www.bilibili.com/opus/d1",
|
},
|
config,
|
0,
|
)
|
self.assertEqual("20260804-100000_文字_午盘_ A_B_", collector.suggested_base(item, config))
|
|
def test_check_filters_window_and_is_idempotent(self) -> None:
|
code, first = self.run_check()
|
self.assertEqual(0, code)
|
self.assertEqual("TODO_GENERATED", first["status"])
|
self.assertEqual(2, first["new_items"])
|
self.assertEqual(1, first["skipped_old"])
|
self.assertEqual(1, first["skipped_future"])
|
queue_path = Path(first["queue_path"])
|
self.assertTrue(queue_path.is_file())
|
queue = json.loads(queue_path.read_text(encoding="utf-8"))
|
self.assertEqual(2, len(queue["items"]))
|
self.assertTrue(queue["browser_interaction_required"])
|
self.assertFalse(queue["authentication_data_allowed"])
|
self.assertEqual(2, len(collector.load_manifest(self.state_dir / "manifest.jsonl")))
|
|
code, second = self.run_check()
|
self.assertEqual(0, code)
|
self.assertEqual("NO_NEW_ITEMS", second["status"])
|
self.assertEqual(0, second["new_items"])
|
self.assertIsNone(second["queue_path"])
|
self.assertEqual([queue_path], list((self.state_dir / "queues").glob("*.json")))
|
self.assertEqual(2, len(collector.load_manifest(self.state_dir / "manifest.jsonl")))
|
|
def test_safe_move_hash_manifest_and_handoff_are_idempotent(self) -> None:
|
queued = self.queue_video()
|
source = self.download_dir / "extension-result.mp4"
|
payload = b"small synthetic completed video fixture"
|
source.write_bytes(payload)
|
completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
|
os.utime(source, (completed_mtime, completed_mtime))
|
mapping = self.root / "mapping.json"
|
self.write_json(
|
mapping,
|
{
|
"schema_version": 1,
|
"items": [{"source_file": source.name, "bvid": "BV1TEST00001"}],
|
},
|
)
|
|
code, moved = collector.run(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(0, code)
|
self.assertEqual("COMPLETE", moved["status"])
|
self.assertFalse(source.exists())
|
target = Path(moved["items"][0]["target"])
|
self.assertEqual(f"{queued['suggested_stem']}.mp4", target.name)
|
self.assertEqual(payload, target.read_bytes())
|
expected_hash = hashlib.sha256(payload).hexdigest()
|
self.assertEqual(expected_hash, moved["items"][0]["sha256"])
|
latest, _ = collector.latest_entities(collector.load_manifest(self.state_dir / "manifest.jsonl"))
|
event = latest[queued["entity_id"]]
|
self.assertEqual("VIDEO_MOVED", event["status"])
|
self.assertEqual("READY_FOR_HANDOFF", event["video_processing_status"])
|
self.assertEqual(expected_hash, event["sha256"])
|
|
manifest_before_rerun = (self.state_dir / "manifest.jsonl").read_bytes()
|
code, moved_again = collector.run(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(0, code)
|
self.assertEqual("ALREADY_MOVED", moved_again["items"][0]["status"])
|
self.assertEqual(expected_hash, moved_again["items"][0]["sha256"])
|
self.assertEqual(manifest_before_rerun, (self.state_dir / "manifest.jsonl").read_bytes())
|
self.assertEqual(payload, target.read_bytes())
|
|
code, handoff = collector.run(
|
["--config", str(self.config_path), "handoff", "--now", NOW]
|
)
|
self.assertEqual(0, code)
|
self.assertEqual("GENERATED", handoff["write"])
|
draft = Path(handoff["handoff_path"])
|
text = draft.read_text(encoding="utf-8")
|
self.assertIn("message_type=video_processing_request", text)
|
self.assertIn("status=PROCESSING_REQUESTED", text)
|
self.assertRegex(text, r"handoff_id=HANDOFF-BILI-DYNAMIC-VIDEO-PROCESSING-[0-9A-F]{24}")
|
for field in (
|
"source_ai_id=video-downloader",
|
"source_thread_id=019fcc5d-798f-7ea1-8325-3a4d1f2dc5a5",
|
"source_role_instance_id=case_analysis.video_downloader",
|
"target_ai_id=media-processor",
|
"target_thread_id=019fb7a4-bdfd-79f2-bd6b-e67e2b7d8efd",
|
"target_role_instance_id=case_analysis.media_processor",
|
"reply_thread_id=019fcc5d-798f-7ea1-8325-3a4d1f2dc5a5",
|
"scope:",
|
"evidence:",
|
"expected_action:",
|
):
|
self.assertIn(field, text)
|
self.assertNotIn("generated_at=", text)
|
self.assertIn(json.dumps(str(target), ensure_ascii=False), text)
|
self.assertNotIn("cookie", text.lower())
|
|
code, repeated = collector.run(
|
["--config", str(self.config_path), "handoff", "--now", "2026-08-04T13:00:00+08:00"]
|
)
|
self.assertEqual(0, code)
|
self.assertEqual("REUSED", repeated["write"])
|
self.assertEqual(draft, Path(repeated["handoff_path"]))
|
|
target.write_bytes(b"tampered after manifest")
|
config = collector.load_config(self.config_path)
|
with collector.StateLock(config.lock_path):
|
with self.assertRaises(collector.CollectorError) as caught:
|
collector.generate_handoff(config, None, collector.parse_now(NOW))
|
self.assertEqual("E_HANDOFF_HASH", caught.exception.code)
|
|
def test_existing_target_preflight_has_zero_business_side_effects(self) -> None:
|
queued = self.queue_video()
|
source = self.download_dir / "extension-result.mp4"
|
source.write_bytes(b"new download")
|
completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
|
os.utime(source, (completed_mtime, completed_mtime))
|
target = self.video_dir / f"{queued['suggested_stem']}.mp4"
|
target.write_bytes(b"existing unrelated file")
|
mapping = self.root / "mapping.json"
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": source.name, "dynamic_id": "dynamic-video-001"}]},
|
)
|
manifest_before = (self.state_dir / "manifest.jsonl").read_bytes()
|
config = collector.load_config(self.config_path)
|
with collector.StateLock(config.lock_path):
|
with self.assertRaisesRegex(collector.CollectorError, "refusing to overwrite") as caught:
|
collector.move_completed(config, mapping, collector.parse_now(NOW))
|
self.assertEqual("E_TARGET_EXISTS", caught.exception.code)
|
self.assertEqual(b"existing unrelated file", target.read_bytes())
|
self.assertEqual(b"new download", source.read_bytes())
|
self.assertEqual(manifest_before, (self.state_dir / "manifest.jsonl").read_bytes())
|
latest, _ = collector.latest_entities(collector.load_manifest(self.state_dir / "manifest.jsonl"))
|
self.assertEqual("TODO_QUEUED", latest[queued["entity_id"]]["status"])
|
|
code, retry = self.run_check()
|
self.assertEqual(0, code)
|
self.assertEqual("NO_NEW_ITEMS", retry["status"])
|
|
def test_incomplete_or_outside_download_is_a_safety_stop(self) -> None:
|
queued = self.queue_video()
|
incomplete = self.download_dir / "clip.part"
|
incomplete.write_bytes(b"partial")
|
mapping = self.root / "mapping.json"
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": incomplete.name, "bvid": queued["bvid"]}]},
|
)
|
code = collector.main(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(3, code)
|
self.assertTrue(incomplete.exists())
|
|
outside = self.root / "outside.mp4"
|
outside.write_bytes(b"outside")
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": str(outside), "bvid": queued["bvid"]}]},
|
)
|
code = collector.main(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(3, code)
|
self.assertTrue(outside.exists())
|
|
def test_exact_selectors_and_batch_uniqueness_fail_before_business_writes(self) -> None:
|
queued = self.queue_video()
|
events = collector.load_manifest(self.state_dir / "manifest.jsonl")
|
text_event = next(event for event in events if event["content_type"] == "text")
|
source = self.download_dir / "one.mp4"
|
source.write_bytes(b"one physical source")
|
completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
|
os.utime(source, (completed_mtime, completed_mtime))
|
manifest_path = self.state_dir / "manifest.jsonl"
|
|
cases = [
|
(
|
"unknown supplied selector",
|
[{"source_file": source.name, "bvid": queued["bvid"], "dynamic_id": "unknown-dynamic"}],
|
"E_MAPPING_SELECTOR_UNKNOWN",
|
),
|
(
|
"selectors disagree",
|
[
|
{
|
"source_file": source.name,
|
"bvid": queued["bvid"],
|
"dynamic_id": text_event["dynamic_id"],
|
}
|
],
|
"E_MAPPING_SELECTOR_CONFLICT",
|
),
|
(
|
"entity repeated",
|
[
|
{"source_file": source.name, "bvid": queued["bvid"]},
|
{"source_file": source.name, "dynamic_id": queued["dynamic_id"]},
|
],
|
"E_ENTITY_DUPLICATE",
|
),
|
]
|
for label, items, expected_code in cases:
|
with self.subTest(label=label):
|
mapping = self.root / "mapping.json"
|
self.write_json(mapping, {"schema_version": 1, "items": items})
|
manifest_before = manifest_path.read_bytes()
|
code, payload, _ = self.run_main_json(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(3, code)
|
self.assertEqual(expected_code, payload["error_code"])
|
self.assertEqual(manifest_before, manifest_path.read_bytes())
|
self.assertEqual(b"one physical source", source.read_bytes())
|
self.assertEqual([], list(self.video_dir.iterdir()))
|
|
second = dict(queued)
|
second.update(
|
{
|
"event_id": "second-video-event",
|
"entity_id": "second-video-entity",
|
"dynamic_id": "dynamic-video-002",
|
"bvid": "BV1TEST00002",
|
"source_url": "https://www.bilibili.com/video/BV1TEST00002",
|
"dedupe_keys": [
|
"bvid:bv1test00002",
|
"dynamic:dynamic-video-002",
|
"url:https://www.bilibili.com/video/BV1TEST00002",
|
],
|
"suggested_stem": str(queued["suggested_stem"]) + "_second",
|
}
|
)
|
collector.append_manifest(manifest_path, [second])
|
mapping = self.root / "same-source.json"
|
self.write_json(
|
mapping,
|
{
|
"schema_version": 1,
|
"items": [
|
{"source_file": source.name, "bvid": queued["bvid"]},
|
{"source_file": str(source), "bvid": second["bvid"]},
|
],
|
},
|
)
|
manifest_before = manifest_path.read_bytes()
|
code, payload, _ = self.run_main_json(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(3, code)
|
self.assertEqual("E_SOURCE_DUPLICATE", payload["error_code"])
|
self.assertEqual(manifest_before, manifest_path.read_bytes())
|
self.assertEqual(b"one physical source", source.read_bytes())
|
self.assertEqual([], list(self.video_dir.iterdir()))
|
|
def test_retained_source_warning_is_sticky_with_or_without_source(self) -> None:
|
queued = self.queue_video()
|
source = self.download_dir / "retained.mp4"
|
payload_bytes = b"retained synthetic bytes"
|
source.write_bytes(payload_bytes)
|
completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
|
os.utime(source, (completed_mtime, completed_mtime))
|
mapping = self.root / "mapping.json"
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
|
)
|
path_class = type(source)
|
original_unlink = path_class.unlink
|
|
def fail_only_source(path: Path, *args: object, **kwargs: object) -> None:
|
if path == source:
|
raise PermissionError("injected source delete failure")
|
original_unlink(path, *args, **kwargs)
|
|
with mock.patch.object(path_class, "unlink", fail_only_source):
|
code, first = collector.run(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(4, code)
|
self.assertEqual("VIDEO_MOVED_SOURCE_RETAINED", first["items"][0]["status"])
|
target = Path(first["items"][0]["target"])
|
expected_hash = hashlib.sha256(payload_bytes).hexdigest()
|
self.assertEqual(expected_hash, first["items"][0]["sha256"])
|
self.assertTrue(source.exists())
|
manifest_before = (self.state_dir / "manifest.jsonl").read_bytes()
|
|
code, second = collector.run(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(4, code)
|
self.assertEqual("VIDEO_MOVED_SOURCE_RETAINED", second["items"][0]["status"])
|
self.assertEqual(first["items"][0]["source_delete_error"], second["items"][0]["source_delete_error"])
|
self.assertEqual(manifest_before, (self.state_dir / "manifest.jsonl").read_bytes())
|
self.assertEqual(payload_bytes, target.read_bytes())
|
|
source.unlink()
|
code, third = collector.run(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(4, code)
|
self.assertEqual("VIDEO_MOVED_SOURCE_RETAINED", third["items"][0]["status"])
|
self.assertEqual(expected_hash, third["items"][0]["sha256"])
|
self.assertEqual(first["items"][0]["source_delete_error"], third["items"][0]["source_delete_error"])
|
self.assertEqual(manifest_before, (self.state_dir / "manifest.jsonl").read_bytes())
|
self.assertEqual(payload_bytes, target.read_bytes())
|
|
def test_disallowed_latest_statuses_fail_before_stage_b(self) -> None:
|
queued = self.queue_video()
|
source = self.download_dir / "sentinel.mp4"
|
source.write_bytes(b"sentinel source")
|
mapping = self.root / "mapping.json"
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
|
)
|
manifest_path = self.state_dir / "manifest.jsonl"
|
status_cases: list[object] = [
|
"PROCESSING_HANDOFF_CONFIRMED",
|
"PROCESSING",
|
"PROCESSING_FAILED",
|
None,
|
"",
|
"UNKNOWN_STATE",
|
]
|
for status in status_cases:
|
with self.subTest(status=status):
|
event = dict(queued)
|
if status is None:
|
event.pop("status", None)
|
else:
|
event["status"] = status
|
manifest_path.write_bytes(collector.canonical_json_bytes(event))
|
before = {
|
str(path.relative_to(self.root)): path.read_bytes()
|
for path in self.root.rglob("*")
|
if path.is_file()
|
}
|
code, payload, _ = self.run_main_json(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
after = {
|
str(path.relative_to(self.root)): path.read_bytes()
|
for path in self.root.rglob("*")
|
if path.is_file()
|
}
|
self.assertEqual(3, code)
|
self.assertEqual("SAFETY_STOP", payload["status"])
|
self.assertEqual("E_STATUS", payload["error_code"])
|
self.assertEqual(before, after)
|
self.assertEqual([], list(self.video_dir.iterdir()))
|
self.assert_lock_quiescent()
|
|
def test_manifest_secret_gate_is_shared_by_all_commands(self) -> None:
|
queued = self.queue_video()
|
source = self.download_dir / "secret-check.mp4"
|
source.write_bytes(b"source remains")
|
mapping = self.root / "mapping.json"
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
|
)
|
manifest_path = self.state_dir / "manifest.jsonl"
|
base_event = dict(queued)
|
commands = [
|
["check", "--input", str(FIXTURE), "--now", NOW],
|
["move-completed", "--mapping", str(mapping), "--now", NOW],
|
["handoff", "--now", NOW],
|
]
|
for index, command in enumerate(commands):
|
with self.subTest(command=command[0]):
|
event = dict(base_event)
|
if index == 0:
|
event["cookie"] = "forbidden-manifest-value"
|
else:
|
event["nested"] = {"token": "forbidden-manifest-value"}
|
manifest_path.write_bytes(collector.canonical_json_bytes(event))
|
before = manifest_path.read_bytes()
|
code, payload, stdout = self.run_main_json(["--config", str(self.config_path), *command])
|
self.assertEqual(3, code)
|
self.assertEqual("E_SECRET_FIELD", payload["error_code"])
|
self.assertNotIn("forbidden-manifest-value", stdout)
|
self.assertEqual(before, manifest_path.read_bytes())
|
self.assertEqual(b"source remains", source.read_bytes())
|
self.assertEqual([], list(self.video_dir.iterdir()))
|
|
def test_terminal_projection_rejects_missing_blank_or_malformed_evidence(self) -> None:
|
queued = self.queue_video()
|
mapping = self.root / "terminal-mapping.json"
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": "source-is-absent.mp4", "bvid": queued["bvid"]}]},
|
)
|
manifest_path = self.state_dir / "manifest.jsonl"
|
valid_target = str(self.video_dir / "persisted-target.mp4")
|
cases = [
|
("moved missing local", "VIDEO_MOVED", "local_file", None, True),
|
("moved blank local", "VIDEO_MOVED", "local_file", " ", False),
|
("moved control local", "VIDEO_MOVED", "local_file", "bad\npath", False),
|
("moved missing sha", "VIDEO_MOVED", "sha256", None, True),
|
("moved blank sha", "VIDEO_MOVED", "sha256", "", False),
|
("moved malformed sha", "VIDEO_MOVED", "sha256", "A" * 64, False),
|
("retained missing reason", "VIDEO_MOVED_SOURCE_RETAINED", "failure_reason", None, True),
|
("retained blank reason", "VIDEO_MOVED_SOURCE_RETAINED", "failure_reason", " ", False),
|
("retained control reason", "VIDEO_MOVED_SOURCE_RETAINED", "failure_reason", "bad\rreason", False),
|
]
|
for label, status, field, value, remove_field in cases:
|
with self.subTest(label=label):
|
event = dict(queued)
|
event.update(
|
{
|
"status": status,
|
"local_file": valid_target,
|
"sha256": "a" * 64,
|
"failure_reason": "PermissionError: retained source",
|
"video_processing_status": "READY_FOR_HANDOFF",
|
}
|
)
|
if remove_field:
|
event.pop(field, None)
|
else:
|
event[field] = value
|
manifest_path.write_bytes(collector.canonical_json_bytes(event))
|
before = {
|
str(path.relative_to(self.root)): path.read_bytes()
|
for path in self.root.rglob("*")
|
if path.is_file()
|
}
|
code, payload, stdout = self.run_main_json(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
after = {
|
str(path.relative_to(self.root)): path.read_bytes()
|
for path in self.root.rglob("*")
|
if path.is_file()
|
}
|
self.assertEqual(3, code)
|
self.assertEqual("SAFETY_STOP", payload["status"])
|
self.assertEqual("E_MANIFEST", payload["error_code"])
|
self.assertEqual(before, after)
|
self.assertNotIn("bad\npath", stdout)
|
self.assertNotIn("bad\rreason", stdout)
|
if isinstance(value, str) and value.strip():
|
self.assertNotIn(value, stdout)
|
self.assertEqual([], list(self.video_dir.iterdir()))
|
self.assert_lock_quiescent()
|
|
def test_handoff_rejects_control_or_invalid_manifest_scalars_without_writes(self) -> None:
|
queued = self.queue_video()
|
source = self.download_dir / "handoff-source.mp4"
|
source_bytes = b"handoff scalar validation bytes"
|
source.write_bytes(source_bytes)
|
completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
|
os.utime(source, (completed_mtime, completed_mtime))
|
mapping = self.root / "handoff-mapping.json"
|
self.write_json(
|
mapping,
|
{"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
|
)
|
code, moved = collector.run(
|
["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
|
)
|
self.assertEqual(0, code)
|
target = Path(moved["items"][0]["target"])
|
latest, _ = collector.latest_entities(collector.load_manifest(self.state_dir / "manifest.jsonl"))
|
moved_event = dict(latest[queued["entity_id"]])
|
manifest_path = self.state_dir / "manifest.jsonl"
|
cases = [
|
("entity newline", "entity_id", str(moved_event["entity_id"]) + "\nstatus=FORGED"),
|
("entity format", "entity_id", "not-an-internal-id"),
|
("bvid newline", "bvid", str(moved_event["bvid"]) + "\r\nstatus=FORGED"),
|
("bvid format", "bvid", "NOT-A-BVID"),
|
("published newline", "published_at", str(moved_event["published_at"]) + "\nstatus=FORGED"),
|
("published noncanonical", "published_at", "2026-08-04T03:00:00+00:00"),
|
("title newline", "title", str(moved_event["title"]) + "\nstatus=FORGED"),
|
("source url newline", "source_url", str(moved_event["source_url"]) + "\nstatus=FORGED"),
|
("local file newline", "local_file", str(moved_event["local_file"]) + "\nstatus=FORGED"),
|
("sha newline", "sha256", str(moved_event["sha256"]) + "\nstatus=FORGED"),
|
]
|
for label, field, value in cases:
|
with self.subTest(label=label):
|
event = dict(moved_event)
|
event[field] = value
|
manifest_path.write_bytes(collector.canonical_json_bytes(event))
|
before = {
|
str(path.relative_to(self.root)): path.read_bytes()
|
for path in self.root.rglob("*")
|
if path.is_file()
|
}
|
code, payload, stdout = self.run_main_json(
|
["--config", str(self.config_path), "handoff", "--now", NOW]
|
)
|
after = {
|
str(path.relative_to(self.root)): path.read_bytes()
|
for path in self.root.rglob("*")
|
if path.is_file()
|
}
|
self.assertEqual(3, code)
|
self.assertEqual("SAFETY_STOP", payload["status"])
|
self.assertEqual("E_MANIFEST", payload["error_code"])
|
self.assertNotIn("status=FORGED", stdout)
|
self.assertNotIn(value, stdout)
|
self.assertEqual(before, after)
|
self.assertEqual(source_bytes, target.read_bytes())
|
self.assert_lock_quiescent()
|
|
def test_native_handoff_route_is_required_and_validated_before_state_write(self) -> None:
|
base = json.loads(self.config_path.read_text(encoding="utf-8"))
|
for label, mutate in (
|
("missing field", lambda route: route.pop("target_thread_id")),
|
("invalid thread", lambda route: route.__setitem__("target_thread_id", "NOT-A-THREAD")),
|
("envelope injection", lambda route: route.__setitem__("target_ai_id", "media-processor\nstatus=FORGED")),
|
):
|
with self.subTest(label=label):
|
invalid = json.loads(json.dumps(base))
|
mutate(invalid["native_handoff"])
|
config = self.root / f"invalid-{label}.json"
|
self.write_json(config, invalid)
|
self.assertFalse(self.state_dir.exists())
|
code, payload, _ = self.run_main_json(
|
["--config", str(config), "check", "--input", str(FIXTURE), "--now", NOW]
|
)
|
self.assertEqual(2, code)
|
self.assertEqual("E_CONFIG", payload["error_code"])
|
self.assertFalse(self.state_dir.exists())
|
|
def test_secret_fields_are_rejected_and_not_persisted(self) -> None:
|
exported = json.loads(FIXTURE.read_text(encoding="utf-8"))
|
exported["cookie"] = "forbidden-value"
|
bad = self.root / "bad.json"
|
self.write_json(bad, exported)
|
code = collector.main(
|
["--config", str(self.config_path), "check", "--input", str(bad), "--now", NOW]
|
)
|
self.assertEqual(3, code)
|
self.assertFalse((self.state_dir / "manifest.jsonl").exists())
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
