Stop tracking dev/ (ignored, kept locally)
The fake Libreflow server, test assets script and end-to-end test stay
on disk; dev/ is in .gitignore since 3f87904.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
NDemaerschalkandClaude Opus 5.5 committed
1 parent
ce30491cb8
commit
4161ec4ad5
3 files changed
-588
No files matched your search
@@ -1,196 +0,0 @@
|
||||
"""Fake Libreflow IPC server (kabaret.ipc + wtdcmds commands) to test the Scene Manager.
|
||||
|
||||
python dev/fake_libreflow.py <workdir>
|
||||
|
||||
Serves a project with one shot casting bob (main_character, x2), tree (bg_set), rock (main_props),
|
||||
lamp (main_props, nothing published), car (vehicles, v001 not on this site) and ghost (not in the flow);
|
||||
chair (set_props) is not cast, for Load Publish. The asset files come
|
||||
from dev/make_test_assets.py (<workdir>/lib). Writes <workdir>/kabaret/server.json:
|
||||
point config LIBREFLOW_SERVER_FILE to it (see dev/test_scene_manager.py).
|
||||
|
||||
Test-only endpoints: POST /_test/publish {file_oid, name, available, path},
|
||||
POST /_test/download {revision_oid} (a requested revision reaches this site),
|
||||
POST /_test/sessions {uid: session}, POST /_test/calls (commands received).
|
||||
State lives in memory: restart the server for a clean state.
|
||||
"""
|
||||
import json, os, shutil, sys, time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
|
||||
ROOT = sys.argv[1]
|
||||
TOKEN = "secret-token"
|
||||
PROJECT = "/melvile"
|
||||
SHOT = "/melvile/films/f01/sequences/s010/shots/p010"
|
||||
SESSIONS = {"uid-1": {"session_name": "libreflow_dev", "cluster_name": "melvile", "is_gui": True, "current_oid": SHOT}}
|
||||
|
||||
def rev(name, path, available=True, comment=""):
|
||||
return {"name": name, "oid": f"x/history/revisions/{name}", "path": path, "sync_status": "Available" if available else "NotAvailable",
|
||||
"available": available, "date": time.time(), "user": "nero", "comment": comment}
|
||||
|
||||
ASSETS = { # asset oid -> (type, {file_oid: (label, [revisions oldest first])})
|
||||
f"{PROJECT}/asset_types/main_character/assets/bob": ("main_character", {f"{PROJECT}/asset_types/main_character/assets/bob/tasks/rigging/files/rig_blend": ("rigging/rig.blend", [
|
||||
rev("v001", f"{ROOT}/lib/bob/v001/rig.blend"), rev("v002", f"{ROOT}/lib/bob/v002/rig.blend", comment="two meshes")])}),
|
||||
f"{PROJECT}/asset_types/bg_set/assets/tree": ("bg_set", {f"{PROJECT}/asset_types/bg_set/assets/tree/tasks/modeling/files/model_blend": ("modeling/model.blend", [
|
||||
rev("v001", f"{ROOT}/lib/tree/v001/model.blend")])}),
|
||||
f"{PROJECT}/asset_types/main_props/assets/rock": ("main_props", {f"{PROJECT}/asset_types/main_props/assets/rock/tasks/modeling/files/rock_blend": ("modeling/rock.blend", [
|
||||
rev("v001", f"{ROOT}/lib/rock/v001/rock.blend")])}),
|
||||
f"{PROJECT}/asset_types/main_props/assets/lamp": ("main_props", {}),
|
||||
f"{PROJECT}/asset_types/vehicles/assets/car": ("vehicles", {f"{PROJECT}/asset_types/vehicles/assets/car/tasks/rigging/files/car_blend": ("rigging/car.blend", [
|
||||
rev("v001", f"{ROOT}/lib/car/v001/car.blend", available=False)])}),
|
||||
f"{PROJECT}/asset_types/set_props/assets/chair": ("set_props", {
|
||||
f"{PROJECT}/asset_types/set_props/assets/chair/tasks/shading/files/chair_shading_blend": ("shading/chair_shading.blend", [
|
||||
rev("v001", f"{ROOT}/lib/chair/v001/chair.blend")]),
|
||||
f"{PROJECT}/asset_types/set_props/assets/chair/tasks/modeling/files/chair_blend": ("modeling/chair.blend", [
|
||||
rev("v001", f"{ROOT}/lib/chair/v001/chair.blend"), rev("v002", f"{ROOT}/lib/chair/v001/chair.blend", available=False)])}),
|
||||
}
|
||||
CASTING = [("bob", "main_character", 2), ("tree", "bg_set", 1), ("rock", "main_props", 1), ("lamp", "main_props", 1), ("car", "vehicles", 1), ("ghost", "main_character", 1)]
|
||||
CALLS = []
|
||||
|
||||
def set_oids():
|
||||
for _a, (_t, files) in ASSETS.items():
|
||||
for f, (_l, revs) in files.items():
|
||||
for r in revs: r["oid"] = f"{f}/history/revisions/{r['name']}"
|
||||
|
||||
def find_rev(oid):
|
||||
return next((r for _a, (_t, files) in ASSETS.items() for _l, revs in files.values() for r in revs if r["oid"] == oid), None)
|
||||
|
||||
set_oids()
|
||||
|
||||
def file_data(file_oid):
|
||||
for asset_oid, (atype, files) in ASSETS.items():
|
||||
if file_oid in files:
|
||||
label, revs = files[file_oid]
|
||||
avail = [r for r in revs if r["available"]]
|
||||
return {"label": label, "file_oid": file_oid, "latest": revs[-1] if revs else None, "latest_available": avail[-1] if avail else None}
|
||||
raise ValueError(f"No file {file_oid}")
|
||||
|
||||
def context(oid):
|
||||
parts = oid.strip("/").split("/")
|
||||
if oid.startswith(PROJECT + "/films"):
|
||||
return {"oid": oid, "kind": "shot", "project": "melvile", "film": parts[2], "film_oid": "/".join([""] + parts[:3]),
|
||||
"sequence": parts[4], "sequence_oid": "/".join([""] + parts[:5]), "shot": parts[6], "shot_oid": SHOT}
|
||||
if "/assets/" in oid:
|
||||
return {"oid": oid, "kind": "asset", "project": "melvile", "asset_type": parts[2], "asset": parts[4]}
|
||||
if oid == PROJECT:
|
||||
return {"oid": oid, "kind": "project", "project": "melvile"}
|
||||
raise ValueError(f"Oid {oid} does not exist")
|
||||
|
||||
def cmd(name, kw):
|
||||
import fnmatch
|
||||
if name == "wtd_ping":
|
||||
return {"version": "0.1.0", **({"project": "melvile", "user": "nero", "site": "lyon", "root": ROOT} if kw.get("oid") else {})}
|
||||
if name == "wtd_get_context":
|
||||
return context(kw["oid"])
|
||||
if name == "wtd_list_shots":
|
||||
return [context(SHOT), context(SHOT.replace("p010", "p020"))]
|
||||
if name == "wtd_list_assets":
|
||||
return [context(oid) for oid in ASSETS]
|
||||
if name == "wtd_get_casting_publishes":
|
||||
if not kw["oid"].startswith(SHOT):
|
||||
raise ValueError("Not a shot")
|
||||
patterns = kw.get("file_patterns") or ["*/*.blend"]
|
||||
out = []
|
||||
for aname, atype, nb in CASTING:
|
||||
oid = next((o for o in ASSETS if o.endswith("/assets/" + aname)), None)
|
||||
a = {"name": aname, "kitsu_id": "k-" + aname, "asset_type": atype, "nb_occurences": nb, "label": "animate", "oid": oid, "candidates": [oid] if oid else [], "file": None, "error": None}
|
||||
if oid is None:
|
||||
a["error"] = "Asset not found in the flow"
|
||||
else:
|
||||
files = ASSETS[oid][1]
|
||||
match = next((f for p in patterns for f, (label, _r) in files.items() if fnmatch.fnmatchcase(label, p)), None)
|
||||
if match is None:
|
||||
a["error"] = "No file matching " + ", ".join(patterns)
|
||||
else:
|
||||
a["file"] = file_data(match)
|
||||
if a["file"]["latest_available"] is None:
|
||||
a["error"] = "No published revision available on this site"
|
||||
out.append(a)
|
||||
return {"shot_oid": SHOT, "assets": out}
|
||||
if name == "wtd_get_asset_files":
|
||||
return [file_data(f) for f in ASSETS[kw["oid"]][1]]
|
||||
if name == "wtd_get_latest_revisions":
|
||||
res = {}
|
||||
for f in kw["file_oids"]:
|
||||
try: res[f] = file_data(f)
|
||||
except ValueError as exc: res[f] = {"file_oid": f, "error": str(exc)}
|
||||
return res
|
||||
if name == "wtd_get_revisions":
|
||||
for _a, (_t, files) in ASSETS.items():
|
||||
if kw["file_oid"] in files:
|
||||
return list(reversed(files[kw["file_oid"]][1]))
|
||||
raise ValueError("No file")
|
||||
if name == "wtd_request_revisions":
|
||||
res = {}
|
||||
for oid in kw["revision_oids"]:
|
||||
r = find_rev(oid)
|
||||
if r is None:
|
||||
res[oid] = {"oid": oid, "error": f"Could not find {oid}"}; continue
|
||||
if not r["available"]: r["sync_status"] = "Requested"
|
||||
res[oid] = r
|
||||
return res
|
||||
if name == "wtd_assetize":
|
||||
src = kw["source_path"]
|
||||
if not os.path.isfile(src): raise ValueError(f"Source file {src} does not exist")
|
||||
atype, aname, task = kw["asset_type"], kw["asset_name"], kw["task"]
|
||||
asset_oid = f"{PROJECT}/asset_types/{atype}/assets/{aname}"
|
||||
created = asset_oid not in ASSETS
|
||||
if created and not kw.get("create_asset"):
|
||||
raise ValueError(f"Asset {aname} does not exist (pass create_asset=true to create it)")
|
||||
if task not in ("modeling", "rigging"):
|
||||
raise ValueError(f"Task {task} does not exist on {asset_oid} (available: modeling, rigging)")
|
||||
fname = kw.get("file_name") or aname
|
||||
ASSETS.setdefault(asset_oid, (atype, {}))
|
||||
files = ASSETS[asset_oid][1]
|
||||
file_oid = f"{asset_oid}/tasks/{task}/files/{fname}_blend"
|
||||
label, revs = files.setdefault(file_oid, (f"{task}/{fname}.blend", []))
|
||||
name = f"v{len(revs) + 1:03d}"
|
||||
dest = f"{ROOT}/lib/{aname}/{name}/{fname}.blend"; os.makedirs(os.path.dirname(dest), exist_ok=True); shutil.copy(src, dest)
|
||||
revs.append(rev(name, dest, comment=kw.get("comment", ""))); set_oids()
|
||||
return {"asset_oid": asset_oid, "asset_created": created, **file_data(file_oid)}
|
||||
raise KeyError(name)
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def log_message(self, *a): pass
|
||||
def reply(self, code, body):
|
||||
data = json.dumps(body).encode(); self.send_response(code); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(data))); self.end_headers(); self.wfile.write(data)
|
||||
def auth(self):
|
||||
if self.headers.get("Authorization") != f"Bearer {TOKEN}":
|
||||
self.reply(401, {"detail": "Invalid token"}); return False
|
||||
return True
|
||||
def do_GET(self):
|
||||
if not self.auth(): return
|
||||
if self.path == "/sessions": return self.reply(200, SESSIONS)
|
||||
self.reply(404, {"detail": "Not found"})
|
||||
def do_POST(self):
|
||||
if not self.auth(): return
|
||||
url = urlparse(self.path); qs = parse_qs(url.query)
|
||||
body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}")
|
||||
if url.path == "/_test/publish": # simulate a new publish
|
||||
f = body["file_oid"]
|
||||
for _a, (_t, files) in ASSETS.items():
|
||||
if f in files: files[f][1].append(rev(body["name"], body.get("path", "/nowhere/" + body["name"]), body.get("available", True)))
|
||||
set_oids()
|
||||
return self.reply(200, {"result": True})
|
||||
if url.path == "/_test/download":
|
||||
r = find_rev(body["revision_oid"]); r["available"] = True; r["sync_status"] = "Available"
|
||||
return self.reply(200, {"result": True})
|
||||
if url.path == "/_test/sessions":
|
||||
SESSIONS.clear(); SESSIONS.update(body); return self.reply(200, {"result": True})
|
||||
if url.path == "/_test/calls":
|
||||
return self.reply(200, {"result": CALLS})
|
||||
name = url.path.rsplit("/", 1)[-1]
|
||||
uid = qs.get("session_uid", [None])[0]
|
||||
if uid is None and len(SESSIONS) > 1: return self.reply(400, {"detail": "Several sessions connected: session_uid required"})
|
||||
if uid is not None and uid not in SESSIONS: return self.reply(404, {"detail": f"Session {uid} not found"})
|
||||
CALLS.append([name, body.get("kwargs", {})])
|
||||
try:
|
||||
return self.reply(200, {"result": cmd(name, body.get("kwargs", {}))})
|
||||
except KeyError as exc: return self.reply(404, {"detail": f"Unknown command {exc}"})
|
||||
except Exception as exc: return self.reply(500, {"detail": str(exc)})
|
||||
|
||||
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||||
os.makedirs(f"{ROOT}/kabaret", exist_ok=True)
|
||||
with open(f"{ROOT}/kabaret/server.json", "w") as fh:
|
||||
json.dump({"host": "127.0.0.1", "port": server.server_address[1], "token": TOKEN, "pid": os.getpid()}, fh)
|
||||
print("listening", server.server_address[1], flush=True)
|
||||
server.serve_forever()
|
||||
@@ -1,72 +0,0 @@
|
||||
"""Write the asset files used by fake_libreflow.py into <workdir>/lib.
|
||||
|
||||
blender -b --factory-startup --python dev/make_test_assets.py -- <workdir>
|
||||
|
||||
bob: a rig (COL_bob: armature + meshes), v001 has 1 mesh, v002 has 2.
|
||||
tree: COL_tree holding only LOD collections (COL_tree_LOD0, COL_tree_LOD1), plus an
|
||||
unrelated top-level collection. v002 adds a mesh outside the LODs, v003 has no LOD.
|
||||
car: a rig, COL_car: armature + named LODs COL_car_LODlow, _LODhigh, _LODmid.
|
||||
chair: COL_chair (not cast, loaded with Load Publish).
|
||||
rock: a single top-level collection not named after the asset (fallback case), its mesh
|
||||
using a node group linked from Blender's essentials (an indirect library).
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
import bpy
|
||||
|
||||
WORKDIR = sys.argv[sys.argv.index("--") + 1]
|
||||
|
||||
|
||||
def mesh_object(name):
|
||||
mesh = bpy.data.meshes.new(name)
|
||||
mesh.from_pydata([(0, 0, 0), (1, 0, 0), (0, 1, 0)], [], [(0, 1, 2)])
|
||||
return bpy.data.objects.new(name, mesh)
|
||||
|
||||
|
||||
def write(path, collection_name, meshes=1, rig=False, extra=False, node_group=False, lods=()):
|
||||
bpy.ops.wm.read_homefile(use_empty=True)
|
||||
scene = bpy.context.scene
|
||||
collection = bpy.data.collections.new(collection_name)
|
||||
scene.collection.children.link(collection)
|
||||
parent = None
|
||||
if rig:
|
||||
parent = bpy.data.objects.new("RIG_" + collection_name, bpy.data.armatures.new("RIG_" + collection_name))
|
||||
collection.objects.link(parent)
|
||||
bpy.context.view_layer.objects.active = parent
|
||||
bpy.ops.object.mode_set(mode="EDIT")
|
||||
bone = parent.data.edit_bones.new("root")
|
||||
bone.tail = (0, 0, 1)
|
||||
bpy.ops.object.mode_set(mode="OBJECT")
|
||||
for index in range(meshes):
|
||||
ob = mesh_object(f"GEO_{collection_name}_{index}")
|
||||
ob.parent = parent
|
||||
collection.objects.link(ob)
|
||||
for lod in lods:
|
||||
lod_collection = bpy.data.collections.new(f"{collection_name}_LOD{lod}")
|
||||
collection.children.link(lod_collection)
|
||||
ob = mesh_object(f"GEO_{collection_name}_LOD{lod}")
|
||||
ob.parent = parent
|
||||
lod_collection.objects.link(ob)
|
||||
if node_group:
|
||||
essentials = os.path.join(bpy.utils.system_resource("DATAFILES"), "assets", "nodes", "geometry_nodes_essentials.blend")
|
||||
with bpy.data.libraries.load(essentials, link=True) as (_data_from, data_to):
|
||||
data_to.node_groups = ["Smooth by Angle"]
|
||||
ob.modifiers.new("Smooth", "NODES").node_group = data_to.node_groups[0]
|
||||
if extra:
|
||||
scene.collection.children.link(bpy.data.collections.new("WIP_other"))
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
bpy.ops.wm.save_as_mainfile(filepath=path)
|
||||
|
||||
|
||||
lib = os.path.join(WORKDIR, "lib")
|
||||
write(os.path.join(lib, "bob", "v001", "rig.blend"), "COL_bob", meshes=1, rig=True, extra=True)
|
||||
write(os.path.join(lib, "bob", "v002", "rig.blend"), "COL_bob", meshes=2, rig=True, extra=True)
|
||||
write(os.path.join(lib, "tree", "v001", "model.blend"), "COL_tree", meshes=0, extra=True, lods=["0", "1"])
|
||||
write(os.path.join(lib, "tree", "v002", "model.blend"), "COL_tree", meshes=1, extra=True, lods=["0", "1"])
|
||||
write(os.path.join(lib, "tree", "v003", "model.blend"), "COL_tree", meshes=1, extra=True)
|
||||
write(os.path.join(lib, "rock", "v001", "rock.blend"), "rock_geo", node_group=True)
|
||||
write(os.path.join(lib, "car", "v001", "car.blend"), "COL_car", meshes=0, rig=True, lods=["low", "high", "mid"])
|
||||
write(os.path.join(lib, "chair", "v001", "chair.blend"), "COL_chair")
|
||||
print("Test assets written in", lib)
|
||||
@@ -1,320 +0,0 @@
|
||||
"""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"
|
||||
CAR = "/melvile/asset_types/vehicles/assets/car"
|
||||
CAR_FILE = CAR + "/tasks/rigging/files/car_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", "car", "ghost"}, str(errors))
|
||||
|
||||
print("Load")
|
||||
bpy.ops.wtd_scene.load()
|
||||
pump()
|
||||
state = loaded()
|
||||
check("bob: 2 numbered instances of v002", state.get("bob") == ("v002", ["COL_bob_01", "COL_bob_02"]), str(state.get("bob")))
|
||||
check("tree: 1 numbered collection instance", state.get("tree") == ("v001", ["tree_01"]), str(state.get("tree")))
|
||||
check("rock: single top-level collection fallback", state.get("rock") == ("v001", ["rock_geo_01"]))
|
||||
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_01"].all_objects),
|
||||
)
|
||||
check("shot manager 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("Request")
|
||||
check("car can be requested", sm.missing_files(scene) == [CAR_FILE], str(sm.missing_files(scene)))
|
||||
bpy.ops.wtd_scene.request()
|
||||
pump()
|
||||
check("car v001 requested", message() == "Requested car v001", message())
|
||||
check("nothing left to request", sm.missing_files(scene) == [])
|
||||
calls = fake("/_test/calls", {})
|
||||
check("request sent to Libreflow", ["wtd_request_revisions", {"revision_oids": [CAR_FILE + "/history/revisions/v001"]}] in calls)
|
||||
fake("/_test/download", {"revision_oid": CAR_FILE + "/history/revisions/v001"})
|
||||
bpy.ops.wtd_scene.refresh()
|
||||
pump()
|
||||
bpy.ops.wtd_scene.load(asset_oid=CAR)
|
||||
pump()
|
||||
check("car loaded once downloaded", loaded().get("car", (None,))[0] == "v001", message())
|
||||
|
||||
print("LOD")
|
||||
tree_lib, car_lib = bpy.data.libraries["tree"], bpy.data.libraries["car"]
|
||||
tree, car = lib.instances(tree_lib)[0], lib.instances(car_lib)[0]
|
||||
check("numbered LODs: instance shows the first", lib.current_lod(tree_lib, tree) == "0" and tree.instance_collection.name == "COL_tree_LOD0")
|
||||
check("named LODs in config order", list(lib.lods(car_lib, car)) == ["high", "mid", "low"], str(list(lib.lods(car_lib, car))))
|
||||
check("named LODs: override shows the first in config order", lib.current_lod(car_lib, car) == "high")
|
||||
hidden = {c.name: (c.hide_viewport, c.hide_render) for c in lib.lods(car_lib, car).values()}
|
||||
check(
|
||||
"other LODs hidden (viewport and render)",
|
||||
hidden == {"COL_car_LODhigh": (False, False), "COL_car_LODmid": (True, True), "COL_car_LODlow": (True, True)},
|
||||
str(hidden),
|
||||
)
|
||||
evaluated = {ob.name for ob in bpy.context.evaluated_depsgraph_get().objects}
|
||||
check("hidden LODs not evaluated", "GEO_COL_car_LODhigh" in evaluated and not evaluated & {"GEO_COL_car_LODmid", "GEO_COL_car_LODlow"})
|
||||
bpy.ops.wtd_scene.set_lod(lod="low", library="car", instance=car.name)
|
||||
check("override switched to low", lib.current_lod(car_lib, car) == "low")
|
||||
for ob in bpy.context.view_layer.objects:
|
||||
ob.select_set(ob in (tree, bpy.data.objects["RIG_COL_car"]))
|
||||
check("LOD of selected", bpy.ops.wtd_scene.set_lod(lod="1", selected=True) == {"FINISHED"})
|
||||
check("only the selected instances having it switched", (lib.current_lod(tree_lib, tree), lib.current_lod(car_lib, car)) == ("1", "low"))
|
||||
check("missing LOD refused", bpy.ops.wtd_scene.set_lod(lod="5", library="tree", instance="tree_01") == {"CANCELLED"})
|
||||
check("LOD order: list, numbers, names", sorted(["b", "10", "low", "2", "a", "high"], key=lib.lod_order) == ["high", "low", "2", "10", "a", "b"])
|
||||
second = lib.add_instance(scene, bpy.context.view_layer, car_lib)
|
||||
check(
|
||||
"second override: LODs found by their linked names",
|
||||
list(lib.lods(car_lib, second)) == ["high", "mid", "low"] and lib.current_lod(car_lib, second) == "high",
|
||||
str([c.name for c in second.children]),
|
||||
)
|
||||
lib.remove_instance(second)
|
||||
lib.update("tree", {"name": "v002", "path": os.path.join(WORKDIR, "lib", "tree", "v002", "model.blend")})
|
||||
tree = lib.instances(bpy.data.libraries["tree"])[0]
|
||||
check("LOD kept by an update", lib.current_lod(bpy.data.libraries["tree"], tree) == "1")
|
||||
check("content outside the LODs: instance refused", bpy.ops.wtd_scene.set_lod(lod="0", library="tree", instance="tree_01") == {"CANCELLED"})
|
||||
lib.update("tree", {"name": "v003", "path": os.path.join(WORKDIR, "lib", "tree", "v003", "model.blend")})
|
||||
tree = lib.instances(bpy.data.libraries["tree"])[0]
|
||||
check("LODs gone: whole collection instanced", tree.instance_collection.name == "COL_tree" and not lib.lods(bpy.data.libraries["tree"], tree))
|
||||
lib.update("tree", {"name": "v001", "path": os.path.join(WORKDIR, "lib", "tree", "v001", "model.blend")})
|
||||
tree = lib.instances(bpy.data.libraries["tree"])[0]
|
||||
check("LODs back: default LOD", lib.current_lod(bpy.data.libraries["tree"], tree) == "0")
|
||||
|
||||
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")
|
||||
bpy.ops.wtd_scene.request(file_oid=BOB_FILE)
|
||||
pump()
|
||||
status = sm.version_status(bpy.data.libraries["bob"])
|
||||
check("v003 requested", status == ("PENDING", "v003 requested, waiting for the download"), str(status))
|
||||
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"})
|
||||
fake("/_test/publish", {"file_oid": BOB_FILE, "name": "v005", "available": False})
|
||||
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="v005")
|
||||
pump()
|
||||
check("picking a version not on this site requests it", message() == "Requested bob v005", message())
|
||||
check("still on v002", loaded()["bob"][0] == "v002")
|
||||
|
||||
print("Remove")
|
||||
bpy.ops.wtd_scene.remove(instance="COL_bob_02")
|
||||
check("one instance removed", loaded()["bob"][1] == ["COL_bob_01"])
|
||||
for ob in bpy.context.view_layer.objects:
|
||||
ob.select_set(False)
|
||||
bpy.data.objects["tree_01"].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", "car", "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, "car": 1, "rock": 1, "tree": 1})
|
||||
check("numbers go on after the highest one", loaded()["bob"][1] == ["COL_bob_01", "COL_bob_02"], str(loaded()["bob"][1]))
|
||||
|
||||
print("Load publish")
|
||||
CHAIR = "/melvile/asset_types/set_props/assets/chair"
|
||||
check("project of the context", sm.project_oid(scene) == "/melvile")
|
||||
sm.STATE.assets = ipc.call("wtd_list_assets", None, oid=sm.project_oid(scene))
|
||||
check("any asset listed, cast or not", CHAIR in [a["oid"] for a in sm.STATE.assets])
|
||||
sm.browse_asset(next(a for a in sm.STATE.assets if a["oid"] == CHAIR))
|
||||
chair_file = sm.default_file("set_props", sm.STATE.browse["files"])
|
||||
check("default file follows the file patterns", chair_file["label"] == "modeling/chair.blend", chair_file["label"])
|
||||
check("versions newest first", [r["name"] for r in chair_file["revisions"]] == ["v002", "v001"])
|
||||
check("default version: latest on this site", sm.default_revision(chair_file)["name"] == "v001")
|
||||
result = bpy.ops.wtd_scene.load_publish("EXEC_DEFAULT", file=chair_file["file_oid"], revision="v002")
|
||||
pump()
|
||||
check(
|
||||
"version not on this site: requested, not loaded",
|
||||
result == {"CANCELLED"} and "chair" not in loaded()
|
||||
and ["wtd_request_revisions", {"revision_oids": [chair_file["file_oid"] + "/history/revisions/v002"]}] in fake("/_test/calls", {}),
|
||||
)
|
||||
bpy.ops.wtd_scene.load_publish("EXEC_DEFAULT", file=chair_file["file_oid"], revision="v001", count=2)
|
||||
check("loaded twice, numbered", loaded().get("chair") == ("v001", ["chair_01", "chair_02"]), str(loaded().get("chair")))
|
||||
check("its versions followed in Loaded", sm.version_status(bpy.data.libraries["chair"])[0] == "PENDING")
|
||||
sm.browse_asset(next(a for a in sm.STATE.assets if a["asset"] == "bob"))
|
||||
check(
|
||||
"another version of a loaded file refused",
|
||||
bpy.ops.wtd_scene.load_publish("EXEC_DEFAULT", file=BOB_FILE, revision="v004") == {"CANCELLED"} and loaded()["bob"][0] == "v002",
|
||||
)
|
||||
bpy.ops.wtd_scene.load_publish("EXEC_DEFAULT", file=BOB_FILE, revision="v002")
|
||||
check("same version: one more numbered instance", loaded()["bob"][1] == ["COL_bob_01", "COL_bob_02", "COL_bob_03"], str(loaded()["bob"][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_01"]), 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", "car", "chair", "lamp", "rock", "tree"})
|
||||
car_lib = bpy.data.libraries["car"]
|
||||
check("LOD kept after reopening", lib.current_lod(car_lib, lib.instances(car_lib)[0]) == "low")
|
||||
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