test_node.py 文件源码

def test_node_creation_kwargs_delegate():
    """Checks that a node can be passed a keyword argument using delegation"""
    ns = multiprocessing.Manager().Namespace()
    ns.kwarg = 42

    def kwarguser(intval, **kwargs):  # kwargs is there to accept extra arguments nicely (timedelta)
        ns.kwarg -= intval
        return ns.kwarg

    n1 = pyzmp.Node(kwargs={'intval': ns.kwarg, }, target=kwarguser)
    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.kwarg == 0


# @nose.SkipTest  # to help debugging ( FIXME : how to programmatically start only one test - maybe in fixture - ? )