Added dev test tools and CLAUDE.md
dev/: test asset files, a fake Libreflow IPC server and an end-to-end Scene Manager test (excluded from the extension build). CLAUDE.md: notes for working on the add-on (conventions, testing, design findings, open points). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Nero Frame 13andClaude Opus 5.5 committed
1 parent
c238f4db7c
commit
017bdfc2fd
4 files changed
+513
No files matched your search
@@ -0,0 +1,220 @@
|
||||
"""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/chars/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_chars", "COL_sets", "COL_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)
|
||||
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": "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="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="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")
|
||||
Reference in new issue
Block a user