From 88ccf6fd6c18ef6cad3ce12015b585d493a60aa3 Mon Sep 17 00:00:00 2001
From: MB-X Bilibili Pipeline <mbx-bili-pipeline@localhost>
Date: Sat, 05 Sep 2026 22:25:04 +0800
Subject: [PATCH] chore(project-info): archive Bilibili dynamic 1f35c538413b
---
dev/project-dev/test/test_bili_dynamic_refresh.py | 405 ++++++++++++++++++++++++++-------------------------------
1 files changed, 184 insertions(+), 221 deletions(-)
diff --git a/dev/project-dev/test/test_bili_dynamic_refresh.py b/dev/project-dev/test/test_bili_dynamic_refresh.py
index 9d05344..669e44a 100644
--- a/dev/project-dev/test/test_bili_dynamic_refresh.py
+++ b/dev/project-dev/test/test_bili_dynamic_refresh.py
@@ -1,11 +1,14 @@
from __future__ import annotations
import hashlib
+import hmac
import importlib.util
+import io
import json
import gc
import os
import shutil
+import secrets
import subprocess
import sys
import tempfile
@@ -42,6 +45,7 @@
for path in (self.archive, self.intake, self.downloads, self.videos):
path.mkdir()
self.config_path = self.root / "config.json"
+ self.controller_keys: dict[str, bytes] = {}
self.write_json(
self.config_path,
{
@@ -93,20 +97,7 @@
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"]),
- )
+ self.sign_evidence(value)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
@@ -116,12 +107,30 @@
if path.exists():
shutil.rmtree(path)
path.mkdir()
+ self.controller_keys.clear()
def begin(self) -> dict[str, object]:
- code, result = collector.run(["--config", str(self.config_path), "refresh-begin", "--now", NOW])
- self.assertEqual(0, code)
+ key = secrets.token_bytes(32)
+ config = collector.load_config(self.config_path)
+ result = refresh.refresh_begin(
+ config,
+ self.config_path,
+ collector.parse_datetime(NOW, "test now"),
+ _controller_key_commitment=hashlib.sha256(key).hexdigest(),
+ )
self.assertEqual("BROWSER_REFRESH_REQUIRED", result["status"])
+ self.controller_keys[str(result["run_id"])] = key
return result
+
+ def sign_evidence(self, evidence: dict[str, object]) -> None:
+ key = self.controller_keys.get(str(evidence.get("run_id")))
+ attestation = evidence.get("controller_attestation")
+ if key is None or not isinstance(attestation, dict):
+ return
+ attestation["binding_sha256"] = None
+ attestation["binding_sha256"] = hmac.new(
+ key, refresh._controller_attestation_payload(evidence), hashlib.sha256
+ ).hexdigest()
def formal_pending(self, config: collector.CollectorConfig) -> dict[str, object]:
return {
@@ -244,9 +253,6 @@
"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:
@@ -262,17 +268,22 @@
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,
- )
+ runtime = collector.load_config(self.config_path).refresh
+ assert runtime is not None
+ evidence["controller_attestation"] = {
+ "controller_id": "bili-supported-chrome-controller-v1",
+ "controller_sha256": hashlib.sha256(refresh.CONTROLLER_SOURCE.read_bytes()).hexdigest(),
+ "binding_algorithm": "hmac-sha256-controller-envelope-v1",
+ "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,
+ "binding_sha256": None,
+ }
+ self.sign_evidence(evidence)
return evidence
def set_runtime_diagnostic(self, evidence: dict[str, object], code: str) -> None:
@@ -297,33 +308,36 @@
}
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 = self.bind_for_test(begin, evidence)
+ return collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
+
+ def bind_for_test(self, begin: dict[str, object], evidence: dict[str, object]) -> Path:
+ self.sign_evidence(evidence)
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"),
+ key = self.controller_keys[str(begin["run_id"])]
+ value, payload, _, _ = refresh._validate_evidence(config, pending, path, _controller_key=key)
+ pending["evidence_identity"] = {
+ "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest()
+ }
+ pending["controller_binding_sha256"] = value["controller_attestation"]["binding_sha256"]
+ pending["phase"] = "EVIDENCE_BOUND"
+ pending["last_transition_at"] = "2026-08-13T02:00:24Z"
+ refresh._write_pending(config, pending)
+ return path
+
+ def validate_for_test(self, begin: dict[str, object], evidence: dict[str, object]) -> None:
+ 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._validate_evidence(
+ config, pending, path, _controller_key=self.controller_keys[str(begin["run_id"])]
)
- 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]]:
@@ -447,10 +461,10 @@
}
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.validate_for_test(
+ begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=None)
+ )
self.assertFalse((self.archive / "manifest.jsonl").exists())
def test_complete_observed_card_with_missing_item_cannot_claim_no_new(self) -> None:
@@ -460,10 +474,8 @@
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.validate_for_test(begin, evidence)
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
self.assertFalse((self.archive / "manifest.jsonl").exists())
@@ -474,10 +486,8 @@
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.validate_for_test(begin, evidence)
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
self.assertFalse((self.archive / "manifest.jsonl").exists())
@@ -489,10 +499,8 @@
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.validate_for_test(begin, evidence)
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
def test_observation_overlapping_component_identity_fails_closed(self) -> None:
@@ -510,10 +518,8 @@
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.validate_for_test(begin, evidence)
self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
def test_new_text_commits_both_manifests_and_artifact(self) -> None:
@@ -565,24 +571,23 @@
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"])
+ refresh.refresh_begin(
+ collector.load_config(self.config_path), self.config_path,
+ collector.parse_now("2026-08-13T10:02:10+08:00"),
+ _controller_key_commitment=hashlib.sha256(b"next-controller").hexdigest(),
+ )
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()
+ config = collector.load_config(self.config_path)
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"),
- )
+ self.bind_for_test(begin, evidence)
original = refresh._write_readback
injected = {"done": False}
@@ -628,7 +633,7 @@
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:
+ def test_current_formal_catalog_shape_is_140_111_103_8_0_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")
@@ -648,7 +653,7 @@
),
)
_, _, counts = refresh.load_formal_catalog(rebound)
- self.assertEqual({"events": 48, "components": 22, "saved": 16, "video": 5, "retryable": 1}, counts)
+ self.assertEqual({"events": 140, "components": 111, "saved": 103, "video": 8, "retryable": 0}, 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:
@@ -723,7 +728,10 @@
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))
+ refresh.refresh_begin(
+ config, self.config_path, collector.parse_now(NOW),
+ _controller_key_commitment=hashlib.sha256(b"test-controller").hexdigest(),
+ )
finally:
refresh._ensure_started_slot = original
pending = refresh._load_pending(config)
@@ -1174,10 +1182,8 @@
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.validate_for_test(begin, evidence)
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()
@@ -1261,127 +1267,109 @@
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")
+ input_stream = io.StringIO('{"page_authoritative":true,"no_new":true}\n')
+ output_stream = io.StringIO()
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"})
+ controller.run_product(
+ collector.load_config(self.config_path), self.config_path,
+ collector.parse_datetime(NOW, "now"), input_stream=input_stream,
+ output_stream=output_stream,
+ )
+ self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", failure.exception.code)
+ self.assertEqual({"authoritative": False, "saved": False, "no_new": False}, failure.exception.details)
+ self.assertEqual(0, input_stream.tell())
+ self.assertEqual("", output_stream.getvalue())
+ self.assertFalse((self.state / "refresh" / "pending.json").exists())
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():
+ for fixture_case in ("new", "empty", "unparsed", "identity", "access"):
with self.subTest(fixture_case=fixture_case):
self.reset_runtime_fixture()
+ raw = None
+ if fixture_case != "access":
+ 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"))
+
+ def response(request_id: int, result: object, *, error: str | None = None) -> str:
+ return json.dumps({
+ "schema_version": 1,
+ "type": "supported_chrome_response",
+ "request_id": request_id,
+ "ok": error is None,
+ "result": result,
+ "error_code": error,
+ }, ensure_ascii=False)
+
+ protocol_input = "\n".join([
+ response(1, [{"tab_id": "fixture-tab", "url": "https://space.bilibili.com/1420210197/dynamic"}]),
+ response(2, None),
+ response(3, raw, error="ACCESS_BLOCKED" if fixture_case == "access" else None),
+ ]) + "\n"
completed = subprocess.run(
- ["node", str(runner), str(refresh.EXTRACTOR_SOURCE), fixture_case],
- capture_output=True, check=True, timeout=10,
+ [
+ sys.executable, "-B", str(COLLECTOR_PATH), "--config", str(self.config_path),
+ "refresh-run", "--now", NOW,
+ ],
+ input=protocol_input, text=True, capture_output=True, timeout=20,
)
- raw = json.loads(completed.stdout.decode("utf-8"))
- begin = self.begin()
- calls = {"reload": 0, "evaluate": 0}
+ lines = [json.loads(line) for line in completed.stdout.splitlines()]
+ requests = [line for line in lines if line.get("type") == "supported_chrome_request"]
+ terminal = lines[-1]
+ self.assertEqual([], requests)
+ self.assertEqual((3, "SAFETY_STOP", "E_TRUSTED_ADAPTER_REQUIRED"), (completed.returncode, terminal["status"], terminal["error_code"]))
+ self.assertEqual({"authoritative": False, "saved": False, "no_new": False}, terminal["details"])
+ self.assertFalse((self.state / "refresh" / "pending.json").exists())
- class Fake:
- def open_tabs(inner) -> list[dict[str, object]]:
- return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]
+ def test_reviewer_public_cli_cannot_create_or_bind_caller_schema3(self) -> None:
+ self.assertFalse(hasattr(refresh, "_attest_controller_evidence"))
+ self.assertFalse(hasattr(refresh, "bind_controller_evidence"))
+ self.assertFalse(hasattr(controller, "_attest_controller_evidence"))
+ self.assertFalse(hasattr(controller, "bind_controller_evidence"))
+ with self.assertRaises(collector.CollectorError) as blocked:
+ collector.run(["--config", str(self.config_path), "refresh-begin", "--now", NOW])
+ self.assertEqual("E_CONTROLLER_ENTRY_REQUIRED", blocked.exception.code)
+ config = collector.load_config(self.config_path)
+ self.assertFalse(refresh._pending_path(config).exists())
- 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()
+ pending = refresh._load_pending(config)
+ self.assertNotIn("controller_capability", pending)
+ self.assertRegex(str(pending["controller_key_commitment"]), r"^[0-9a-f]{64}$")
- 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"),
+ forged = self.evidence(
+ begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker()
)
- 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"]))
+ path = Path(str(begin["evidence_path"]))
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(json.dumps(forged, ensure_ascii=False), encoding="utf-8")
+ with self.assertRaises(collector.CollectorError) as rejected:
+ collector.run([
+ "--config", str(self.config_path), "refresh-commit", "--input", str(path),
+ "--now", "2026-08-13T10:00:25+08:00",
+ ])
+ self.assertEqual("E_CONTROLLER_REQUIRED", rejected.exception.code)
+ self.assertFalse((self.archive / "manifest.jsonl").exists())
+
+ def test_reviewer_expired_pending_is_zero_adapter_calls_and_zero_evidence(self) -> None:
+ begin = self.begin()
+ protocol_input = io.StringIO('{"page_authoritative":true}\n')
+ protocol_output = io.StringIO()
+ with self.assertRaises(collector.CollectorError) as expired:
+ controller.run_product(
+ collector.load_config(self.config_path), self.config_path,
+ collector.parse_datetime("2026-08-13T10:02:00.001+08:00", "wall"),
+ input_stream=protocol_input, output_stream=protocol_output,
+ )
+ self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", expired.exception.code)
+ self.assertEqual(0, protocol_input.tell())
+ self.assertEqual("", protocol_output.getvalue())
+ self.assertFalse(Path(str(begin["evidence_path"])).exists())
+ pending = refresh._load_pending(collector.load_config(self.config_path))
+ self.assertEqual("AWAITING_EVIDENCE", pending["phase"])
def test_reviewer_deadline_plus_epsilon_never_binds_or_saves(self) -> None:
begin = self.begin()
@@ -1395,34 +1383,12 @@
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"),
+ controller.run_product(
+ collector.load_config(self.config_path), self.config_path,
+ collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
)
- self.assertEqual("E_OVERALL_DEADLINE", failure.exception.code)
+ self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", failure.exception.code)
self.assertFalse(Path(str(begin["evidence_path"])).exists())
pending = refresh._load_pending(collector.load_config(self.config_path))
self.assertIsNotNone(pending)
@@ -1441,19 +1407,16 @@
assert pending is not None
pending["schema_version"] = refresh.LEGACY_PENDING_SCHEMA
pending.pop("runtime_contract")
- pending.pop("controller_capability")
+ pending.pop("controller_key_commitment")
+ pending.pop("controller_binding_sha256")
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)
+ controller.run_product(config, self.config_path, collector.parse_datetime(now_text, "now"))
+ self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", failure.exception.code)
continue
if entrypoint == "begin":
invoke = lambda: refresh.refresh_begin(config, self.config_path, collector.parse_datetime(now_text, "now"))
--
Gitblit v1.9.3