from __future__ import annotations
|
|
import hashlib
|
import importlib.util
|
import json
|
import gc
|
import os
|
import shutil
|
import subprocess
|
import sys
|
import tempfile
|
import time
|
import unittest
|
from pathlib import Path
|
|
|
PROJECT_DEV = Path(__file__).resolve().parents[1]
|
if str(PROJECT_DEV) not in sys.path:
|
sys.path.insert(0, str(PROJECT_DEV))
|
COLLECTOR_PATH = PROJECT_DEV / "bili_dynamic_collector.py"
|
SPEC = importlib.util.spec_from_file_location("bili_dynamic_collector", COLLECTOR_PATH)
|
assert SPEC and SPEC.loader
|
collector = importlib.util.module_from_spec(SPEC)
|
sys.modules[SPEC.name] = collector
|
SPEC.loader.exec_module(collector)
|
import bili_dynamic_refresh as refresh
|
import bili_dynamic_refresh_controller as controller
|
|
|
NOW = "2026-08-13T10:00:00+08:00"
|
|
|
class BiliDynamicRefreshTests(unittest.TestCase):
|
def setUp(self) -> None:
|
self.temp = tempfile.TemporaryDirectory()
|
self.root = Path(self.temp.name)
|
self.state = self.root / "state"
|
self.archive = self.root / "archive"
|
self.intake = self.root / "intake"
|
self.downloads = self.root / "downloads"
|
self.videos = self.root / "videos"
|
for path in (self.archive, self.intake, self.downloads, self.videos):
|
path.mkdir()
|
self.config_path = self.root / "config.json"
|
self.write_json(
|
self.config_path,
|
{
|
"schema_version": 1,
|
"creator": {
|
"name": "青枫浦上Q",
|
"uid": "1420210197",
|
"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"},
|
"refresh": {
|
"archive_dir": "archive",
|
"formal_manifest": "archive/manifest.jsonl",
|
"intake_dir": "intake",
|
"overall_deadline_seconds": 120,
|
"refresh_action_timeout_seconds": 35,
|
"observation_timeout_seconds": 45,
|
"page_internal_settle_timeout_seconds": 15,
|
"max_refresh_count": 1,
|
"run_history_slots": 168,
|
"max_items": 200,
|
"max_images_per_item": 20,
|
"max_image_bytes": 20971520,
|
"max_text_bytes": 2097152,
|
},
|
},
|
)
|
|
def tearDown(self) -> None:
|
self.temp.cleanup()
|
|
def write_json(self, path: Path, value: object) -> None:
|
if isinstance(value, dict) and value.get("schema_version") == 3 and isinstance(value.get("controller_attestation"), dict):
|
config = collector.load_config(self.config_path)
|
pending = refresh._load_pending(config)
|
if pending is not None:
|
attestation = value["controller_attestation"]
|
refresh._attest_controller_evidence(
|
pending, value,
|
action_dispatched=bool(attestation["action_dispatched"]),
|
monotonic_run_started_ms=int(attestation["monotonic_run_started_ms"]),
|
monotonic_action_started_ms=attestation["monotonic_action_started_ms"],
|
monotonic_action_finished_ms=attestation["monotonic_action_finished_ms"],
|
monotonic_observation_started_ms=attestation["monotonic_observation_started_ms"],
|
monotonic_observation_finished_ms=attestation["monotonic_observation_finished_ms"],
|
monotonic_evidence_write_started_ms=int(attestation["monotonic_evidence_write_started_ms"]),
|
)
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
|
|
def reset_runtime_fixture(self) -> None:
|
"""Reset only disposable per-test runtime/formal trees between matrix rows."""
|
for path in (self.state, self.archive, self.intake):
|
if path.exists():
|
shutil.rmtree(path)
|
path.mkdir()
|
|
def begin(self) -> dict[str, object]:
|
code, result = collector.run(["--config", str(self.config_path), "refresh-begin", "--now", NOW])
|
self.assertEqual(0, code)
|
self.assertEqual("BROWSER_REFRESH_REQUIRED", result["status"])
|
return result
|
|
def formal_pending(self, config: collector.CollectorConfig) -> dict[str, object]:
|
return {
|
"schema_version": 2,
|
"run_id": "a" * 32,
|
"owner_nonce": "b" * 32,
|
"phase": "TRANSACTION_INTENT",
|
"task_id": refresh.TASK_ID,
|
"creator_uid": config.creator_uid,
|
"creator_dynamic_url": config.creator_dynamic_url,
|
"started_at": "2026-08-13T02:00:00Z",
|
"deadline_at": "2026-08-13T02:00:30Z",
|
"window_start": "2026-08-10T02:00:00Z",
|
"window_end": "2026-08-13T02:00:00Z",
|
"config_sha256": hashlib.sha256(self.config_path.read_bytes()).hexdigest(),
|
"state_manifest_preimage": refresh._identity(config.manifest_path).as_dict(),
|
"formal_manifest_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(),
|
"evidence_path": str(config.state_dir / "refresh" / "incoming" / ("a" * 32 + ".json")),
|
"intake_root": str(config.refresh.intake_dir / ("a" * 32)),
|
"evidence_identity": {"bytes": 2, "sha256": hashlib.sha256(b"{}").hexdigest()},
|
"planned_terminal": None,
|
"transaction_identity": {
|
"transaction_id": "c" * 32,
|
"state_preimage": refresh._identity(config.manifest_path).as_dict(),
|
"formal_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(),
|
"state_candidate": refresh._identity(config.manifest_path).as_dict(),
|
"formal_candidate": refresh._identity(config.refresh.formal_manifest).as_dict(),
|
"created_artifacts": [],
|
"input_item_count": 0,
|
"new_item_count": 0,
|
"formal_lock_claim": None,
|
},
|
"last_transition_at": "2026-08-13T02:00:00Z",
|
}
|
|
def extractor(self) -> dict[str, object]:
|
contract_raw = refresh.OBSERVATION_CONTRACT.read_bytes()
|
contract = json.loads(contract_raw)
|
return {
|
"contract_id": contract["contract_id"],
|
"contract_sha256": hashlib.sha256(contract_raw).hexdigest(),
|
"parser_version": contract["parser_version"],
|
"parser_sha256": hashlib.sha256(refresh.EXTRACTOR_SOURCE.read_bytes()).hexdigest(),
|
}
|
|
def evidence(
|
self,
|
begin: dict[str, object],
|
*,
|
outcome: str,
|
observations: list[dict[str, object]] | None,
|
marker: dict[str, object] | None,
|
items: list[dict[str, object]] | None = None,
|
discovery: str = "NOT_USED",
|
action_outcome: str = "CONFIRMED",
|
observation_outcome: str | None = None,
|
refresh_count: int = 1,
|
observation_count: int | None = None,
|
) -> dict[str, object]:
|
outcome_map = {
|
"READABLE": "READABLE",
|
"UNREADABLE_TIMEOUT": "TIMEOUT",
|
"UNREADABLE_ERROR": "ERROR",
|
"ACCESS_BLOCKED": "ACCESS_BLOCKED",
|
}
|
observation_outcome = observation_outcome or outcome_map[outcome]
|
if action_outcome == "PRE_DISPATCH_ERROR":
|
refresh_started = refresh_finished = read_finished = "2026-08-13T02:00:01Z"
|
action_elapsed = observation_elapsed = 0
|
observation_count = 0
|
refresh_action = None
|
else:
|
refresh_started = "2026-08-13T02:00:01Z"
|
refresh_finished = "2026-08-13T02:00:16Z"
|
read_finished = "2026-08-13T02:00:24Z"
|
action_elapsed = 15000
|
observation_elapsed = 8000
|
observation_count = 1 if observation_count is None else observation_count
|
refresh_action = "reload"
|
evidence = {
|
"schema_version": 3,
|
"run_id": begin["run_id"],
|
"transport": "codex_chrome_visible_page",
|
"requested_url": "https://space.bilibili.com/1420210197/dynamic",
|
"final_url": "https://space.bilibili.com/1420210197/dynamic",
|
"refresh_action": refresh_action,
|
"refresh_count": refresh_count,
|
"refresh_started_at": refresh_started,
|
"refresh_finished_at": refresh_finished,
|
"read_finished_at": read_finished,
|
"page_outcome": outcome,
|
"page_title": "青枫浦上Q个人动态-青枫浦上Q动态记录-哔哩哔哩视频",
|
"creator": {"uid": "1420210197", "name": "青枫浦上Q", "profile_url": "https://space.bilibili.com/1420210197"},
|
"extractor": self.extractor(),
|
"page_observation": None if observations is None else {
|
"schema_version": 1,
|
"limits": json.loads(refresh.OBSERVATION_CONTRACT.read_text(encoding="utf-8"))["limits"],
|
"observations": observations,
|
"terminal_marker": marker,
|
},
|
"items": items or [],
|
"discovery_summary": {"status": discovery, "item_count": 0},
|
"safe_diagnostics": {
|
"code": "OBSERVATION_TIMEOUT" if outcome == "UNREADABLE_TIMEOUT" else "OBSERVATION_ERROR" if outcome == "UNREADABLE_ERROR" else "ACCESS_INTERSTITIAL" if outcome == "ACCESS_BLOCKED" else "NONE",
|
"overall_deadline_seconds": 120,
|
"refresh_action_timeout_seconds": 35,
|
"observation_timeout_seconds": 45,
|
},
|
"runtime_contract": {
|
"contract_id": begin["runtime_contract_id"],
|
"contract_bytes": Path(refresh.RUNTIME_CONTRACT).stat().st_size,
|
"contract_sha256": begin["runtime_contract_sha256"],
|
},
|
"runtime_observation": {
|
"refresh_action_outcome": action_outcome,
|
"refresh_action_elapsed_ms": action_elapsed,
|
"refresh_count": refresh_count,
|
"observation_outcome": observation_outcome,
|
"observation_elapsed_ms": observation_elapsed,
|
"observation_count": observation_count,
|
},
|
}
|
config = collector.load_config(self.config_path)
|
pending = refresh._load_pending(config)
|
assert pending is not None
|
if action_outcome == "PRE_DISPATCH_ERROR":
|
action_started = action_finished = observation_started = observation_finished = None
|
else:
|
action_started = 1000
|
action_finished = 1000 + action_elapsed
|
if observation_count:
|
observation_started = action_finished
|
observation_finished = observation_started + observation_elapsed
|
else:
|
observation_started = observation_finished = None
|
write_started = max(
|
1000,
|
action_finished or 0,
|
observation_finished or 0,
|
)
|
refresh._attest_controller_evidence(
|
pending,
|
evidence,
|
action_dispatched=action_outcome != "PRE_DISPATCH_ERROR",
|
monotonic_run_started_ms=0,
|
monotonic_action_started_ms=action_started,
|
monotonic_action_finished_ms=action_finished,
|
monotonic_observation_started_ms=observation_started,
|
monotonic_observation_finished_ms=observation_finished,
|
monotonic_evidence_write_started_ms=write_started,
|
)
|
return evidence
|
|
def set_runtime_diagnostic(self, evidence: dict[str, object], code: str) -> None:
|
diagnostics = evidence["safe_diagnostics"]
|
assert isinstance(diagnostics, dict)
|
diagnostics["code"] = code
|
|
@staticmethod
|
def observation(cards: list[dict[str, object]], unparsed: list[dict[str, object]] | None = None) -> dict[str, object]:
|
unparsed = unparsed or []
|
return {
|
"ordinal": 0,
|
"observed_at": "2026-08-13T02:00:24Z",
|
"cursor_before": 0,
|
"cursor_after": len(cards) + len(unparsed),
|
"visible_node_count": len(cards) + len(unparsed),
|
"complete_card_count": len(cards),
|
"unparsed_node_count": len(unparsed),
|
"cards": cards,
|
"unparsed_nodes": unparsed,
|
"limit_hit": "NONE",
|
}
|
|
def commit(self, begin: dict[str, object], evidence: dict[str, object]) -> tuple[int, dict[str, object]]:
|
if "controller_attestation" in evidence:
|
config = collector.load_config(self.config_path)
|
pending = refresh._load_pending(config)
|
assert pending is not None
|
attestation = evidence["controller_attestation"]
|
assert isinstance(attestation, dict)
|
refresh._attest_controller_evidence(
|
pending,
|
evidence,
|
action_dispatched=bool(attestation["action_dispatched"]),
|
monotonic_run_started_ms=int(attestation["monotonic_run_started_ms"]),
|
monotonic_action_started_ms=attestation["monotonic_action_started_ms"],
|
monotonic_action_finished_ms=attestation["monotonic_action_finished_ms"],
|
monotonic_observation_started_ms=attestation["monotonic_observation_started_ms"],
|
monotonic_observation_finished_ms=attestation["monotonic_observation_finished_ms"],
|
monotonic_evidence_write_started_ms=int(attestation["monotonic_evidence_write_started_ms"]),
|
)
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
config = collector.load_config(self.config_path)
|
pending = refresh._load_pending(config)
|
assert pending is not None
|
refresh.bind_controller_evidence(
|
config, pending, path,
|
transitioned_at=collector.parse_datetime("2026-08-13T10:00:24+08:00", "test transitioned_at"),
|
)
|
return collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
|
@staticmethod
|
def bound_card_item(identifier: str = "456") -> tuple[dict[str, object], dict[str, object]]:
|
source = f"https://www.bilibili.com/opus/{identifier}"
|
card = {
|
"position": 0,
|
"identifiers": {"dynamic_id": identifier, "opus_id": identifier, "bvid": None},
|
"stable_keys": [f"dynamic:{identifier}", f"opus:{identifier}", f"url:{source}"],
|
"published_at": "2026-08-13T01:55:00Z",
|
"content_type": "text",
|
"source_url": source,
|
}
|
item = {
|
"dynamic_id": identifier, "opus_id": identifier, "bvid": None, "content_type": "text",
|
"published_at": "2026-08-13T01:55:00Z", "title": "offline body",
|
"source_url": source, "body_text": "offline body", "body_complete": True,
|
"duration_seconds": None, "artifacts": [],
|
}
|
return card, item
|
|
@staticmethod
|
def end_marker() -> dict[str, object]:
|
selector = "DYNAMIC_FEED_END_TEXT"
|
text = "没有更多动态了"
|
return {
|
"observation_ordinal": 0, "kind": "EXACT_END_OF_FEED", "selector_id": selector,
|
"normalized_text": text, "marker_sha256": hashlib.sha256(f"{selector}\n{text}".encode()).hexdigest(),
|
}
|
|
def test_real_timeout_and_412_replay_never_claim_no_new(self) -> None:
|
begin = self.begin()
|
before = (self.archive / "manifest.jsonl").read_bytes() if (self.archive / "manifest.jsonl").exists() else b""
|
evidence = self.evidence(begin, outcome="UNREADABLE_TIMEOUT", observations=None, marker=None, discovery="BLOCKED_412")
|
code, result = self.commit(begin, evidence)
|
self.assertEqual(4, code)
|
self.assertEqual("REFRESH_FAILED_PAGE_UNREADABLE", result["status"])
|
self.assertFalse(result["no_new_confirmed"])
|
self.assertFalse(result["formal_manifest_changed"])
|
after = (self.archive / "manifest.jsonl").read_bytes() if (self.archive / "manifest.jsonl").exists() else b""
|
self.assertEqual(before, after)
|
self.assertFalse((self.state / "refresh" / "pending.json").exists())
|
slot = json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8"))
|
self.assertEqual("TERMINAL", slot["run_state"])
|
self.assertEqual("BLOCKED_412", evidence["discovery_summary"]["status"])
|
|
def test_empty_page_requires_exact_marker_for_no_new(self) -> None:
|
begin = self.begin()
|
config = collector.load_config(self.config_path)
|
pending_before = refresh._load_pending(config)
|
self.assertIsNotNone(pending_before)
|
observation = self.observation([])
|
marker = {
|
"observation_ordinal": 0,
|
"kind": "EXACT_END_OF_FEED",
|
"selector_id": "DYNAMIC_FEED_END_TEXT",
|
"normalized_text": "没有更多动态了",
|
"marker_sha256": hashlib.sha256("DYNAMIC_FEED_END_TEXT\n没有更多动态了".encode()).hexdigest(),
|
}
|
code, result = self.commit(begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=marker))
|
self.assertEqual(0, code)
|
self.assertEqual("REFRESH_CONFIRMED_NO_NEW", result["status"])
|
self.assertTrue(result["coverage_complete"])
|
slot_path = Path(str(result["run_evidence_path"]))
|
slot_payload = slot_path.read_bytes()
|
slot = json.loads(slot_payload)
|
receipt = slot["transaction_receipt"]
|
self.assertEqual("NO_FORMAL_CHANGE", receipt["business_commit_kind"])
|
self.assertEqual(receipt["state_preimage"], receipt["state_candidate"])
|
self.assertEqual(receipt["formal_preimage"], receipt["formal_candidate"])
|
latest_path = self.state / "refresh" / "runs" / "latest.json"
|
latest = json.loads(latest_path.read_text(encoding="utf-8"))
|
self.assertEqual(
|
{"schema_version", "slot_index", "hour_epoch", "run_id", "slot_bytes", "slot_sha256", "status", "terminal_at"},
|
set(latest),
|
)
|
self.assertEqual(len(slot_payload), latest["slot_bytes"])
|
self.assertEqual(hashlib.sha256(slot_payload).hexdigest(), latest["slot_sha256"])
|
|
latest_path.unlink()
|
replay_pending = dict(pending_before)
|
replay_pending["phase"] = "TERMINAL_RECORDED"
|
replay_pending["planned_terminal"] = {key: slot[key] for key in (
|
"terminal_at", "status", "error_code", "exit_code", "coverage_proof", "state_manifest",
|
"formal_manifest", "artifact_tree_sha256", "refresh_action", "refresh_count", "page_authoritative",
|
"coverage_complete", "evidence_sha256", "input_item_count", "new_item_count", "saved_artifact_count",
|
"transaction_receipt",
|
)}
|
replay_pending["planned_terminal"]["formal_manifest_changed"] = False
|
refresh._write_pending(config, replay_pending, create=True)
|
replay = refresh._recover_or_replay(
|
config, self.config_path, collector.parse_now("2026-08-13T10:00:26+08:00")
|
)
|
self.assertEqual("REFRESH_CONFIRMED_NO_NEW", replay["status"])
|
self.assertEqual(latest, json.loads(latest_path.read_text(encoding="utf-8")))
|
|
def test_unparsed_node_is_mutually_exclusive_and_forces_partial(self) -> None:
|
begin = self.begin()
|
node = {"position": 0, "node_fingerprint_sha256": "a" * 64, "reason_code": "PARSER_REJECTED"}
|
observation = self.observation([], [node])
|
marker = {
|
"observation_ordinal": 0,
|
"kind": "EXACT_END_OF_FEED",
|
"selector_id": "DYNAMIC_FEED_END_TEXT",
|
"normalized_text": "已经到底了",
|
"marker_sha256": hashlib.sha256("DYNAMIC_FEED_END_TEXT\n已经到底了".encode()).hexdigest(),
|
}
|
code, result = self.commit(begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=marker))
|
self.assertEqual(4, code)
|
self.assertEqual("PARTIAL_DISCOVERY_UNCONFIRMED", result["status"])
|
self.assertFalse(result["coverage_complete"])
|
|
def test_node_position_overlap_fails_before_formal_write(self) -> None:
|
begin = self.begin()
|
card = {
|
"position": 0,
|
"identifiers": {"dynamic_id": "123", "opus_id": None, "bvid": None},
|
"stable_keys": ["dynamic:123", "url:https://www.bilibili.com/opus/123"],
|
"published_at": "2026-08-13T01:55:00Z",
|
"content_type": "text",
|
"source_url": "https://www.bilibili.com/opus/123",
|
}
|
node = {"position": 0, "node_fingerprint_sha256": "b" * 64, "reason_code": "PARSER_REJECTED"}
|
observation = self.observation([card], [node])
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, self.evidence(begin, outcome="READABLE", observations=[observation], marker=None))
|
with self.assertRaisesRegex(collector.CollectorError, "positions overlap"):
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertFalse((self.archive / "manifest.jsonl").exists())
|
|
def test_complete_observed_card_with_missing_item_cannot_claim_no_new(self) -> None:
|
begin = self.begin()
|
card, _ = self.bound_card_item()
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([card])],
|
marker=self.end_marker(), items=[],
|
)
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
with self.assertRaises(collector.CollectorError) as failure:
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
|
self.assertFalse((self.archive / "manifest.jsonl").exists())
|
|
def test_extra_item_without_observed_card_fails_closed(self) -> None:
|
begin = self.begin()
|
_, item = self.bound_card_item()
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([])],
|
marker=self.end_marker(), items=[item],
|
)
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
with self.assertRaises(collector.CollectorError) as failure:
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
|
self.assertFalse((self.archive / "manifest.jsonl").exists())
|
|
def test_observation_item_content_binding_mismatch_fails_closed(self) -> None:
|
begin = self.begin()
|
card, item = self.bound_card_item()
|
item["content_type"] = "article"
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([card])],
|
marker=self.end_marker(), items=[item],
|
)
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
with self.assertRaises(collector.CollectorError) as failure:
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
|
|
def test_observation_overlapping_component_identity_fails_closed(self) -> None:
|
begin = self.begin()
|
card, item = self.bound_card_item()
|
other_card, other_item = self.bound_card_item("457")
|
other_card["identifiers"]["dynamic_id"] = "456"
|
other_card["stable_keys"] = [
|
"dynamic:456", "opus:457", "url:https://www.bilibili.com/opus/457",
|
]
|
other_item["dynamic_id"] = "456"
|
for position, row in enumerate((card, other_card)):
|
row["position"] = position
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([card, other_card])],
|
marker=self.end_marker(), items=[item, other_item],
|
)
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
with self.assertRaises(collector.CollectorError) as failure:
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
|
|
def test_new_text_commits_both_manifests_and_artifact(self) -> None:
|
begin = self.begin()
|
card = {
|
"position": 0,
|
"identifiers": {"dynamic_id": "456", "opus_id": "456", "bvid": None},
|
"stable_keys": ["dynamic:456", "opus:456", "url:https://www.bilibili.com/opus/456"],
|
"published_at": "2026-08-13T01:55:00Z",
|
"content_type": "text",
|
"source_url": "https://www.bilibili.com/opus/456",
|
}
|
item = {
|
"dynamic_id": "456", "opus_id": "456", "bvid": None, "content_type": "text",
|
"published_at": "2026-08-13T01:55:00Z", "title": "离线合成正文",
|
"source_url": "https://www.bilibili.com/opus/456", "body_text": "仅测试文本",
|
"body_complete": True, "duration_seconds": None, "artifacts": [],
|
}
|
observation = self.observation([card])
|
marker = {
|
"observation_ordinal": 0,
|
"kind": "EXACT_END_OF_FEED",
|
"selector_id": "DYNAMIC_FEED_END_TEXT",
|
"normalized_text": "没有更多动态了",
|
"marker_sha256": hashlib.sha256("DYNAMIC_FEED_END_TEXT\n没有更多动态了".encode()).hexdigest(),
|
}
|
code, result = self.commit(begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=marker, items=[item]))
|
self.assertEqual(0, code)
|
self.assertEqual("NEW_ITEMS_SAVED", result["status"])
|
self.assertEqual(1, result["new_items"])
|
self.assertEqual(1, result["saved_artifacts"])
|
self.assertEqual(1, len(collector.load_manifest(self.state / "manifest.jsonl")))
|
formal = (self.archive / "manifest.jsonl").read_text(encoding="utf-8")
|
self.assertIn('"event_type":"DYNAMIC_CONTENT_SAVED"', formal)
|
artifacts = [path for path in self.archive.glob("*.txt")]
|
self.assertEqual(1, len(artifacts))
|
self.assertEqual("仅测试文本\n", artifacts[0].read_text(encoding="utf-8"))
|
|
slot = json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8"))
|
self.assertEqual("DUAL_MANIFEST_COMMIT", slot["transaction_receipt"]["business_commit_kind"])
|
self.assertEqual("BUSINESS_COMMITTED", slot["transaction_receipt"]["phase"])
|
|
def test_hourly_slot_and_dead_pending_recovery_are_bounded(self) -> None:
|
begin = self.begin()
|
with self.assertRaises(collector.CollectorError) as busy:
|
collector.run(["--config", str(self.config_path), "refresh-begin", "--now", "2026-08-13T10:00:10+08:00"])
|
self.assertEqual("E_BUSY", busy.exception.code)
|
code, terminal = collector.run(["--config", str(self.config_path), "refresh-begin", "--now", "2026-08-13T10:02:01+08:00"])
|
self.assertEqual(4, code)
|
self.assertEqual("E_EVIDENCE_MISSING_AFTER_DEADLINE", terminal["error_code"])
|
with self.assertRaises(collector.CollectorError) as occupied:
|
collector.run(["--config", str(self.config_path), "refresh-begin", "--now", "2026-08-13T10:02:10+08:00"])
|
self.assertEqual("E_RUN_HOUR_OCCUPIED", occupied.exception.code)
|
|
def test_latest_commit_failure_retains_pending_and_reopen_rebuilds_index(self) -> None:
|
begin = self.begin()
|
observation = self.observation([])
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[observation], marker=self.end_marker(), items=[]
|
)
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
config = collector.load_config(self.config_path)
|
pending = refresh._load_pending(config)
|
assert pending is not None
|
refresh.bind_controller_evidence(
|
config, pending, path,
|
transitioned_at=collector.parse_datetime("2026-08-13T10:00:24+08:00", "test transitioned_at"),
|
)
|
original = refresh._write_readback
|
injected = {"done": False}
|
|
def fail_latest(target: Path, payload: bytes, description: str) -> None:
|
if description == "latest run index" and not injected["done"]:
|
injected["done"] = True
|
raise OSError("injected latest failure")
|
original(target, payload, description)
|
|
refresh._write_readback = fail_latest
|
try:
|
with self.assertRaisesRegex(OSError, "injected latest failure"):
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
finally:
|
refresh._write_readback = original
|
self.assertTrue(refresh._pending_path(config).is_file())
|
self.assertFalse((self.state / "refresh" / "runs" / "latest.json").exists())
|
replay = refresh._recover_or_replay(
|
config, self.config_path, collector.parse_now("2026-08-13T10:00:26+08:00"),
|
allow_final_evidence=True,
|
)
|
self.assertEqual("REFRESH_CONFIRMED_NO_NEW", replay["status"])
|
self.assertTrue((self.state / "refresh" / "runs" / "latest.json").is_file())
|
self.assertFalse(refresh._pending_path(config).exists())
|
|
def test_state_lock_is_kernel_backed_and_owner_metadata_recovers(self) -> None:
|
config = collector.load_config(self.config_path)
|
config.state_dir.mkdir()
|
stale = {
|
"pid": 99999999,
|
"process_created_at": "0",
|
"run_id": None,
|
"acquired_at": "2026-08-13T00:00:00Z",
|
}
|
self.write_json(config.state_dir / ".collector.lock.owner.json", stale)
|
with collector.StateLock(config.lock_path):
|
self.assertEqual(1, config.lock_path.stat().st_size)
|
self.assertTrue((config.state_dir / ".collector.lock.owner.json").is_file())
|
with self.assertRaises(collector.CollectorError) as locked:
|
with collector.StateLock(config.lock_path):
|
pass
|
self.assertEqual("E_STATE_LOCKED", locked.exception.code)
|
self.assertTrue(config.lock_path.is_file())
|
self.assertFalse((config.state_dir / ".collector.lock.owner.json").exists())
|
|
def test_frozen_formal_catalog_shape_is_48_22_16_5_1_read_only(self) -> None:
|
formal = Path(__file__).resolve().parents[3] / "ana-data" / "news-青枫浦上Q" / "manifest.jsonl"
|
if not formal.exists():
|
self.skipTest("formal read-only acceptance snapshot is not present")
|
before = (formal.stat().st_size, hashlib.sha256(formal.read_bytes()).hexdigest())
|
config = collector.load_config(self.config_path)
|
archive = formal.parent
|
rebound = collector.CollectorConfig(
|
config.creator_name, config.creator_uid, config.creator_dynamic_url, config.timezone_name,
|
config.window_hours, config.state_dir, config.download_dir, config.video_dir,
|
config.minimum_complete_age_seconds, config.title_max_length, config.allowed_source_hosts,
|
config.allowed_video_extensions, config.native_handoff,
|
collector.RefreshConfig(
|
archive, formal, config.refresh.intake_dir, config.refresh.overall_deadline_seconds,
|
config.refresh.refresh_action_timeout_seconds, config.refresh.observation_timeout_seconds,
|
config.refresh.page_internal_settle_timeout_seconds,
|
1, 168, 200, 20, config.refresh.max_image_bytes, config.refresh.max_text_bytes,
|
),
|
)
|
_, _, counts = refresh.load_formal_catalog(rebound)
|
self.assertEqual({"events": 48, "components": 22, "saved": 16, "video": 5, "retryable": 1}, counts)
|
self.assertEqual(before, (formal.stat().st_size, hashlib.sha256(formal.read_bytes()).hexdigest()))
|
|
def test_mixed_history_union_saved_precedence_and_schema2_identity(self) -> None:
|
artifact = self.archive / "saved.txt"
|
artifact.write_bytes(b"saved\n")
|
digest = hashlib.sha256(artifact.read_bytes()).hexdigest()
|
config = collector.load_config(self.config_path)
|
pending = {
|
"schema_version": 1, "creator": config.creator_name, "creator_uid": config.creator_uid,
|
"item_type": "article", "stable_id": "700", "source_url": "https://www.bilibili.com/opus/700",
|
"published_at": "2026-08-12T01:00:00Z", "collected_at": "2026-08-12T02:00:00Z",
|
"status": "CONTENT_ACCESS_PENDING_CHROME",
|
}
|
saved = {
|
**pending, "status": "SAVED", "path": "saved.txt", "bytes": artifact.stat().st_size,
|
"sha256": digest.upper(),
|
}
|
formal = self.archive / "manifest.jsonl"
|
formal.write_bytes(collector.canonical_json_bytes(pending) + collector.canonical_json_bytes(saved))
|
_, tokens, counts = refresh.load_formal_catalog(config)
|
self.assertEqual({"events": 2, "components": 1, "saved": 1, "video": 0, "retryable": 0}, counts)
|
self.assertEqual({"opus:700", "url:https://www.bilibili.com/opus/700"}, tokens)
|
|
def test_schema2_status_entity_and_artifact_identity_fail_closed(self) -> None:
|
artifact = self.archive / "new.txt"
|
artifact.write_bytes(b"new\n")
|
source = "https://www.bilibili.com/opus/701"
|
keys = ["dynamic:701", "opus:701", f"url:{source}"]
|
config = collector.load_config(self.config_path)
|
event = {
|
"schema_version": 2, "event_type": "DYNAMIC_CONTENT_SAVED", "creator": config.creator_name,
|
"creator_uid": config.creator_uid, "entity_id": collector.entity_id_for_keys(keys),
|
"dynamic_id": "701", "opus_id": "701", "bvid": None, "dedupe_keys": keys,
|
"content_type": "text", "published_at": "2026-08-12T01:00:00Z", "title": "saved",
|
"source_url": source, "collected_at": "2026-08-12T02:00:00Z",
|
"artifacts": [{"kind": "text", "sequence": 1, "path": "new.txt", "bytes": 4,
|
"sha256": hashlib.sha256(b"new\n").hexdigest()}],
|
"duration_seconds": None, "page_run_id": "a" * 32, "coverage_complete": True, "status": "SAVED",
|
}
|
mutations = (
|
("status", {**event, "status": "UNKNOWN"}, "E_CATALOG_STATUS"),
|
("integer_uid", {**event, "creator_uid": int(config.creator_uid)}, "E_CATALOG_STATUS"),
|
("entity", {**event, "entity_id": "0" * 24}, "E_CATALOG_IDENTITY_CONFLICT"),
|
("artifact", {**event, "artifacts": [{**event["artifacts"][0], "path": "absent.txt"}]}, "E_CATALOG_ARTIFACT"),
|
)
|
for label, bad, expected in mutations:
|
with self.subTest(label=label):
|
(self.archive / "manifest.jsonl").write_bytes(collector.canonical_json_bytes(bad))
|
with self.assertRaises(collector.CollectorError) as failure:
|
refresh.load_formal_catalog(config)
|
self.assertEqual(expected, failure.exception.code)
|
|
def test_schema1_stable_id_must_be_one_to_thirty_two_ascii_digits(self) -> None:
|
config = collector.load_config(self.config_path)
|
base = {
|
"schema_version": 1, "creator": config.creator_name, "creator_uid": config.creator_uid,
|
"item_type": "article", "source_url": "https://www.bilibili.com/opus/123",
|
"published_at": "2026-08-12T01:00:00Z", "collected_at": "2026-08-12T02:00:00Z",
|
"status": "CONTENT_ACCESS_PENDING_CHROME",
|
}
|
for stable_id in ("1" * 33, "１２３"):
|
with self.subTest(stable_id=stable_id):
|
row = {**base, "stable_id": stable_id}
|
(self.archive / "manifest.jsonl").write_bytes(collector.canonical_json_bytes(row))
|
with self.assertRaises(collector.CollectorError) as failure:
|
refresh.load_formal_catalog(config)
|
self.assertEqual("E_CATALOG_IDENTITY_CONFLICT", failure.exception.code)
|
|
def test_pending_before_started_rebuilds_same_run_and_third_slot_is_untouched(self) -> None:
|
config = collector.load_config(self.config_path)
|
original = refresh._ensure_started_slot
|
refresh._ensure_started_slot = lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("after pending"))
|
try:
|
with self.assertRaisesRegex(OSError, "after pending"):
|
refresh.refresh_begin(config, self.config_path, collector.parse_now(NOW))
|
finally:
|
refresh._ensure_started_slot = original
|
pending = refresh._load_pending(config)
|
self.assertIsNotNone(pending)
|
run_id = pending["run_id"]
|
result = refresh.refresh_begin(config, self.config_path, collector.parse_now("2026-08-13T10:00:01+08:00"))
|
self.assertEqual(run_id, result["run_id"])
|
self.assertEqual("RUN_STARTED", json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8"))["run_state"])
|
|
Path(str(result["run_evidence_path"])).write_bytes(b'{"third":true}')
|
before = Path(str(result["run_evidence_path"])).read_bytes()
|
with self.assertRaises(collector.CollectorError) as failure:
|
refresh._recover_or_replay(config, self.config_path, collector.parse_now("2026-08-13T10:00:02+08:00"))
|
self.assertEqual("E_RECOVERY_AMBIGUOUS", failure.exception.code)
|
self.assertEqual(before, Path(str(result["run_evidence_path"])).read_bytes())
|
|
def test_terminal_never_overwrites_third_slot_and_recorded_replay_binds_plan(self) -> None:
|
begin = self.begin()
|
slot_path = Path(str(begin["run_evidence_path"]))
|
third = b'{"third":"terminal"}'
|
evidence = self.evidence(begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker())
|
original_write_pending = refresh._write_pending
|
injected = {"done": False}
|
|
def inject_third(config, pending, *, create=False):
|
original_write_pending(config, pending, create=create)
|
if pending.get("planned_terminal") is not None and not injected["done"]:
|
injected["done"] = True
|
slot_path.write_bytes(third)
|
|
refresh._write_pending = inject_third
|
try:
|
with self.assertRaises(collector.CollectorError) as failure:
|
self.commit(begin, evidence)
|
finally:
|
refresh._write_pending = original_write_pending
|
self.assertEqual("E_RUN_EVIDENCE_COMMIT", failure.exception.code)
|
self.assertEqual(third, slot_path.read_bytes())
|
|
config = collector.load_config(self.config_path)
|
pending = refresh._load_pending(config)
|
self.assertIsNotNone(pending)
|
self.assertIsNotNone(pending["planned_terminal"])
|
slot, payload = refresh._terminal_slot_from_planned(config, pending, pending["planned_terminal"])
|
slot_path.write_bytes(payload)
|
pending["phase"] = "TERMINAL_RECORDED"
|
pending["planned_terminal"]["transaction_receipt"]["intent_sha256"] = "0" * 64
|
refresh._write_pending(config, pending)
|
before = slot_path.read_bytes()
|
with self.assertRaises(collector.CollectorError) as replay_failure:
|
refresh._recover_or_replay(config, self.config_path, collector.parse_now("2026-08-13T10:00:26+08:00"))
|
self.assertEqual("E_RUN_EVIDENCE_COMMIT", replay_failure.exception.code)
|
self.assertEqual(before, slot_path.read_bytes())
|
|
def test_mixed_history_cross_line_identity_conflict_fails_closed(self) -> None:
|
config = collector.load_config(self.config_path)
|
first = {
|
"schema_version": 1, "creator": config.creator_name, "creator_uid": config.creator_uid,
|
"item_type": "video", "stable_id": "BV1HA3o6oEJJ",
|
"source_url": "https://www.bilibili.com/video/BV1HA3o6oEJJ",
|
"published_at": "2026-08-12T01:00:00Z", "collected_at": "2026-08-12T02:00:00Z",
|
"status": "VIDEO_DOWNLOAD_PENDING_EXTENSION",
|
}
|
second = {**first, "stable_id": "BV1DVMX6XEPq"}
|
(self.archive / "manifest.jsonl").write_bytes(
|
collector.canonical_json_bytes(first) + collector.canonical_json_bytes(second)
|
)
|
with self.assertRaises(collector.CollectorError) as failure:
|
refresh.load_formal_catalog(config)
|
self.assertEqual("E_CATALOG_IDENTITY_CONFLICT", failure.exception.code)
|
|
def test_dead_formal_lock_takeover_uses_archive_quarantine_and_current_identity(self) -> None:
|
config = collector.load_config(self.config_path)
|
config.state_dir.mkdir()
|
pending = {
|
"schema_version": 2,
|
"run_id": "a" * 32,
|
"owner_nonce": "b" * 32,
|
"phase": "TRANSACTION_INTENT",
|
"task_id": refresh.TASK_ID,
|
"creator_uid": config.creator_uid,
|
"creator_dynamic_url": config.creator_dynamic_url,
|
"started_at": "2026-08-13T02:00:00Z",
|
"deadline_at": "2026-08-13T02:00:30Z",
|
"window_start": "2026-08-10T02:00:00Z",
|
"window_end": "2026-08-13T02:00:00Z",
|
"config_sha256": hashlib.sha256(self.config_path.read_bytes()).hexdigest(),
|
"state_manifest_preimage": refresh._identity(config.manifest_path).as_dict(),
|
"formal_manifest_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(),
|
"evidence_path": str(config.state_dir / "refresh" / "incoming" / ("a" * 32 + ".json")),
|
"intake_root": str(config.refresh.intake_dir / ("a" * 32)),
|
"evidence_identity": {"bytes": 2, "sha256": hashlib.sha256(b"{}").hexdigest()},
|
"planned_terminal": None,
|
"transaction_identity": {
|
"transaction_id": "c" * 32,
|
"state_preimage": refresh._identity(config.manifest_path).as_dict(),
|
"formal_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(),
|
"state_candidate": refresh._identity(config.manifest_path).as_dict(),
|
"formal_candidate": refresh._identity(config.refresh.formal_manifest).as_dict(),
|
"created_artifacts": [],
|
"input_item_count": 0,
|
"new_item_count": 0,
|
"formal_lock_claim": None,
|
},
|
"last_transition_at": "2026-08-13T02:00:00Z",
|
}
|
old_record = {
|
"schema_version": 1,
|
"task_id": refresh.TASK_ID,
|
"run_id": pending["run_id"],
|
"owner_nonce": pending["owner_nonce"],
|
"transaction_id": pending["transaction_identity"]["transaction_id"],
|
"recovery_generation": 0,
|
"holder_pid": 99999999,
|
"holder_process_created_at": "0",
|
"lock_created_at": "2026-08-13T02:00:00Z",
|
}
|
old_payload = collector.canonical_json_bytes(old_record, newline=False)
|
old_claim = {
|
"claim_state": "HELD",
|
"lock_path": str(refresh._formal_lock_path(config)),
|
"lock_bytes": len(old_payload),
|
"lock_sha256": hashlib.sha256(old_payload).hexdigest(),
|
"lock_record": old_record,
|
}
|
pending["transaction_identity"]["formal_lock_claim"] = old_claim
|
refresh._write_pending(config, pending, create=True)
|
refresh._create_new(refresh._formal_lock_path(config), old_payload)
|
refresh._recovery_formal_lock(config, pending, collector.parse_now("2026-08-13T10:00:25+08:00"))
|
claim = pending["transaction_identity"]["formal_lock_claim"]
|
self.assertEqual("HELD", claim["claim_state"])
|
self.assertEqual(1, claim["lock_record"]["recovery_generation"])
|
self.assertEqual(os.getpid(), claim["lock_record"]["holder_pid"])
|
takeover = pending["transaction_identity"]["takeover"]
|
self.assertEqual("NEW_LOCK_HELD", takeover["phase"])
|
quarantine = Path(takeover["quarantine_path"])
|
self.assertEqual(config.refresh.archive_dir, quarantine.parent.parent)
|
self.assertEqual(old_payload, quarantine.read_bytes())
|
self.assertEqual(collector.canonical_json_bytes(claim["lock_record"], newline=False), refresh._formal_lock_path(config).read_bytes())
|
lock_path = refresh._formal_lock_path(config)
|
lock_path.unlink()
|
claim["claim_state"] = "PLANNED"
|
claim["lock_record"]["holder_pid"] = 99999999
|
claim["lock_record"]["holder_process_created_at"] = "0"
|
dead_payload = collector.canonical_json_bytes(claim["lock_record"], newline=False)
|
claim["lock_bytes"] = len(dead_payload)
|
claim["lock_sha256"] = hashlib.sha256(dead_payload).hexdigest()
|
takeover["phase"] = "CREATE_PLANNED"
|
prior_generation = takeover["next_generation"]
|
prior_attempt = takeover["claim_attempt"]
|
refresh._write_pending(config, pending)
|
refresh._recovery_formal_lock(config, pending, collector.parse_now("2026-08-13T10:00:26+08:00"))
|
claim = pending["transaction_identity"]["formal_lock_claim"]
|
takeover = pending["transaction_identity"]["takeover"]
|
self.assertEqual(prior_generation, claim["lock_record"]["recovery_generation"])
|
self.assertEqual(prior_attempt + 1, takeover["claim_attempt"])
|
self.assertEqual(os.getpid(), claim["lock_record"]["holder_pid"])
|
|
lock_path.unlink()
|
claim["claim_state"] = "PLANNED"
|
claim["lock_record"]["holder_pid"] = 99999999
|
claim["lock_record"]["holder_process_created_at"] = "0"
|
dead_payload = collector.canonical_json_bytes(claim["lock_record"], newline=False)
|
claim["lock_bytes"] = len(dead_payload)
|
claim["lock_sha256"] = hashlib.sha256(dead_payload).hexdigest()
|
takeover["phase"] = "CREATE_PLANNED"
|
refresh._write_pending(config, pending)
|
refresh._create_new(lock_path, dead_payload)
|
original_rename = refresh._rename_no_overwrite
|
rename_calls: list[tuple[Path, Path]] = []
|
|
def fail_after_intent(source: Path, target: Path) -> None:
|
rename_calls.append((source, target))
|
raise OSError("crash before failed-claim rename")
|
|
refresh._rename_no_overwrite = fail_after_intent
|
try:
|
with self.assertRaisesRegex(OSError, "crash before failed-claim rename"):
|
refresh._recovery_formal_lock(config, pending, collector.parse_now("2026-08-13T10:00:27+08:00"))
|
finally:
|
refresh._rename_no_overwrite = original_rename
|
persisted = refresh._load_pending(config)
|
intent = persisted["transaction_identity"]["takeover"]["failed_claim_intent"]
|
self.assertEqual(str(rename_calls[0][1]), intent["path"])
|
self.assertEqual(dead_payload, lock_path.read_bytes())
|
self.assertFalse(Path(intent["path"]).exists())
|
|
refresh._recovery_formal_lock(config, persisted, collector.parse_now("2026-08-13T10:00:27+08:00"))
|
pending = persisted
|
takeover = pending["transaction_identity"]["takeover"]
|
self.assertEqual(1, len(takeover["failed_claims"]))
|
failed = takeover["failed_claims"][0]
|
self.assertEqual(dead_payload, Path(failed["path"]).read_bytes())
|
self.assertTrue((refresh._quarantine_root(config) / ".owner.json").is_file())
|
refresh._formal_lock_path(config).unlink()
|
refresh._release_formal_lock(config, pending)
|
self.assertFalse(refresh._formal_lock_path(config).exists())
|
self.assertFalse(quarantine.exists())
|
self.assertFalse(refresh._quarantine_root(config).exists())
|
|
def test_quarantine_rename_is_no_overwrite_and_empty_cleanup_reopens(self) -> None:
|
config = collector.load_config(self.config_path)
|
source = refresh._formal_lock_path(config)
|
root = refresh._quarantine_root(config)
|
root.mkdir()
|
target = root / "third.json"
|
source.write_bytes(b"owned")
|
target.write_bytes(b"third")
|
with self.assertRaises(collector.CollectorError) as collision:
|
refresh._rename_no_overwrite(source, target)
|
self.assertEqual("E_RECOVERY_AMBIGUOUS", collision.exception.code)
|
self.assertEqual(b"owned", source.read_bytes())
|
self.assertEqual(b"third", target.read_bytes())
|
|
target.unlink()
|
flushed: list[Path] = []
|
original_fsync = refresh._fsync_directory
|
refresh._fsync_directory = lambda path: flushed.append(Path(path))
|
try:
|
refresh._rename_no_overwrite(source, target)
|
finally:
|
refresh._fsync_directory = original_fsync
|
self.assertFalse(source.exists())
|
self.assertEqual(b"owned", target.read_bytes())
|
self.assertIn(self.archive, flushed)
|
self.assertIn(root, flushed)
|
|
target.unlink()
|
pending = {
|
"run_id": "a" * 32,
|
"owner_nonce": "b" * 32,
|
"transaction_identity": {
|
"formal_lock_claim": {
|
"lock_bytes": 0, "lock_sha256": hashlib.sha256(b"").hexdigest()
|
},
|
"takeover": {"old_lock_never_created": True, "failed_claims": []},
|
},
|
}
|
refresh._release_formal_lock(config, pending)
|
self.assertFalse(root.exists())
|
|
def test_formal_claim_requires_exact_readback_and_parent_durability_before_held(self) -> None:
|
config = collector.load_config(self.config_path)
|
config.state_dir.mkdir()
|
now = collector.parse_now("2026-08-13T10:00:25+08:00")
|
formal_before = refresh._identity(config.refresh.formal_manifest).as_dict()
|
|
def prepare() -> tuple[dict[str, object], dict[str, object], Path]:
|
pending = self.formal_pending(config)
|
claim = refresh._new_formal_claim(config, pending, 1, now)
|
pending["transaction_identity"]["formal_lock_claim"] = claim
|
pending["transaction_identity"]["takeover"] = {
|
"phase": "CREATE_PLANNED",
|
"old_lock_sha256": hashlib.sha256(b"").hexdigest(),
|
"old_lock_hex": "",
|
"quarantine_path": str(refresh._quarantine_root(config) / "never-created.json"),
|
"next_generation": 1,
|
"takeover_reason": "TEST",
|
"planned_at": collector.canonical_datetime(now),
|
"claim_attempt": 1,
|
"old_lock_never_created": True,
|
}
|
refresh._write_pending(config, pending, create=True)
|
return pending, claim, refresh._formal_lock_path(config)
|
|
pending, _, lock = prepare()
|
original_create = refresh._create_new
|
refresh._create_new = lambda path, payload: path.write_bytes(b"third-party-drift")
|
try:
|
with self.assertRaises(collector.CollectorError) as drift:
|
refresh._recovery_formal_lock(config, pending, now)
|
finally:
|
refresh._create_new = original_create
|
self.assertEqual("E_FORMAL_LOCK", drift.exception.code)
|
self.assertEqual(b"third-party-drift", lock.read_bytes())
|
persisted = refresh._load_pending(config)
|
self.assertEqual("CREATE_PLANNED", persisted["transaction_identity"]["takeover"]["phase"])
|
self.assertEqual("PLANNED", persisted["transaction_identity"]["formal_lock_claim"]["claim_state"])
|
with self.assertRaises(collector.CollectorError) as replay_drift:
|
refresh._recovery_formal_lock(config, persisted, now)
|
self.assertEqual("E_RECOVERY_AMBIGUOUS", replay_drift.exception.code)
|
self.assertEqual(b"third-party-drift", lock.read_bytes())
|
lock.unlink()
|
refresh._pending_path(config).unlink()
|
|
pending, _, lock = prepare()
|
original_read = Path.read_bytes
|
|
def fail_lock_read(path: Path) -> bytes:
|
if path == lock:
|
raise OSError("injected formal lock readback failure")
|
return original_read(path)
|
|
Path.read_bytes = fail_lock_read
|
try:
|
with self.assertRaisesRegex(OSError, "readback failure"):
|
refresh._recovery_formal_lock(config, pending, now)
|
finally:
|
Path.read_bytes = original_read
|
self.assertEqual("CREATE_PLANNED", refresh._load_pending(config)["transaction_identity"]["takeover"]["phase"])
|
lock.unlink()
|
refresh._pending_path(config).unlink()
|
|
pending, claim, lock = prepare()
|
original_fsync = refresh._fsync_directory
|
|
def fail_parent_fsync(path: Path) -> None:
|
if Path(path) == self.archive:
|
raise OSError("injected parent directory fsync failure")
|
original_fsync(path)
|
|
refresh._fsync_directory = fail_parent_fsync
|
try:
|
with self.assertRaisesRegex(OSError, "parent directory fsync failure"):
|
refresh._recovery_formal_lock(config, pending, now)
|
finally:
|
refresh._fsync_directory = original_fsync
|
persisted = refresh._load_pending(config)
|
self.assertEqual("CREATE_PLANNED", persisted["transaction_identity"]["takeover"]["phase"])
|
self.assertEqual("PLANNED", persisted["transaction_identity"]["formal_lock_claim"]["claim_state"])
|
self.assertEqual(refresh._claim_payload(claim), lock.read_bytes())
|
refresh._recovery_formal_lock(config, persisted, now)
|
replayed = refresh._load_pending(config)
|
self.assertEqual("NEW_LOCK_HELD", replayed["transaction_identity"]["takeover"]["phase"])
|
self.assertEqual("HELD", replayed["transaction_identity"]["formal_lock_claim"]["claim_state"])
|
self.assertEqual(formal_before, refresh._identity(config.refresh.formal_manifest).as_dict())
|
refresh._release_formal_lock(config, replayed)
|
|
def test_formal_claim_real_child_kill_reopens_create_and_new_lock_held(self) -> None:
|
config = collector.load_config(self.config_path)
|
config.state_dir.mkdir()
|
formal_before = refresh._identity(config.refresh.formal_manifest).as_dict()
|
script = r'''import hashlib, sys, time
|
from pathlib import Path
|
sys.path.insert(0, sys.argv[1])
|
import bili_dynamic_collector as core
|
import bili_dynamic_refresh as refresh
|
config = core.load_config(Path(sys.argv[2]))
|
marker = Path(sys.argv[3])
|
mode = sys.argv[4]
|
pending = refresh._load_pending(config)
|
now = core.parse_now("2026-08-13T10:00:25+08:00")
|
claim = refresh._new_formal_claim(config, pending, 1, now)
|
pending["transaction_identity"]["formal_lock_claim"] = claim
|
pending["transaction_identity"]["takeover"] = {
|
"phase": "CREATE_PLANNED", "old_lock_sha256": hashlib.sha256(b"").hexdigest(),
|
"old_lock_hex": "", "quarantine_path": str(refresh._quarantine_root(config) / "never-created.json"),
|
"next_generation": 1, "takeover_reason": "CHILD_KILL_TEST",
|
"planned_at": core.canonical_datetime(now), "claim_attempt": 1, "old_lock_never_created": True,
|
}
|
refresh._write_pending(config, pending)
|
if mode == "CREATE_PLANNED":
|
original = refresh._fsync_directory
|
def stop_at_parent(path):
|
if Path(path) == config.refresh.archive_dir and refresh._formal_lock_path(config).exists():
|
marker.write_text("created", encoding="utf-8")
|
time.sleep(120)
|
original(path)
|
refresh._fsync_directory = stop_at_parent
|
refresh._recovery_formal_lock(config, pending, now)
|
marker.write_text("held", encoding="utf-8")
|
time.sleep(120)
|
'''
|
for mode, marker_text in (("CREATE_PLANNED", "created"), ("NEW_LOCK_HELD", "held")):
|
with self.subTest(mode=mode):
|
pending = self.formal_pending(config)
|
refresh._write_pending(config, pending, create=True)
|
marker = self.root / f"{mode}.marker"
|
process = subprocess.Popen(
|
[sys.executable, "-B", "-c", script, str(PROJECT_DEV), str(self.config_path), str(marker), mode],
|
stdout=subprocess.DEVNULL,
|
stderr=subprocess.DEVNULL,
|
)
|
deadline = time.monotonic() + 15
|
while time.monotonic() < deadline and not marker.exists() and process.poll() is None:
|
time.sleep(0.05)
|
self.assertTrue(marker.exists(), f"child failed before {mode} marker; exit={process.poll()}")
|
self.assertEqual(marker_text, marker.read_text(encoding="utf-8"))
|
process.kill()
|
process.wait(timeout=10)
|
child_pid = process.pid
|
if os.name == "nt":
|
process._handle.Close()
|
del process
|
gc.collect()
|
dead_deadline = time.monotonic() + 10
|
while time.monotonic() < dead_deadline:
|
try:
|
collector.process_created_at(child_pid)
|
except ProcessLookupError:
|
break
|
time.sleep(0.05)
|
else:
|
self.fail(f"child process identity remained queryable after kill: {child_pid}")
|
persisted = refresh._load_pending(config)
|
expected_phase = "CREATE_PLANNED" if mode == "CREATE_PLANNED" else "NEW_LOCK_HELD"
|
self.assertEqual(expected_phase, persisted["transaction_identity"]["takeover"]["phase"])
|
self.assertEqual(formal_before, refresh._identity(config.refresh.formal_manifest).as_dict())
|
refresh._recovery_formal_lock(config, persisted, collector.parse_now("2026-08-13T10:00:26+08:00"))
|
recovered = refresh._load_pending(config)
|
self.assertEqual("NEW_LOCK_HELD", recovered["transaction_identity"]["takeover"]["phase"])
|
self.assertEqual("HELD", recovered["transaction_identity"]["formal_lock_claim"]["claim_state"])
|
self.assertEqual(formal_before, refresh._identity(config.refresh.formal_manifest).as_dict())
|
refresh._release_formal_lock(config, recovered)
|
refresh._pending_path(config).unlink()
|
|
def test_runtime_v2_config_and_begin_bind_exact_contract(self) -> None:
|
begin = self.begin()
|
self.assertEqual("bili-supported-chrome-visible-runtime-v2", begin["runtime_contract_id"])
|
self.assertEqual(120, begin["overall_deadline_seconds"])
|
self.assertEqual(35, begin["refresh_action_timeout_seconds"])
|
self.assertEqual(45, begin["observation_timeout_seconds"])
|
self.assertEqual(15, begin["page_internal_settle_timeout_seconds"])
|
pending = refresh._load_pending(collector.load_config(self.config_path))
|
self.assertEqual(3, pending["schema_version"])
|
self.assertEqual(begin["runtime_contract_sha256"], pending["runtime_contract"]["contract_sha256"])
|
|
config = json.loads(self.config_path.read_text(encoding="utf-8"))
|
config["refresh"]["timeout_seconds"] = 30
|
legacy = self.root / "legacy.json"
|
self.write_json(legacy, config)
|
with self.assertRaises(collector.CollectorError) as invalid:
|
collector.load_config(legacy)
|
self.assertEqual("E_CONFIG", invalid.exception.code)
|
|
def test_runtime_pre_dispatch_error_is_zero_action_unreadable(self) -> None:
|
begin = self.begin()
|
evidence = self.evidence(
|
begin, outcome="UNREADABLE_ERROR", observations=None, marker=None,
|
action_outcome="PRE_DISPATCH_ERROR", observation_outcome="NOT_ATTEMPTED", refresh_count=0,
|
)
|
self.set_runtime_diagnostic(evidence, "ACTION_PRE_DISPATCH")
|
before = refresh._identity(self.archive / "manifest.jsonl").as_dict()
|
code, result = self.commit(begin, evidence)
|
self.assertEqual(4, code)
|
self.assertEqual("REFRESH_FAILED_PAGE_UNREADABLE", result["status"])
|
self.assertEqual(0, result["refresh_count"])
|
self.assertEqual(before, refresh._identity(self.archive / "manifest.jsonl").as_dict())
|
|
def test_runtime_pre_dispatch_claimed_observation_fails_closed(self) -> None:
|
for observation_outcome in ("READABLE", "ERROR"):
|
with self.subTest(observation_outcome=observation_outcome):
|
begin = self.begin()
|
evidence = self.evidence(
|
begin, outcome="UNREADABLE_ERROR", observations=None, marker=None,
|
action_outcome="PRE_DISPATCH_ERROR", observation_outcome="NOT_ATTEMPTED", refresh_count=0,
|
)
|
self.set_runtime_diagnostic(evidence, "ACTION_PRE_DISPATCH")
|
evidence["runtime_observation"]["observation_outcome"] = observation_outcome
|
evidence["runtime_observation"]["observation_count"] = 1
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
with self.assertRaises(collector.CollectorError) as invalid:
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertEqual("E_EVIDENCE_SCHEMA", invalid.exception.code)
|
refresh._pending_path(collector.load_config(self.config_path)).unlink()
|
Path(str(begin["run_evidence_path"])).unlink()
|
|
def test_runtime_post_dispatch_error_matrix(self) -> None:
|
begin = self.begin()
|
readable_empty = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(),
|
action_outcome="POST_DISPATCH_ERROR",
|
)
|
self.set_runtime_diagnostic(readable_empty, "ACTION_POST_DISPATCH")
|
code, result = self.commit(begin, readable_empty)
|
self.assertEqual(4, code)
|
self.assertEqual("PARTIAL_DISCOVERY_UNCONFIRMED", result["status"])
|
self.assertFalse(result["no_new_confirmed"])
|
|
def test_runtime_post_dispatch_read_error_and_deadline_exhausted(self) -> None:
|
cases = (
|
("ERROR", "UNREADABLE_ERROR", "ACTION_POST_DISPATCH", 1),
|
("DEADLINE_EXHAUSTED", "UNREADABLE_TIMEOUT", "ACTION_POST_DISPATCH", 0),
|
)
|
for observation_outcome, page_outcome, code_name, observation_count in cases:
|
with self.subTest(observation_outcome=observation_outcome):
|
begin = self.begin()
|
evidence = self.evidence(
|
begin, outcome=page_outcome, observations=None, marker=None,
|
action_outcome="POST_DISPATCH_ERROR", observation_outcome=observation_outcome,
|
observation_count=observation_count,
|
)
|
self.set_runtime_diagnostic(evidence, code_name)
|
if observation_count == 0:
|
evidence["runtime_observation"]["observation_elapsed_ms"] = 0
|
evidence["read_finished_at"] = evidence["refresh_finished_at"]
|
code, result = self.commit(begin, evidence)
|
self.assertEqual(4, code)
|
self.assertEqual("REFRESH_FAILED_PAGE_UNREADABLE", result["status"])
|
if observation_outcome == "ERROR":
|
Path(str(result["run_evidence_path"])).unlink()
|
(self.state / "refresh" / "runs" / "latest.json").unlink()
|
|
def test_runtime_timeout_cannot_confirm_no_new(self) -> None:
|
begin = self.begin()
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(),
|
action_outcome="TIMEOUT",
|
)
|
self.set_runtime_diagnostic(evidence, "ACTION_TIMEOUT")
|
code, result = self.commit(begin, evidence)
|
self.assertEqual(4, code)
|
self.assertEqual("PARTIAL_DISCOVERY_UNCONFIRMED", result["status"])
|
self.assertFalse(result["no_new_confirmed"])
|
|
def test_runtime_supported_control_latency_is_inside_new_budgets(self) -> None:
|
begin = self.begin()
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(),
|
)
|
evidence["refresh_finished_at"] = "2026-08-13T02:00:22.500000Z"
|
evidence["read_finished_at"] = "2026-08-13T02:00:44Z"
|
evidence["runtime_observation"]["refresh_action_elapsed_ms"] = 21500
|
evidence["runtime_observation"]["observation_elapsed_ms"] = 21500
|
evidence["controller_attestation"]["monotonic_action_finished_ms"] = 22500
|
evidence["controller_attestation"]["monotonic_observation_started_ms"] = 22500
|
evidence["controller_attestation"]["monotonic_observation_finished_ms"] = 44000
|
evidence["controller_attestation"]["monotonic_evidence_write_started_ms"] = 44000
|
code, result = self.commit(begin, evidence)
|
self.assertEqual(0, code)
|
self.assertEqual("REFRESH_CONFIRMED_NO_NEW", result["status"])
|
|
def test_runtime_post_dispatch_error_readable_new_saves_without_coverage(self) -> None:
|
begin = self.begin()
|
card, item = self.bound_card_item("987654")
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([card])], marker=self.end_marker(),
|
items=[item], action_outcome="POST_DISPATCH_ERROR",
|
)
|
self.set_runtime_diagnostic(evidence, "ACTION_POST_DISPATCH")
|
code, result = self.commit(begin, evidence)
|
self.assertEqual(0, code)
|
self.assertEqual("NEW_ITEMS_SAVED", result["status"])
|
self.assertFalse(result["coverage_complete"])
|
|
def test_reviewer_trusted_controller_calls_once_and_caller_forgery_fails(self) -> None:
|
begin = self.begin()
|
calls = {"tabs": 0, "reload": 0, "goto": 0, "evaluate": 0}
|
|
class Fake:
|
def open_tabs(inner) -> list[dict[str, object]]:
|
calls["tabs"] += 1
|
return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]
|
|
def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
|
calls["reload"] += 1
|
|
def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
|
calls["goto"] += 1
|
return {"url": url}
|
|
def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
|
calls["evaluate"] += 1
|
return {
|
"schema_version": 1,
|
"ready_state": "complete", "visibility_state": "visible",
|
"final_url": "https://space.bilibili.com/1420210197/dynamic",
|
"page_title": "青枫浦上Q个人动态-青枫浦上Q动态记录-哔哩哔哩视频",
|
"creator": {"uid": "1420210197", "name": "青枫浦上Q", "profile_url": "https://space.bilibili.com/1420210197"},
|
"cards": [], "unparsed_nodes": [], "terminal_marker_text": "已经到底了", "limit_hit": "NONE",
|
}
|
|
ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 0.023, 0.024])
|
path = controller.run_once(
|
collector.load_config(self.config_path), begin, Fake(), monotonic=lambda: next(ticks),
|
wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
|
)
|
self.assertEqual({"tabs": 1, "reload": 1, "goto": 0, "evaluate": 1}, calls)
|
code, result = collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertEqual((0, "REFRESH_CONFIRMED_NO_NEW"), (code, result["status"]))
|
|
Path(str(result["run_evidence_path"])).unlink()
|
(self.state / "refresh" / "runs" / "latest.json").unlink()
|
begin = self.begin()
|
forged = self.evidence(begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker())
|
forged["controller_attestation"]["binding_sha256"] = "0" * 64
|
path = Path(str(begin["evidence_path"]))
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(json.dumps(forged), encoding="utf-8")
|
with self.assertRaises(collector.CollectorError) as failure:
|
collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
|
self.assertIn(failure.exception.code, {"E_CONTROLLER_ATTESTATION", "E_CONTROLLER_REQUIRED"})
|
|
def test_reviewer_real_js_fixture_adapts_to_accepted_schema(self) -> None:
|
runner = PROJECT_DEV / "test" / "fixtures" / "bili_dynamic_collector" / "page_extract_fixture_runner.js"
|
expected = {
|
"new": (0, "NEW_ITEMS_SAVED", None),
|
"empty": (0, "REFRESH_CONFIRMED_NO_NEW", None),
|
"unparsed": (4, "PARTIAL_DISCOVERY_UNCONFIRMED", "E_COVERAGE_INCOMPLETE"),
|
"identity": (3, "REFRESH_BLOCKED_AUTH_OR_ACCESS", "E_CREATOR_MISMATCH"),
|
}
|
for fixture_case, wanted in expected.items():
|
with self.subTest(fixture_case=fixture_case):
|
self.reset_runtime_fixture()
|
completed = subprocess.run(
|
["node", str(runner), str(refresh.EXTRACTOR_SOURCE), fixture_case],
|
capture_output=True, check=True, timeout=10,
|
)
|
raw = json.loads(completed.stdout.decode("utf-8"))
|
begin = self.begin()
|
calls = {"reload": 0, "evaluate": 0}
|
|
class Fake:
|
def open_tabs(inner) -> list[dict[str, object]]:
|
return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]
|
|
def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
|
calls["reload"] += 1
|
|
def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
|
raise AssertionError("exact fixture tab must use reload")
|
|
def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
|
calls["evaluate"] += 1
|
self.assertIn("projectInfoCollectVisibleDynamicNodes({page_internal_settle_timeout_ms:15000})", source)
|
return raw
|
|
ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 0.023, 0.024])
|
path = controller.run_once(
|
collector.load_config(self.config_path), begin, Fake(),
|
monotonic=lambda: next(ticks),
|
wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
|
)
|
code, result = collector.run([
|
"--config", str(self.config_path), "refresh-commit", "--input", str(path),
|
"--now", "2026-08-13T10:00:25+08:00",
|
])
|
self.assertEqual(wanted, (code, result["status"], result["error_code"]))
|
self.assertEqual({"reload": 1, "evaluate": 1}, calls)
|
|
self.reset_runtime_fixture()
|
begin = self.begin()
|
|
class AccessBlocked:
|
def open_tabs(inner) -> list[dict[str, object]]:
|
return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]
|
|
def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
|
return None
|
|
def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
|
raise AssertionError("exact fixture tab must use reload")
|
|
def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
|
raise PermissionError("synthetic access interstitial")
|
|
ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 0.023, 0.024])
|
path = controller.run_once(
|
collector.load_config(self.config_path), begin, AccessBlocked(),
|
monotonic=lambda: next(ticks),
|
wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
|
)
|
code, result = collector.run([
|
"--config", str(self.config_path), "refresh-commit", "--input", str(path),
|
"--now", "2026-08-13T10:00:25+08:00",
|
])
|
self.assertEqual((3, "REFRESH_BLOCKED_AUTH_OR_ACCESS", "E_ACCESS_BLOCKED"), (code, result["status"], result["error_code"]))
|
|
def test_reviewer_deadline_plus_epsilon_never_binds_or_saves(self) -> None:
|
begin = self.begin()
|
evidence = self.evidence(begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker())
|
path = Path(str(begin["evidence_path"]))
|
self.write_json(path, evidence)
|
before = refresh._identity(self.archive / "manifest.jsonl").as_dict()
|
code, result = collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:02:01+08:00"])
|
self.assertEqual((4, "E_OVERALL_DEADLINE"), (code, result["error_code"]))
|
self.assertEqual(before, refresh._identity(self.archive / "manifest.jsonl").as_dict())
|
|
def test_reviewer_controller_commit_crossing_deadline_removes_unbound_evidence(self) -> None:
|
begin = self.begin()
|
|
class Fake:
|
def open_tabs(inner) -> list[dict[str, object]]:
|
return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]
|
|
def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
|
return None
|
|
def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
|
raise AssertionError("exact fixture tab must use reload")
|
|
def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
|
return {
|
"schema_version": 1, "ready_state": "complete", "visibility_state": "visible",
|
"final_url": "https://space.bilibili.com/1420210197/dynamic",
|
"page_title": "青枫浦上Q个人动态-青枫浦上Q动态记录-哔哩哔哩视频",
|
"creator": {"uid": "1420210197", "name": "青枫浦上Q", "profile_url": "https://space.bilibili.com/1420210197"},
|
"cards": [], "unparsed_nodes": [], "terminal_marker_text": "已经到底了", "limit_hit": "NONE",
|
}
|
|
ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 119.999, 120.001])
|
with self.assertRaises(collector.CollectorError) as failure:
|
controller.run_once(
|
collector.load_config(self.config_path), begin, Fake(),
|
monotonic=lambda: next(ticks),
|
wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
|
)
|
self.assertEqual("E_OVERALL_DEADLINE", failure.exception.code)
|
self.assertFalse(Path(str(begin["evidence_path"])).exists())
|
pending = refresh._load_pending(collector.load_config(self.config_path))
|
self.assertIsNotNone(pending)
|
self.assertEqual("AWAITING_EVIDENCE", pending["phase"])
|
self.assertFalse((self.archive / "manifest.jsonl").exists())
|
|
def test_reviewer_legacy_pending_never_requests_browser_action(self) -> None:
|
for slot_present in (False, True):
|
for after_deadline in (False, True):
|
for entrypoint in ("begin", "commit", "controller"):
|
with self.subTest(slot_present=slot_present, after_deadline=after_deadline, entrypoint=entrypoint):
|
self.reset_runtime_fixture()
|
begin = self.begin()
|
config = collector.load_config(self.config_path)
|
pending = refresh._load_pending(config)
|
assert pending is not None
|
pending["schema_version"] = refresh.LEGACY_PENDING_SCHEMA
|
pending.pop("runtime_contract")
|
pending.pop("controller_capability")
|
refresh._write_pending(config, pending)
|
if not slot_present:
|
Path(str(begin["run_evidence_path"])).unlink()
|
now_text = "2026-08-13T10:02:01+08:00" if after_deadline else "2026-08-13T10:00:30+08:00"
|
if entrypoint == "controller":
|
class NeverCalled:
|
def open_tabs(inner) -> list[dict[str, object]]:
|
raise AssertionError("legacy pending must not touch the browser adapter")
|
|
with self.assertRaises(collector.CollectorError) as failure:
|
controller.run_once(config, begin, NeverCalled())
|
self.assertEqual("E_CONTROLLER_PENDING", failure.exception.code)
|
continue
|
if entrypoint == "begin":
|
invoke = lambda: refresh.refresh_begin(config, self.config_path, collector.parse_datetime(now_text, "now"))
|
else:
|
invoke = lambda: refresh.refresh_commit(config, self.config_path, Path(str(begin["evidence_path"])), collector.parse_datetime(now_text, "now"))
|
if after_deadline:
|
result = invoke()
|
self.assertEqual(("REFRESH_FAILED_PAGE_UNREADABLE", "E_LEGACY_EVIDENCE_MISSING_AFTER_DEADLINE"), (result["status"], result["error_code"]))
|
self.assertNotEqual("BROWSER_REFRESH_REQUIRED", result["status"])
|
else:
|
with self.assertRaises(collector.CollectorError) as failure:
|
invoke()
|
self.assertEqual("E_LEGACY_RECOVERY_ONLY", failure.exception.code)
|
self.assertFalse((self.archive / "manifest.jsonl").exists())
|
|
def test_reviewer_identity_mismatch_is_durable_blocked_terminal(self) -> None:
|
identities = {
|
"uid": lambda evidence: evidence["creator"].__setitem__("uid", "1"),
|
"name": lambda evidence: evidence["creator"].__setitem__("name", "其他用户"),
|
"profile": lambda evidence: evidence["creator"].__setitem__("profile_url", "https://space.bilibili.com/1"),
|
"final_url": lambda evidence: evidence.__setitem__("final_url", "https://space.bilibili.com/1/dynamic"),
|
}
|
outcomes = {
|
"CONFIRMED": "NONE",
|
"TIMEOUT": "ACTION_TIMEOUT",
|
"POST_DISPATCH_ERROR": "ACTION_POST_DISPATCH",
|
}
|
for identity_name, mutate in identities.items():
|
for action_outcome, diagnostic in outcomes.items():
|
with self.subTest(identity=identity_name, action_outcome=action_outcome):
|
self.reset_runtime_fixture()
|
begin = self.begin()
|
evidence = self.evidence(
|
begin, outcome="READABLE", observations=[self.observation([])],
|
marker=self.end_marker(), action_outcome=action_outcome,
|
)
|
self.set_runtime_diagnostic(evidence, diagnostic)
|
mutate(evidence)
|
code, result = self.commit(begin, evidence)
|
self.assertEqual((3, "REFRESH_BLOCKED_AUTH_OR_ACCESS", "E_CREATOR_MISMATCH"), (code, result["status"], result["error_code"]))
|
slot = json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8"))
|
self.assertEqual("TERMINAL", slot["run_state"])
|
self.assertFalse(refresh._pending_path(collector.load_config(self.config_path)).exists())
|
self.assertFalse(result["no_new_confirmed"])
|
self.assertFalse((self.archive / "manifest.jsonl").exists())
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
