from __future__ import annotations
|
|
import hashlib
|
import io
|
import json
|
import errno
|
import shutil
|
import subprocess
|
import sys
|
import tempfile
|
import unittest
|
from datetime import datetime, timezone
|
from pathlib import Path
|
|
PROJECT_DEV = Path(__file__).resolve().parents[2]
|
ROOT = PROJECT_DEV.parents[1]
|
sys.path.insert(0, str(PROJECT_DEV))
|
|
import bili_dynamic_collector as collector
|
import bili_dynamic_refresh_controller as public_controller
|
from bili_dynamic_refresh_native_host.constants import EXTENSION_ID, EXTENSION_ORIGIN, EXTENSION_NAME, HOST_NAME
|
from bili_dynamic_refresh_native_host.durable import PendingStore
|
from bili_dynamic_refresh_native_host.identity import IdentityError, verify_local_identity, verify_source_tree
|
from bili_dynamic_refresh_native_host.native_host import NativeFrameWriter
|
from bili_dynamic_refresh_native_host.protocol import HostSession, ProtocolError, sign_extension_frame
|
|
EXTENSION_ROOT = PROJECT_DEV / "bili_dynamic_refresh_extension"
|
HOST_ROOT = PROJECT_DEV / "bili_dynamic_refresh_native_host"
|
|
|
class V009ContractTests(unittest.TestCase):
|
def copy_extension(self, root: Path) -> Path:
|
target = root / "extension"
|
shutil.copytree(EXTENSION_ROOT, target)
|
return target
|
|
def facts(self, source_root: Path) -> dict[str, object]:
|
source = verify_source_tree(source_root)
|
manifest_payload = (source_root / "manifest.json").read_bytes()
|
receipt_sha = "2" * 64
|
host_manifest_sha = "3" * 64
|
chrome_sha = "4" * 64
|
started = "2026-08-14T16:00:00+08:00"
|
source_approval = {
|
"schema": 1, "task_id": "DEV-PROJECT-INFO-BILI-DYNAMIC-REFRESH-COLLECTOR-20260813-001",
|
"scope": "controlled-local-unpacked-source-manifest",
|
"source_manifest_path": str(source_root / "source-artifact-manifest.json"),
|
"source_manifest_bytes": source["manifest_bytes"], "source_manifest_sha256": source["manifest_sha256"],
|
"payload_tree_sha256": source["payload_tree_sha256"], "manifest_bytes": len(manifest_payload),
|
"manifest_sha256": hashlib.sha256(manifest_payload).hexdigest(), "extension_id": EXTENSION_ID,
|
"approved_by_role": "dev.reviewer.project", "review_audit_id": "AUDIT", "created_at": started,
|
}
|
load = {
|
"schema": 1, "task_id": source_approval["task_id"], "scope": "local-unpacked-load",
|
"account_holder_confirmed": True, "observed_extension_id": EXTENSION_ID,
|
"observed_name": EXTENSION_NAME, "observed_version": "1.0.0", "observed_enabled": True,
|
"observed_error_count": 0, "source_root_absolute": str(source_root),
|
"source_manifest_bytes": source["manifest_bytes"], "source_manifest_sha256": source["manifest_sha256"],
|
"payload_tree_sha256": source["payload_tree_sha256"], "host_install_receipt_bytes": 100,
|
"host_install_receipt_sha256": receipt_sha, "host_manifest_sha256": host_manifest_sha,
|
"chrome_parent_pid": 1234, "chrome_parent_path_sha256": chrome_sha,
|
"chrome_parent_signature_verified": True, "chrome_parent_started_at": started,
|
"observed_at": started, "approved_by_role": "project.admin", "approval_reason": "synthetic-test-only",
|
}
|
host = {
|
"host_name": HOST_NAME, "allowed_origins": [EXTENSION_ORIGIN], "install_receipt_bytes": 100,
|
"install_receipt_sha256": receipt_sha, "host_manifest_sha256": host_manifest_sha,
|
"chrome_parent_pid": 1234, "chrome_parent_path_sha256": chrome_sha,
|
"chrome_parent_signature_verified": True, "chrome_parent_started_at": started,
|
}
|
return {"source_approval": source_approval, "load_approval": load, "host": host}
|
|
def test_manifest_identity_exact_set_and_forbidden_surfaces(self) -> None:
|
manifest = json.loads((EXTENSION_ROOT / "manifest.json").read_text(encoding="utf-8"))
|
self.assertEqual((3, "1.0.0", "120"), (manifest["manifest_version"], manifest["version"], manifest["minimum_chrome_version"]))
|
self.assertEqual(["alarms", "nativeMessaging", "scripting", "storage", "tabs"], manifest["permissions"])
|
self.assertEqual(["https://space.bilibili.com/*/dynamic*"], manifest["host_permissions"])
|
self.assertEqual({"service_worker": "service_worker.js", "type": "module"}, manifest["background"])
|
self.assertEqual(EXTENSION_ID, verify_source_tree(EXTENSION_ROOT) and EXTENSION_ID)
|
verify_source_tree(HOST_ROOT)
|
product = "\n".join(path.read_text(encoding="utf-8") for root in (EXTENSION_ROOT, HOST_ROOT) for path in root.rglob("*") if path.is_file())
|
for forbidden in ("chrome.cookies", "localStorage", "remote-debugging", "ExtensionInstallForcelist", "ExtensionSettings", "clients2.google.com"):
|
self.assertNotIn(forbidden, product)
|
self.assertEqual([EXTENSION_ORIGIN], json.loads((HOST_ROOT / "native-host-manifest.template.json").read_text(encoding="utf-8"))["allowed_origins"])
|
native_main = (HOST_ROOT / "native_host.py").read_text(encoding="utf-8")
|
for forbidden in ("argparse", "os.environ", "--test-peer", "--fixture", "--transcript"):
|
self.assertNotIn(forbidden, native_main)
|
|
runtime_source = (EXTENSION_ROOT / "runtime.js").read_text(encoding="utf-8")
|
worker_source = (EXTENSION_ROOT / "service_worker.js").read_text(encoding="utf-8")
|
self.assertNotIn("about:blank", runtime_source + worker_source)
|
self.assertNotIn("tabs.update", runtime_source + worker_source)
|
self.assertNotIn("node_repl", runtime_source + worker_source)
|
self.assertIn("periodInMinutes: PERIOD_MINUTES", worker_source)
|
|
def test_external_source_and_visible_load_identity_fail_closed(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
source_root = self.copy_extension(Path(raw))
|
facts = self.facts(source_root)
|
verified = verify_local_identity(source_root, facts)
|
self.assertEqual((EXTENSION_ID, 1234), (verified.extension_id, verified.chrome_parent_pid))
|
mutations = [
|
("source_approval", "approved_by_role", "dev.developer.project.secondary"),
|
("source_approval", "scope", "caller-self-report"),
|
("load_approval", "observed_extension_id", "a" * 32),
|
("load_approval", "observed_enabled", False),
|
("load_approval", "observed_error_count", 1),
|
("load_approval", "approved_by_role", "extension"),
|
("host", "allowed_origins", ["chrome-extension://" + "a" * 32 + "/"]),
|
("host", "chrome_parent_pid", 5678),
|
]
|
for section, key, value in mutations:
|
with self.subTest(section=section, key=key):
|
changed = json.loads(json.dumps(facts))
|
changed[section][key] = value
|
with self.assertRaises(IdentityError):
|
verify_local_identity(source_root, changed)
|
(source_root / "extra.js").write_text("caller-authored", encoding="utf-8")
|
with self.assertRaises(IdentityError):
|
verify_local_identity(source_root, facts)
|
|
def test_actual_host_protocol_durable_permit_and_single_commit(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
temp = Path(raw)
|
source_root = self.copy_extension(temp)
|
identity = verify_local_identity(source_root, self.facts(source_root))
|
committed: list[dict[str, object]] = []
|
session = HostSession(identity, PendingStore(temp / "pending.json"), lambda evidence: committed.append(dict(evidence)) or {"status": "REFRESH_CONFIRMED_NO_NEW"}, "https://space.bilibili.com/1420210197/dynamic", now=lambda: datetime(2026, 8, 14, 8, tzinfo=timezone.utc))
|
hello = {"schema_version": 1, "type": "EXTENSION_HELLO", "sequence": 1, "extension_id": EXTENSION_ID, "version": "1.0.0", "manifest_name": EXTENSION_NAME}
|
challenge = session.start(hello, run_id="run", request_id="request", deadline_at="2026-08-14T08:02:00Z")
|
secret = challenge["secret"]
|
accepted = sign_extension_frame(secret, {"schema_version": 1, "type": "EXTENSION_CHALLENGE_ACCEPTED", "run_id": "run", "request_id": "request", "sequence": 3, "challenge_id": challenge["challenge_id"]})
|
prepare = session.accept(accepted)
|
ready = sign_extension_frame(secret, {"schema_version": 1, "type": "EXTENSION_READY_TO_DISPATCH", "run_id": "run", "request_id": "request", "sequence": 5, "action_id": prepare["action"]["action_id"], "prepared": {"action_id": prepare["action"]["action_id"]}})
|
permit = session.accept(ready)
|
pending = PendingStore(temp / "pending.json").load()
|
self.assertEqual((True, 1, 0), (pending["action_budget_consumed"], pending["refresh_count"], pending["retry_count"]))
|
with self.assertRaises(Exception):
|
session.accept(ready)
|
result = sign_extension_frame(secret, {"schema_version": 1, "type": "EXTENSION_ACTION_RESULT", "run_id": "run", "request_id": "request", "sequence": 7, "permit_id": permit["permit"]["permit_id"], "result": {"tab_id": 7}})
|
request = session.accept(result)
|
observation = sign_extension_frame(secret, {"schema_version": 1, "type": "EXTENSION_OBSERVATION", "run_id": "run", "request_id": "request", "sequence": 9, "observation": {"cards": [], "coverage_complete": True}})
|
final = session.accept(observation)
|
self.assertEqual("HOST_COMMIT_RESULT", final["type"])
|
self.assertEqual(1, len(committed))
|
|
def test_expired_original_deadline_stops_before_dispatch_permit(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
temp = Path(raw)
|
source_root = self.copy_extension(temp)
|
identity = verify_local_identity(source_root, self.facts(source_root))
|
current = [datetime(2026, 8, 14, 8, tzinfo=timezone.utc)]
|
store = PendingStore(temp / "pending.json")
|
session = HostSession(identity, store, lambda evidence: {}, "https://space.bilibili.com/1420210197/dynamic", now=lambda: current[0])
|
hello = {"schema_version": 1, "type": "EXTENSION_HELLO", "sequence": 1, "extension_id": EXTENSION_ID, "version": "1.0.0", "manifest_name": EXTENSION_NAME}
|
challenge = session.start(hello, run_id="run", request_id="request", deadline_at="2026-08-14T08:02:00Z")
|
secret = challenge["secret"]
|
accepted = sign_extension_frame(secret, {"schema_version": 1, "type": "EXTENSION_CHALLENGE_ACCEPTED", "run_id": "run", "request_id": "request", "sequence": 3, "challenge_id": challenge["challenge_id"]})
|
prepare = session.accept(accepted)
|
current[0] = datetime(2026, 8, 14, 8, 2, 0, 1, tzinfo=timezone.utc)
|
ready = sign_extension_frame(secret, {"schema_version": 1, "type": "EXTENSION_READY_TO_DISPATCH", "run_id": "run", "request_id": "request", "sequence": 5, "action_id": prepare["action"]["action_id"], "prepared": {}})
|
with self.assertRaises(ProtocolError) as failure:
|
session.accept(ready)
|
self.assertEqual("E_OVERALL_DEADLINE", failure.exception.code)
|
pending = store.load()
|
self.assertEqual((False, 0, 0), (pending["action_budget_consumed"], pending["refresh_count"], pending["retry_count"]))
|
|
def test_real_subprocess_crash_recovery_is_bounded(self) -> None:
|
child = Path(__file__).with_name("crash_child.py")
|
cases = {
|
"after_open": (0, False),
|
"after_file_fsync": (0, False),
|
"after_replace": (1, True),
|
"after_directory_fsync": (1, True),
|
}
|
for phase, expected in cases.items():
|
with self.subTest(phase=phase), tempfile.TemporaryDirectory() as raw:
|
path = Path(raw) / "pending.json"
|
completed = subprocess.run([sys.executable, "-B", str(child), str(path), phase], timeout=20)
|
self.assertEqual(79, completed.returncode)
|
projection = PendingStore(path).recovery_projection()
|
self.assertEqual(expected, (projection["refresh_count"], projection["may_have_dispatched"]))
|
self.assertEqual(0, projection["retry_count"])
|
|
def test_corrupt_durable_state_projects_conservative_single_action(self) -> None:
|
with tempfile.TemporaryDirectory() as raw:
|
path = Path(raw) / "pending.json"
|
path.write_bytes(b'{"schema":1,"broken":true}\n')
|
projection = PendingStore(path).recovery_projection()
|
self.assertEqual(("E_DISPATCH_DURABILITY_AMBIGUOUS", 1, 0, True), (projection["error_code"], projection["refresh_count"], projection["retry_count"], projection["may_have_dispatched"]))
|
|
def test_real_service_worker_reducer_and_extractor(self) -> None:
|
completed = subprocess.run(
|
["node", str(Path(__file__).with_name("js_contract.mjs")), str(EXTENSION_ROOT / "protocol.js"), str(EXTENSION_ROOT / "page_extract.js")],
|
text=True, capture_output=True, timeout=20, check=True,
|
)
|
result = json.loads(completed.stdout)
|
self.assertEqual({"prepare": 1, "dispatch": 1, "observe": 1}, result["calls"])
|
|
lifecycle = subprocess.run(
|
["node", str(Path(__file__).with_name("runtime_lifecycle.mjs")), str(EXTENSION_ROOT / "runtime.js")],
|
text=True, capture_output=True, timeout=30, check=True,
|
)
|
lifecycle_result = json.loads(lifecycle.stdout)
|
self.assertEqual((100, 0, True), (
|
lifecycle_result["slots"], lifecycle_result["tabs_update"], lifecycle_result["user_tab_preserved"]
|
))
|
|
reproduction = subprocess.run(
|
["node", str(Path(__file__).with_name("codex_stdio_race_repro.mjs"))],
|
text=True, capture_output=True, timeout=20, check=True,
|
)
|
reproduction_result = json.loads(reproduction.stdout)
|
self.assertEqual((True, "E_NATIVE_PEER_CLOSED"), (
|
reproduction_result["old_ordering_uncaught"], reproduction_result["bounded_error_code"]
|
))
|
|
def test_native_frame_writer_sanitizes_only_peer_close(self) -> None:
|
class Stream:
|
def __init__(self, *, write_error: BaseException | None = None, flush_error: BaseException | None = None) -> None:
|
self.write_error = write_error
|
self.flush_error = flush_error
|
self.payloads: list[bytes] = []
|
self.flushes = 0
|
|
def write(self, payload: bytes) -> None:
|
if self.write_error is not None:
|
raise self.write_error
|
self.payloads.append(payload)
|
|
def flush(self) -> None:
|
self.flushes += 1
|
if self.flush_error is not None:
|
raise self.flush_error
|
|
for error in (BrokenPipeError(), EOFError(), OSError(errno.EPIPE, "pipe closed")):
|
with self.subTest(error=type(error).__name__):
|
stream = Stream(write_error=error)
|
writer = NativeFrameWriter(stream) # type: ignore[arg-type]
|
first = writer.write_frame({"schema_version": 1})
|
second = writer.write_frame({"schema_version": 1})
|
self.assertEqual((False, "E_NATIVE_PEER_CLOSED"), (first.written, first.error_code))
|
self.assertEqual(first, second)
|
self.assertEqual([], stream.payloads)
|
|
flush_stream = Stream(flush_error=BrokenPipeError())
|
flushed = NativeFrameWriter(flush_stream).write_frame({"schema_version": 1}) # type: ignore[arg-type]
|
self.assertEqual((False, "E_NATIVE_PEER_CLOSED", 1, 1), (
|
flushed.written, flushed.error_code, len(flush_stream.payloads), flush_stream.flushes
|
))
|
|
with self.assertRaises(PermissionError):
|
NativeFrameWriter(Stream(write_error=PermissionError("denied"))).write_frame({"schema_version": 1}) # type: ignore[arg-type]
|
|
def test_public_controller_does_not_read_caller_transcript(self) -> None:
|
stream = io.StringIO('{"schema_version":4,"authoritative":true,"saved":true,"no_new":true}\n')
|
with self.assertRaises(collector.CollectorError) as failure:
|
public_controller.run_product(None, Path("unused"), datetime.now(timezone.utc), input_stream=stream, output_stream=io.StringIO()) # type: ignore[arg-type]
|
self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", failure.exception.code)
|
self.assertEqual({"authoritative": False, "saved": False, "no_new": False}, failure.exception.details)
|
self.assertEqual(0, stream.tell())
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
