wtd-libreflow-addon/dev/test_scene_manager.py

"""End-to-end test of the Scene Manager against dev/fake_libreflow.py.

    blender -b --factory-startup --python dev/make_test_assets.py -- <workdir>
    python dev/fake_libreflow.py <workdir>          (keep it running, fresh state)
    blender -b --factory-startup --python-exit-code 1 --python dev/test_scene_manager.py -- <workdir>

Prints every check and exits with an error when one fails. Timers do not run in
background mode, so the test runs the background-job poller itself (pump).
"""

import importlib
import json
import os
import sys
import time
import urllib.request

import bpy

WORKDIR = sys.argv[sys.argv.index("--") + 1]
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(REPO))
PACKAGE = os.path.basename(REPO)
addon = importlib.import_module(PACKAGE)
addon.register()
config = sys.modules[PACKAGE + ".config"]
ipc = sys.modules[PACKAGE + ".scene_manager.ipc"]
sm = sys.modules[PACKAGE + ".scene_manager.operators"]
lib = sys.modules[PACKAGE + ".scene_manager.library"]
shot_viewer = sys.modules[PACKAGE + ".shot_viewer"]

SERVER_FILE = os.path.join(WORKDIR, "kabaret", "server.json")
config.LIBREFLOW_SERVER_FILE = SERVER_FILE
SHOT = "/melvile/films/f01/sequences/s010/shots/p010"
BOB_FILE = "/melvile/asset_types/main_character/assets/bob/tasks/rigging/files/rig_blend"
FAILED = []


def check(label, condition, detail=""):
    print(("  ok    " if condition else "  FAIL  ") + label + (f"  ({detail})" if detail else ""))
    if not condition:
        FAILED.append(label)


def pump(timeout=15):
    end = time.time() + timeout
    while ipc._jobs or not ipc._results.empty():
        ipc._poll()
        if time.time() > end:
            check("background jobs finish", False, str(ipc._jobs))
            return
        time.sleep(0.05)


def fake(path, body):
    info = json.load(open(SERVER_FILE))
    request = urllib.request.Request(
        f"http://127.0.0.1:{info['port']}{path}",
        data=json.dumps(body).encode(),
        method="POST",
        headers={"Authorization": "Bearer " + info["token"], "Content-Type": "application/json"},
    )
    return json.loads(urllib.request.urlopen(request).read())["result"]


def loaded():
    found = lib.all_instances()
    return {
        library.name: (library.get("wtd_revision"), [i.name for i in found.get(library.name, [])])
        for library in lib.managed_libraries()
    }


def message():
    return sm.STATE.message[0]


bpy.ops.wm.read_homefile(use_empty=True)
scene = bpy.context.scene

print("Connection")
sm.connect(scene)
pump()
check("connected to the session", sm.connected() and sm.current_session().get("session_name") == "libreflow_dev")
check("user and site known", sm.STATE.ping.get("user") == "nero" and sm.STATE.ping.get("site") == "lyon")

print("Context")
scene.wtd_scene_manager.context_oid = "/melvile/nope"
pump()
check("invalid oid reported", not sm.context_label("/melvile/nope")[1])
scene.wtd_scene_manager.context_oid = SHOT
pump()
check("shot context validated", sm.context_label(SHOT) == ("shot: f01 / s010 / p010", True))
errors = {a["name"]: a["error"] for a in sm.STATE.casting.get(SHOT, {}).get("assets", [])}
check("casting read", set(errors) == {"bob", "tree", "rock", "lamp", "ghost"}, str(errors))

print("Load")
bpy.ops.wtd_scene.load()
pump()
state = loaded()
check("bob: 2 instances of v002", state.get("bob") == ("v002", ["COL_bob", "COL_bob.001"]), str(state.get("bob")))
check("tree: 1 collection instance", state.get("tree") == ("v001", ["tree"]), str(state.get("tree")))
check("rock: single top-level collection fallback", state.get("rock") == ("v001", ["rock_geo"]))
check("lamp and ghost reported", "lamp" in message() and "ghost" in message(), message())
check("type collections", {"COL_main_character", "COL_bg_set", "COL_main_props"} <= set(scene.collection.children.keys()))
check(
    "chars are posable overrides",
    all(ob.override_library for ob in bpy.data.collections["COL_bob"].all_objects),
)
check("shot viewer groups bob as character", "Characters" in shot_viewer.rig_groups(bpy.context.view_layer))
bpy.ops.wtd_scene.load()
pump()
check("load again adds nothing", message().startswith("Nothing new to load"), message())

print("Versions")
check("bob is the latest", sm.version_status(bpy.data.libraries["bob"])[0] == "LATEST")
fake("/_test/publish", {"file_oid": BOB_FILE, "name": "v003", "available": False})
sm.refresh_versions()
pump()
check("v003 not on this site: pending", sm.version_status(bpy.data.libraries["bob"])[0] == "PENDING")
fake(
    "/_test/publish",
    {"file_oid": BOB_FILE, "name": "v004", "available": True, "path": os.path.join(WORKDIR, "lib", "bob", "v001", "rig.blend")},
)
sm.refresh_versions()
pump()
check("v004 available: update", sm.version_status(bpy.data.libraries["bob"])[0] == "UPDATE")
bpy.ops.wtd_scene.update(library="bob")
pump()
check("bob updated to v004", loaded()["bob"][0] == "v004", message())
meshes = {c.name: len([o for o in c.all_objects if o.type == "MESH"]) for c in lib.instances(bpy.data.libraries["bob"])}
check("overrides resynced to v004 (1 mesh)", set(meshes.values()) == {1}, str(meshes))

print("Version picker")
sm.STATE.revisions[BOB_FILE] = ipc.call("wtd_get_revisions", None, file_oid=BOB_FILE)
sm._REVISION_ITEMS[:] = [(r["name"], r["name"], "", "NONE", i) for i, r in enumerate(sm.STATE.revisions[BOB_FILE])]
bpy.ops.wtd_scene.pick_revision("EXEC_DEFAULT", library="bob", revision="v002")
check("rolled back to v002", loaded()["bob"][0] == "v002")
check("unavailable v003 refused", bpy.ops.wtd_scene.pick_revision("EXEC_DEFAULT", library="bob", revision="v003") == {"CANCELLED"})

print("Remove")
bpy.ops.wtd_scene.remove(instance="COL_bob.001")
check("one instance removed", loaded()["bob"][1] == ["COL_bob"])
for ob in bpy.context.view_layer.objects:
    ob.select_set(False)
bpy.data.objects["tree"].select_set(True)
bpy.ops.wtd_scene.remove(selected=True)
check("selected instance removed", loaded()["tree"][1] == [])
bpy.ops.wtd_scene.remove("EXEC_DEFAULT", library="rock")
check("whole asset removed", "rock" not in bpy.data.libraries)
check("its indirect library removed too", set(bpy.data.libraries.keys()) == {"bob", "tree"}, str(bpy.data.libraries.keys()))
bpy.ops.wtd_scene.load()
pump()
check("load restores what is missing", {k: len(v[1]) for k, v in loaded().items()} == {"bob": 2, "rock": 1, "tree": 1})

print("Ambiguous file")
try:
    lib.link_asset(
        {"name": "zzz", "oid": "x", "asset_type": "main_props", "file": {"file_oid": "f"}},
        {"name": "v1", "path": os.path.join(WORKDIR, "lib", "bob", "v001", "rig.blend")},
        SHOT,
    )
    check("ambiguous file refused", False)
except lib.LoadError as exc:
    check("ambiguous file refused", "top-level collections" in str(exc), str(exc))

print("Assetize")
local = bpy.data.collections.new("COL_lamp")
scene.collection.children.link(local)
local.objects.link(bpy.data.objects.new("GEO_lamp", bpy.data.meshes.new("GEO_lamp")))
check("only free local collections offered", [c.name for c in lib.assetize_candidates()] == ["COL_lamp"])
sm.STATE.assets = ipc.call("wtd_list_assets", None, oid=SHOT)
bpy.ops.wtd_scene.assetize(
    "EXEC_DEFAULT", collection="COL_lamp", asset_type="main_props", asset_name="lamp", task="modeling", comment="test"
)
pump()
check("published and replaced", "replaced" in message(), message())
check("lamp linked", loaded().get("lamp") == ("v001", ["COL_lamp"]), str(loaded().get("lamp")))
check("local collection kept, renamed", "COL_lamp_local" in bpy.data.collections)
bad = bpy.data.collections.new("COL_bad")
scene.collection.children.link(bad)
bpy.ops.wtd_scene.assetize("EXEC_DEFAULT", collection="COL_bad", asset_type="main_props", asset_name="bad", task="layout")
pump()
check("unknown task reported", "available: modeling, rigging" in message(), message())

print("Save / reopen")
path = os.path.join(WORKDIR, "scene_manager_test.blend")
bpy.ops.wm.save_as_mainfile(filepath=path)
bpy.ops.wm.open_mainfile(filepath=path)
sm._on_load_post()
check("everything found after reopening", set(loaded()) == {"bob", "lamp", "rock", "tree"})
sm.connect(bpy.context.scene)
pump()
check("versions checked again", sm.version_status(bpy.data.libraries["bob"])[0] == "UPDATE")

print("Several sessions")
fake(
    "/_test/sessions",
    {
        "uid-1": {"session_name": "a", "cluster_name": "melvile", "is_gui": True, "current_oid": SHOT},
        "uid-2": {"session_name": "b", "cluster_name": "melvile", "is_gui": False, "current_oid": "/melvile"},
    },
)
bpy.context.window_manager.wtd_libreflow.session_uid = ""
sm.connect(bpy.context.scene)
pump()
check("asks to pick a session", "pick one" in message(), message())
bpy.ops.wtd_scene.select_session(uid="uid-2")
pump()
check("works once a session is picked", sm.context_label(SHOT)[1])

print("Libreflow not running")
config.LIBREFLOW_SERVER_FILE = os.path.join(WORKDIR, "missing.json")
sm.connect(bpy.context.scene)
pump()
check("reported as not running", not sm.connected() and sm.STATE.connection_error == "Libreflow is not running")

addon.unregister()
print(f"\n{len(FAILED)} check(s) failed: {', '.join(FAILED)}" if FAILED else "\nAll checks passed")
if FAILED:
    raise RuntimeError(f"{len(FAILED)} Scene Manager check(s) failed")