test_node.py 文件源码

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

    def arguser(fortytwo, **kwargs):  # kwargs is there to accept extra arguments nicely (timedelta)
        ns.arg -= fortytwo
        return ns.arg

    n1 = pyzmp.Node(args=(ns.arg,), target=arguser)
    assert not n1.is_alive()
    svc_url = n1.start()
    assert n1.is_alive()
    assert svc_url

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

    assert ns.arg == 0