test_address.py 文件源码

def test_agentchannel_sync_pub():
    """
    Test basic SYNC_PUB AgentChannel operations: initialization, equivalence
    and basic methods.
    """
    sender = AgentAddress('ipc', 'addr0', 'PUB', 'server', 'pickle')
    receiver = AgentAddress('ipc', 'addr0', 'PULL', 'server', 'pickle')
    channel = AgentChannel('SYNC_PUB', sender=sender, receiver=receiver)
    # Equivalence
    assert channel == AgentChannel('SYNC_PUB', sender=sender,
                                   receiver=receiver)
    assert not channel == 'foo'
    assert channel != 'foo'
    # Basic methods
    assert channel.twin() == AgentChannel('SYNC_SUB', sender=receiver.twin(),
                                          receiver=sender.twin())
    # Other attributes
    assert hasattr(channel, 'uuid')
    assert channel.transport == 'ipc'
    assert channel.serializer == 'pickle'