Skip to content

🧪 Testing Example

dagpipe's own test suite runs without any external services — nodes are the only things you need to fake. The pattern: small schema/state factories, then assert on terminal states and history.

🎯 Setup: state factories

Import from dagpipe and declare tiny schemas/states, exactly like the project's conftest.py:

 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
# conftest.py
import pytest
from dagpipe import Payload, Schema, State

TestSchema = Schema({"value": object})
ItemSchema = Schema({
    "raw": object,
    "entity": Schema({"candidate": str | None, "name": str | None}),
    "tags": list | None,
})

class TestState(State):
    schema = TestSchema

class ItemState(State):
    schema = ItemSchema

@pytest.fixture
def make_state():
    def _make(value=1):
        return TestState(payload=Payload({"value": value}))
    return _make

@pytest.fixture
def make_item_state():
    def _make(**payload):
        base = {"raw": object(), "tags": []}
        base.update(payload)
        return ItemState(payload=Payload(base))
    return _make

🏗️ A node to test

1
2
3
4
5
6
7
from dagpipe import Node

class Increment(Node):
    id = "test.increment"

    def resolve(self, state):
        yield self.fork(state, payload_update={"value": state.get("value") + 1})

✅ Single-node test

1
2
3
4
5
6
7
def test_increment(make_state):
    node = Increment()
    (result,) = node.run(make_state(value=1))

    assert result.get("value") == 2
    assert result.depth == 1
    assert result.history == ("test.increment",)

🔀 Graph behavior test

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from dagpipe import Engine, Graph

def test_branching_and_merge(make_item_state):
    class Extract(Node):
        id = "entity.extract"
        def resolve(self, state):
            yield self.fork(state, payload_update={"entity.candidate": "amzn"})
            yield self.fork(state, payload_update={"entity.candidate": "flipkart"})

    class Resolve(Node):
        id = "entity.resolve"
        def resolve(self, state):
            if state.get("entity.candidate"):
                yield self.fork(state, payload_update={
                    "entity.name": state.get("entity.candidate").upper()})

    graph = Graph()
    graph.add_edge(Extract(), Resolve())

    results = Engine(graph).run(make_item_state())
    names = sorted(s.get("entity.name") for s in results)

    assert names == ["AMZN", "FLIPKART"]
    assert all(len(s.history) == 2 for s in results)   # extract → resolve

⚠️ Error-path tests

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import pytest
from dagpipe import Payload, Schema, SchemaError, State

def test_invalid_payload_raises():
    class S(State):
        schema = Schema({"name": str})

    with pytest.raises(SchemaError, match="must be str"):
        S(payload=Payload({"name": 42}))

def test_bad_yield_raises_type_error(make_state):
    class Bad(Node):
        id = "bad.output"
        def resolve(self, state):
            yield "nope"

    with pytest.raises(TypeError, match="must yield State"):
        Bad().run(make_state())

🧱 Testing YAML pipelines offline

Use a tiny importable node module + tmp_path, as the integration suite does:

 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
def test_yaml_pipeline(tmp_path, monkeypatch):
    module_dir = tmp_path / "testnodes"
    module_dir.mkdir()
    (module_dir / "__init__.py").write_text("""
from dagpipe.node import Node

class Increment(Node):
    id = "test.increment"
    def resolve(self, state):
        yield self.fork(state, payload_update={"value": state.get("value") + 1})
""")

    monkeypatch.syspath_prepend(tmp_path)

    yaml_file = tmp_path / "p.yaml"
    yaml_file.write_text("""
schema:
  value: int
initial:
  value: 1
nodes:
  inc:
    class: testnodes.Increment
graph:
  roots:
    - inc
""")

    from dagpipe import load_pipeline
    results = load_pipeline(yaml_file).run()

    assert results[0].get("value") == 2

💡 Tips

  • Assert on terminal states + history — they capture the full behavior without brittle intermediate assertions.
  • Factory fixtures keep tests readable: make_item_state(entity={"candidate": "x"}) beats hand-building Payload dicts everywhere.
  • Guard against side effects — nodes must be pure; tests catch leaks when the input state is still unchanged after run (assert on it explicitly).
  • Run the suite with one command from the repo root:
1
pytest -q