From 4161ec4ad57bf64b3952fe1b140a72b5cdc28a53 Mon Sep 17 00:00:00 2001 From: Nero WTD Date: Wed, 28 Oct 2026 11:24:44 +0100 Subject: [PATCH] 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 --- dev/fake_libreflow.py | 196 ----------------------- dev/make_test_assets.py | 72 --------- dev/test_scene_manager.py | 320 -------------------------------------- 3 files changed, 588 deletions(-) delete mode 100644 dev/fake_libreflow.py delete mode 100644 dev/make_test_assets.py delete mode 100644 dev/test_scene_manager.py diff --git a/dev/fake_libreflow.py b/dev/fake_libreflow.py deleted file mode 100644 index 0875f45..0000000 --- a/dev/fake_libreflow.py +++ /dev/null @@ -1,196 +0,0 @@ -"""Fake Libreflow IPC server (kabaret.ipc + wtdcmds commands) to test the Scene Manager. - - python dev/fake_libreflow.py - -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 (/lib). Writes /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() diff --git a/dev/make_test_assets.py b/dev/make_test_assets.py deleted file mode 100644 index 26dd654..0000000 --- a/dev/make_test_assets.py +++ /dev/null @@ -1,72 +0,0 @@ -"""Write the asset files used by fake_libreflow.py into /lib. - - blender -b --factory-startup --python dev/make_test_assets.py -- - -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) diff --git a/dev/test_scene_manager.py b/dev/test_scene_manager.py deleted file mode 100644 index 0644a10..0000000 --- a/dev/test_scene_manager.py +++ /dev/null @@ -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 -- - python dev/fake_libreflow.py (keep it running, fresh state) - blender -b --factory-startup --python-exit-code 1 --python dev/test_scene_manager.py -- - -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")