FemtoStudio-comfyui-timeline/tests/test_timeline_model.py

"""Global Timeline Refactor — Phase 1 (model layer) JS-asset pins.

``timeline_model.js`` is the PUBLIC standard API any ComfyUI node or
extension binds to (``window.ComfyTimeline``). These are pure file/text
checks (no JS runner — ``node`` is not on PATH); they pin the API
contract so a refactor that drops a mutator, an event, the public
namespace, or the persistence hook surfaces here instead of silently
breaking every downstream consumer in the browser.

The sampling math itself stays covered by the Python<->JS parity tests
(``test_math_parity.py``); ``Curve.sample`` only delegates to the same
``sampleAt`` those tests exercise.
"""

from pathlib import Path

_REPO = Path(__file__).resolve().parent.parent
_JS_DIR = _REPO / "src" / "timelineanim_nodes" / "web" / "js"
_MODEL = _JS_DIR / "timeline_model.js"


def _model_js() -> str:
    return _MODEL.read_text(encoding="utf-8")


# ---- file + module shape -------------------------------------------


def test_timeline_model_js_exists():
    assert _MODEL.is_file(), (
        f"timeline_model.js missing at {_MODEL}. Phase 1 of the global "
        "timeline refactor introduces the model layer here."
    )


def test_model_exports_the_three_classes():
    js = _model_js()
    for marker in [
        "export class Curve",
        "export class CurveStore",
        "export class TimelineModel",
    ]:
        assert marker in js, (
            f"timeline_model.js missing export: {marker!r} — consumers "
            "import these by name."
        )


def test_model_classes_are_event_driven():
    """CurveStore and TimelineModel must extend EventTarget so any
    extension can subscribe via addEventListener. Pin both bases.
    """
    js = _model_js()
    assert "class CurveStore extends EventTarget" in js, (
        "CurveStore must extend EventTarget to emit 'change' events."
    )
    assert "class TimelineModel extends EventTarget" in js, (
        "TimelineModel must extend EventTarget to emit frame/range/"
        "fps/play events."
    )


def test_model_reuses_sampler_math():
    """Curve.sample must delegate to sampler.js sampleAt — NOT
    re-implement curve math — so the model, the renderer, and the
    Python sampler stay byte-for-byte in agreement.
    """
    js = _model_js()
    assert "from './sampler.js'" in js, (
        "timeline_model.js must import from sampler.js."
    )
    assert "sampleAt" in js, (
        "Curve.sample must call sampleAt from sampler.js rather than "
        "duplicating the Hermite/linear/stepped math."
    )


# ---- public singleton / namespace ----------------------------------


def test_public_namespace_is_comfytimeline():
    """The standard global is ``window.ComfyTimeline``. Pin the
    idempotent guard so a re-import can't replace the model out from
    under live subscribers.
    """
    js = _model_js()
    assert "window.ComfyTimeline" in js, (
        "timeline_model.js must publish the singleton on "
        "window.ComfyTimeline."
    )
    assert "if (window.ComfyTimeline) return" in js, (
        "Bootstrap must be idempotent — reuse the existing singleton "
        "on re-import instead of constructing a second model."
    )


def test_singleton_exposes_model_and_curves():
    js = _model_js()
    # The published object must carry both sub-objects + the classes.
    for marker in ["model,", "curves,", "Curve,", "CurveStore,", "TimelineModel,"]:
        assert marker in js, (
            f"Public singleton object missing field: {marker!r}"
        )


def test_es_module_accessor_present():
    js = _model_js()
    assert "export function getTimeline" in js, (
        "Provide getTimeline() so ES-module consumers can import the "
        "singleton instead of reaching for the window global."
    )


def test_does_not_resurrect_legacy_registry_name():
    """The new standard is ComfyTimeline. The model file must not
    define the legacy window.timelineAnim registry (that lives in
    shared.js during the migration and is removed as nodes migrate).
    """
    js = _model_js()
    assert "window.timelineAnim" not in js, (
        "timeline_model.js should not reference the legacy "
        "window.timelineAnim registry — the new namespace is "
        "window.ComfyTimeline."
    )


# ---- TimelineModel API surface -------------------------------------


def test_model_has_full_mutator_surface():
    """Maya-parity transport + range + fps + frame. Pin every mutator
    so a consumer can rely on the contract.
    """
    js = _model_js()
    for marker in [
        "setCurrentFrame(frame, source)",
        "setAnimRange(start, end, source)",
        "setPlayRange(start, end, source)",
        "setFps(fps, source)",
        "setLoop(loop, source)",
        "play(source)",
        "pause(source)",
        "togglePlay(source)",
    ]:
        assert marker in js, (
            f"TimelineModel missing mutator: {marker!r}"
        )


def test_model_tracks_two_ranges_plus_fps_and_transport():
    """The model fields any view reads. Anim range + play range +
    fps + transport state, Maya-style.
    """
    js = _model_js()
    for field in [
        "this.currentFrame",
        "this.animStart",
        "this.animEnd",
        "this.playStart",
        "this.playEnd",
        "this.fps",
        "this.playing",
        "this.loop",
    ]:
        assert field in js, (
            f"TimelineModel must track {field} as a public field."
        )


def test_model_emits_the_documented_events():
    """Subscribers bind to these event names. Pin the emit sites so a
    rename doesn't silently break every listener.
    """
    js = _model_js()
    for ev in ["'frame'", "'range'", "'fps'", "'play'", "'loop'"]:
        assert ev in js, (
            f"TimelineModel must emit a {ev} event."
        )
    assert "new CustomEvent" in js, (
        "Events must be CustomEvents so detail (frame/range/source) "
        "rides along."
    )


def test_mutators_carry_a_source_token_for_echo_suppression():
    """Every mutator threads an optional `source` into the event detail
    so the originating view can ignore its own echo (no feedback loop).
    Pin the exact emit shapes (not just the word 'source', which appears
    dozens of times) so echo-suppression can't be silently dropped.
    """
    js = _model_js()
    assert "this._emit('frame', { frame: f, source })" in js, (
        "setCurrentFrame must emit frame with the source token."
    )
    assert "this._emit('fps', { fps: n, source })" in js, (
        "setFps must emit fps with the source token."
    )
    assert "source," in js, "Range emit detail must carry source."


def test_model_runs_a_raf_playback_loop():
    """Live playback advances the playhead by elapsed*fps on a
    requestAnimationFrame loop, wrapping at the play range when loop is
    on. Pin the loop primitives.
    """
    js = _model_js()
    assert "requestAnimationFrame" in js, (
        "Playback must drive the playhead on a requestAnimationFrame "
        "loop."
    )
    assert "cancelAnimationFrame" in js, (
        "pause() must cancel the rAF loop."
    )
    assert "this.fps" in js and "_playFrame" in js, (
        "The loop must advance a fractional playhead by elapsed*fps."
    )


# ---- CurveStore API surface ----------------------------------------


def test_curvestore_registry_surface():
    js = _model_js()
    for marker in [
        "register(curveId, curve)",
        "unregister(curveId)",
        "get(curveId)",
        "touch(curveId)",
    ]:
        assert marker in js, (
            f"CurveStore missing registry method: {marker!r}"
        )
    assert "new CustomEvent('change'" in js, (
        "CurveStore must emit a 'change' CustomEvent with the curveId "
        "so the bar knows which ticks to repaint."
    )


# ---- persistence: app.graph.extra round-trip -----------------------


def test_persistence_round_trips_via_graph_extra():
    """State rides the workflow JSON via app.graph.extra.timeline:
    synchronous small-state write on change, rehydrate on graph load.
    """
    js = _model_js()
    assert "serialize()" in js, "Model needs serialize() for persistence."
    assert "load(data)" in js, "Model needs load(data) to rehydrate."
    assert "app.graph" in js and "extra" in js and "timeline" in js, (
        "Persistence must read/write app.graph.extra.timeline."
    )


def test_persistence_rehydrates_on_after_configure_graph():
    """Rehydrate hook must be afterConfigureGraph (verified against the
    current frontend extension lifecycle).
    """
    js = _model_js()
    assert "afterConfigureGraph" in js, (
        "Rehydrate must hook the afterConfigureGraph extension "
        "lifecycle so a loaded workflow restores its timeline state."
    )
    assert "registerExtension" in js, (
        "Persistence must register an extension to receive "
        "afterConfigureGraph."
    )


def test_persistence_write_is_synchronous():
    """An immediate workflow save must see the latest timeline state."""
    js = _model_js()
    assert "model.addEventListener(ev, persist)" in js
    assert "g.extra.timeline = model.serialize()" in js
    persistence = js[js.find("function _wirePersistence"):]
    assert "setTimeout" not in persistence, (
        "Timeline persistence must not lag behind an immediate workflow save."
    )


def test_load_does_not_immediately_rewrite():
    """Rehydrate tags its events source:'load' and the persistence
    scheduler skips that source — otherwise loading a workflow would
    instantly mark it dirty by writing identical data back.
    """
    js = _model_js()
    assert "'load'" in js, (
        "load() must tag its emitted events with source 'load'."
    )
    assert "source === 'load'" in js, (
        "The persistence scheduler must skip events whose source is "
        "'load' to avoid a rehydrate->rewrite loop."
    )


# ---- studio-grade hardening (audit follow-ups) ---------------------


def test_api_version_separate_from_schema_version():
    """apiVersion (semver of the public surface) must be distinct from
    schemaVersion (the app.graph.extra serialization integer), so
    consumers can feature-detect without false breakage on persistence
    shape changes.
    """
    js = _model_js()
    assert "apiVersion" in js and "API_VERSION" in js
    assert "schemaVersion" in js and "SCHEMA_VERSION" in js


def test_playback_loop_clamps_dt():
    """rAF pauses in background tabs → huge dt on resume. The loop must
    clamp dt so the playhead doesn't teleport. Pin the EXACT formula so a
    regression (dropped clamp, wrong bound) is caught — the math itself is
    untestable here without a JS runtime, so the formula pin is the lock.
    """
    js = _model_js()
    assert "MAX_TICK_DT" in js, "Playback must cap dt (MAX_TICK_DT)."
    assert "Math.min(Math.max(0, (ts - this._lastTs) / 1000), MAX_TICK_DT)" in js, (
        "dt must be clamped to [0, MAX_TICK_DT]."
    )


def test_playback_loop_wrap_formula():
    """Loop wrap must preserve the fractional overshoot across the
    boundary (constant playback speed through the loop point)."""
    js = _model_js()
    assert "this._playFrame = lo + ((this._playFrame - lo) % span)" in js, (
        "Loop wrap formula changed — playback speed will jitter at the "
        "loop boundary if this drifts."
    )


def test_step_wraps_within_play_range():
    """step(±N) must wrap within the play range, handling negative delta
    (the +span)%span idiom)."""
    js = _model_js()
    assert "((f - this.playStart) % span + span) % span" in js, (
        "step() wrap formula changed — negative steps won't wrap correctly."
    )


def test_playback_reschedules_before_emitting():
    """A throwing subscriber must not freeze transport. The tick must
    reschedule the next rAF BEFORE emitting the frame event.
    """
    js = _model_js()
    tick = js.find("_tick(ts)")
    assert tick > 0
    body = js[tick:tick + 1800]
    sched = body.find("requestAnimationFrame(this._tick)")
    emit = body.find("setCurrentFrame(")
    assert sched > 0 and emit > 0 and sched < emit, (
        "_tick must schedule the next frame before the first "
        "setCurrentFrame emit so a throwing listener can't stall it."
    )


def test_playback_guards_inverted_range():
    js = _model_js()
    assert "span <= 0" in js, (
        "_tick must guard span<=0 (inverted/zero range) and pause "
        "rather than spin."
    )


def test_model_has_derived_helpers():
    js = _model_js()
    for marker in [
        "get duration", "isFrameInPlayRange", "clampToPlayRange",
        "frameAtTime", "timeAtFrame", "goToStart", "goToEnd",
        "snapshot()", "reset(",
    ]:
        assert marker in js, f"Model missing derived/helper: {marker!r}"


def test_model_has_subscribe_convenience():
    js = _model_js()
    assert "subscribe(events, cb)" in js, (
        "Model must offer subscribe(events, cb) -> unsubscribe."
    )


def test_model_has_playback_speed():
    js = _model_js()
    assert "playbackSpeed" in js and "setPlaybackSpeed" in js


def test_model_has_every_frame_playback_mode():
    """Maya 'play every frame' — advance exactly one frame per tick
    instead of elapsed*fps. Pin the field, setter, tick branch, and that
    it's persisted."""
    js = _model_js()
    assert "playbackMode" in js and "setPlaybackMode" in js
    assert "'everyframe'" in js
    assert "this.playbackMode === 'everyframe'" in js, (
        "tick must branch on every-frame mode."
    )
    assert "playbackMode: this.playbackMode" in js, "playbackMode must serialize."


def test_scrub_rebases_play_frame_except_during_playback():
    """An external scrub must take effect mid-playback (rebase the
    fractional playhead), while the loop's own 'playback' updates keep
    sub-frame accumulation."""
    js = _model_js()
    assert "source !== 'playback'" in js, (
        "setCurrentFrame must rebase _playFrame unless source is "
        "'playback'."
    )


def test_curvestore_has_merged_frames_and_prune():
    js = _model_js()
    assert "mergedFrames(" in js, "CurveStore must expose mergedFrames()."
    assert "prune(" in js, "CurveStore must expose prune(validIds)."


def test_curve_dedupes_by_frame():
    js = _model_js()
    # normalize must collapse duplicate rounded frames via a Map keyed on
    # the rounded frame (last wins). Pin the specific construct so the
    # dedupe can't silently revert to a plain sort.
    assert "byFrame" in js and "byFrame.set(Math.round(out.frame)" in js, (
        "Curve.normalize must dedupe keys into a Map keyed by rounded "
        "frame (last-wins)."
    )


def test_load_clamps_current_frame_into_range():
    """A stale/hostile saved currentFrame must be clamped into the anim
    range by load(), not applied verbatim."""
    js = _model_js()
    assert "clampToAnimRange(Math.round(num(data.currentFrame" in js, (
        "load() must clamp the loaded playhead via clampToAnimRange."
    )


def test_serialize_omits_playing():
    """serialize() must NOT include `playing` — else a saved workflow
    would auto-play on load. snapshot() includes it; serialize() omits."""
    js = _model_js()
    start = js.find("serialize() {")
    assert start > 0, "serialize() not found"
    body = js[start:start + 500]
    assert "playing" not in body, (
        "serialize() must omit `playing` (no auto-play on workflow load)."
    )
    snap = js.find("snapshot() {")
    assert snap > 0 and "playing" in js[snap:snap + 400], (
        "snapshot() SHOULD include playing (it's the live read-side state)."
    )


def test_named_timeline_registry_exists():
    js = _model_js()
    assert "class TimelineRegistry" in js
    assert "getOrCreate(name)" in js
    assert "timelines," in js, "Singleton must expose the timelines registry."


def test_singleton_exposes_api_and_ready():
    js = _model_js()
    assert "apiVersion:" in js and "schemaVersion:" in js
    assert "ready" in js, "Singleton must expose a ready Promise."


def test_rehydrate_resets_when_graph_has_no_timeline_data():
    """The model is a singleton that outlives graphs. afterConfigureGraph
    must reset() to defaults when the loaded graph carries NO
    extra.timeline (File→New, or a workflow authored without the pack) —
    otherwise the previous workflow's frame/range leaks into new
    consumer nodes and bakes into their queued prompts."""
    js = _model_js()
    idx = js.find("afterConfigureGraph()")
    assert idx > 0, "persistence extension must hook afterConfigureGraph"
    body = js[idx:idx + 900]
    assert "model.load(data)" in body
    assert "model.reset('load')" in body, (
        "afterConfigureGraph must model.reset('load') when the graph has "
        "no saved timeline — stale singleton state must not leak across "
        "workflows."
    )


def test_load_emits_mode_for_mode_only_change():
    """load() must emit 'mode' when playbackMode changed — the range/fps
    emits only fire when THOSE axes changed, so a mode-only difference
    would otherwise leave the bar's EF button stale. (Behaviorally
    covered in tests/js/model.test.mjs; this pins the source.)"""
    js = _model_js()
    idx = js.find("load(data) {")
    assert idx > 0
    body = js[idx:idx + 2600]
    assert "_emit('mode'" in body, (
        "load() must emit 'mode' on a playbackMode change."
    )