test_node.py 文件源码

def test_node_creation_args():
    """Checks that a node can be passed an argument using inheritance"""
    ns = multiprocessing.Manager().Namespace()
    ns.arg = 42

    class TestArgNode(pyzmp.Node):
        def update(self, *args, **kwargs):
            ns.arg -= args[0]
            return ns.arg

    n1 = TestArgNode(args=(ns.arg,))
    assert not n1.is_alive()
    svc_url = n1.start()
    assert n1.is_alive()
    assert svc_url

    # starting and shutdown should at least guarantee ONE call of update function.

    exitcode = n1.shutdown()
    assert exitcode == 0
    assert not n1.is_alive()

    assert ns.arg == 0