41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from synaptopus.graph import GraphEdgeSpec, GraphSchema, categorizer_node, critic_node, generator_node, policy_node
|
|
from testsupport import (
|
|
AcceptEvenHighPolicy,
|
|
CounterState,
|
|
EvenCritic,
|
|
IncrementingGenerator,
|
|
ThresholdCategorizer,
|
|
)
|
|
|
|
|
|
def test_graph_node_specs_and_schema_are_json_safe(tmp_path) -> None:
|
|
nodes = (
|
|
generator_node("gen", IncrementingGenerator()),
|
|
critic_node("crit", EvenCritic()),
|
|
categorizer_node("cat", ThresholdCategorizer()),
|
|
policy_node("pol", AcceptEvenHighPolicy()),
|
|
)
|
|
schema = GraphSchema(
|
|
nodes=tuple(node.spec() for node in nodes),
|
|
edges=(
|
|
GraphEdgeSpec("gen", "candidate", "crit", "candidate"),
|
|
GraphEdgeSpec("gen", "candidate", "cat", "candidate"),
|
|
GraphEdgeSpec("gen", "candidate", "pol", "candidate"),
|
|
GraphEdgeSpec("crit", "critique", "pol", "critique"),
|
|
GraphEdgeSpec("cat", "category", "pol", "category"),
|
|
),
|
|
)
|
|
|
|
destination = tmp_path / "graph.json"
|
|
schema.save_json(destination)
|
|
loaded = json.loads(destination.read_text(encoding="utf-8"))
|
|
|
|
assert loaded["artifact_type"] == "graph_schema"
|
|
assert loaded["payload"]["nodes"][0]["node_id"] == "gen"
|
|
assert loaded["payload"]["nodes"][0]["node_type"] == "generator"
|
|
assert loaded["payload"]["edges"][0]["source_output"] == "candidate"
|