Spaces:
Running
Running
File size: 4,108 Bytes
017ba8a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 |
import os
import pytest
import tempfile
from lynxkite.core import workspace
from lynxkite.core import ops
def test_save_load():
ws = workspace.Workspace(env="test")
ws.nodes.append(
workspace.WorkspaceNode(
id="1",
type="node_type",
data=workspace.WorkspaceNodeData(title="Node 1", params={}),
position=workspace.NodePosition(x=0, y=0),
)
)
ws.nodes.append(
workspace.WorkspaceNode(
id="2",
type="node_type",
data=workspace.WorkspaceNodeData(title="Node 2", params={}),
position=workspace.NodePosition(x=0, y=0),
)
)
ws.edges.append(
workspace.WorkspaceEdge(
id="edge1",
source="1",
target="2",
sourceHandle="",
targetHandle="",
)
)
path = os.path.join(tempfile.gettempdir(), "test_workspace.json")
try:
workspace.save(ws, path)
assert os.path.exists(path)
loaded_ws = workspace.load(path)
assert loaded_ws.env == ws.env
assert len(loaded_ws.nodes) == len(ws.nodes)
assert len(loaded_ws.edges) == len(ws.edges)
sorted_ws_nodes = sorted(ws.nodes, key=lambda x: x.id)
sorted_loaded_ws_nodes = sorted(loaded_ws.nodes, key=lambda x: x.id)
# We do manual assertion on each attribute because metadata is added at
# loading time, which makes the objects different.
for node, loaded_node in zip(sorted_ws_nodes, sorted_loaded_ws_nodes):
assert node.id == loaded_node.id
assert node.type == loaded_node.type
assert node.data.title == loaded_node.data.title
assert node.data.params == loaded_node.data.params
assert node.position.x == loaded_node.position.x
assert node.position.y == loaded_node.position.y
sorted_ws_edges = sorted(ws.edges, key=lambda x: x.id)
sorted_loaded_ws_edges = sorted(loaded_ws.edges, key=lambda x: x.id)
for edge, loaded_edge in zip(sorted_ws_edges, sorted_loaded_ws_edges):
assert edge.id == loaded_edge.id
assert edge.source == loaded_edge.source
assert edge.target == loaded_edge.target
assert edge.sourceHandle == loaded_edge.sourceHandle
assert edge.targetHandle == loaded_edge.targetHandle
finally:
os.remove(path)
@pytest.fixture(scope="session", autouse=True)
def populate_ops_catalog():
ops.register_passive_op(
env="test",
name="Test Operation",
inputs=[],
params=[
ops.Parameter(name="param_int", default=1),
ops.Parameter(name="param_str", default="test"),
],
)
def test_update_metadata():
ws = workspace.Workspace(env="test")
ws.nodes.append(
workspace.WorkspaceNode(
id="1",
type="basic",
data=workspace.WorkspaceNodeData(
title="Test Operation", params={"param_int": 1}
),
position=workspace.NodePosition(x=0, y=0),
)
)
ws.nodes.append(
workspace.WorkspaceNode(
id="2",
type="basic",
data=workspace.WorkspaceNodeData(title="Unknown Operation", params={}),
position=workspace.NodePosition(x=0, y=0),
)
)
updated_ws = workspace._update_metadata(ws)
assert updated_ws.nodes[0].data.meta.name == "Test Operation"
assert updated_ws.nodes[0].data.error is None
assert updated_ws.nodes[0].data.params == {"param_int": 1}
assert updated_ws.nodes[0].data.meta.params == {
"param_int": ops.Parameter(name="param_int", default=1),
"param_str": ops.Parameter(name="param_str", default="test"),
}
assert not hasattr(updated_ws.nodes[1].data, "meta")
assert updated_ws.nodes[1].data.error == "Unknown operation."
def test_update_metadata_with_empty_workspace():
ws = workspace.Workspace(env="test")
updated_ws = workspace._update_metadata(ws)
assert len(updated_ws.nodes) == 0
|