- MockHostAdapter (tests/mock_host): in-memory editing model with opaque entity ids, id invalidation, token-bucket rate limiting, snapshot/restore with cap, event subscription, and a stdlib-zlib PNG encoder producing time-varying non-black frames - HostToolExecutor: all 26 tool handlers mapped to HostAdapter calls, snapshot orchestration for mutating batches, concurrent scan_timeline frame pipeline with token bucket and RateLimited backoff, undo_session() restoring per-batch snapshots in reverse, get_params min/max/choices reflection - Web UI (core[webui] extra): FastAPI + single-page vanilla JS, chat log, pending-confirmation list with approve/reject endpoints (timeout defaults to reject), progress, undo-session, snapshot management; SSE push + POST, bound to 127.0.0.1 - Host error types (HostError/EntityNotFound/RateLimited) and shared TokenBucket - P2 acceptance chain: import -> place -> split -> ripple delete -> effect + param -> frame verification (non-black, pixel changes at cut and after effect) -> undo_session restores original state Tests: 129 passed via uv run pytest (no network)
86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
"""P2 验收链路:mock 宿主上"导入 → 铺轨 → 切/重建 → 删除 → 加效果 → 取帧验证 →
|
||
整段会话撤销"全链路自动化通过(计划文档 §4 P2 验收)。
|
||
|
||
关键断言:帧非全黑、编辑前后切口帧像素不同、undo_session 后时间线恢复原状。
|
||
"""
|
||
import hashlib
|
||
|
||
from mock_host import MockHostAdapter, decode_png, frame_is_black
|
||
from quercus_core.tools.executor import HostToolExecutor
|
||
from quercus_core.types import Action, Rational
|
||
|
||
FRAME_SIZE = {"width": 64, "height": 36}
|
||
|
||
|
||
def r(x: int) -> dict:
|
||
return {"num": x, "den": 1}
|
||
|
||
|
||
def _frame_at(executor, t_json, size=FRAME_SIZE):
|
||
result = executor.execute(Action(tool="get_frame", params={"time": t_json, "max_size": size}))
|
||
assert result.ok, f"get_frame 失败: {result.summary}"
|
||
assert len(result.images) == 1
|
||
return result.images[0]
|
||
|
||
|
||
def _sha(png: bytes) -> str:
|
||
return hashlib.sha256(png).hexdigest()
|
||
|
||
|
||
def _first_clips(adapter):
|
||
seq_id = adapter.get_project_overview().timeline_ids[0]
|
||
return adapter.get_timeline_structure(seq_id).tracks[0].clips
|
||
|
||
|
||
def test_p2_acceptance_chain():
|
||
adapter = MockHostAdapter()
|
||
executor = HostToolExecutor(adapter)
|
||
|
||
def act(tool, **params):
|
||
result = executor.execute(Action(tool=tool, params=params))
|
||
assert result.ok, f"{tool} 失败: {result.summary}"
|
||
return result
|
||
|
||
# 1) 导入两个素材
|
||
act("import_footage", paths=["/media/alpha.mp4", "/media/beta.mp4"])
|
||
footage = {it["name"]: it["id"] for it in adapter.list_footage()}
|
||
f_alpha = footage["alpha.mp4"]
|
||
f_beta = footage["beta.mp4"]
|
||
|
||
# 2) 铺轨:alpha[0,4) + beta[4,8)
|
||
act("place_clip", clip_id=f_alpha, track_index=1, time=r(0), in_point=r(0), out_point=r(4))
|
||
act("place_clip", clip_id=f_beta, track_index=1, time=r(4), in_point=r(0), out_point=r(4))
|
||
clip_a = _first_clips(adapter)[0]
|
||
|
||
# 3) 切口帧(编辑前):t=3 处是 alpha 的画面
|
||
frame_before = _frame_at(executor, r(3))
|
||
assert not frame_is_black(frame_before, 64, 36), "编辑前切口帧不得全黑"
|
||
decode_png(frame_before) # 必须是合法 PNG
|
||
|
||
# 4) 在 t=2 处切开 alpha
|
||
act("split_clip", clip_id=clip_a.id, time=r(2))
|
||
right_half = next(c for c in _first_clips(adapter) if c.start == Rational(2))
|
||
|
||
# 5) 波纹删除右半段 → beta 前移到 [2,6)
|
||
act("ripple_delete", clip_ids=[right_half.id])
|
||
frame_after_cut = _frame_at(executor, r(3))
|
||
assert not frame_is_black(frame_after_cut, 64, 36)
|
||
# 编辑前后切口帧像素不同(beta 代替了 alpha 出现在 t=3)
|
||
assert _sha(frame_after_cut) != _sha(frame_before), "切口帧像素应随编辑变化"
|
||
|
||
# 6) 给 beta 加效果并设参数 → 画面再变
|
||
beta_clip = next(c for c in _first_clips(adapter) if c.source == f_beta)
|
||
act("add_effect", clip_id=beta_clip.id, effect_type="brightness")
|
||
node_id = adapter._must_clip(beta_clip.id).effects[0].node_id # 白盒取节点
|
||
act("set_param", effect_id=node_id, param="level", value=0.5)
|
||
frame_after_effect = _frame_at(executor, r(3))
|
||
assert not frame_is_black(frame_after_effect, 64, 36)
|
||
assert _sha(frame_after_effect) != _sha(frame_after_cut), "加效果后画面应改变"
|
||
|
||
# 7) 整段会话撤销 → 时间线恢复原状(回到首个变更批次之前:空)
|
||
restored = executor.undo_session()
|
||
assert restored >= 6, f"应撤销 ≥6 个批次,实际 {restored}"
|
||
overview = adapter.get_project_overview()
|
||
assert overview.timeline_ids == [] # 无时间线 = 铺轨前的原状
|
||
assert adapter.list_footage() == [] # 连导入也被撤销(快照覆盖首个批次)
|