Source code for libmuscle.test.test_peer_info

import pytest
from ymmsl.v0_2 import Conduit, Operator, Port
from ymmsl.v0_2 import Identifier as Id
from ymmsl.v0_2 import Reference as Ref

from libmuscle.peer_info import PeerInfo




@pytest.fixture
def peer_info() -> PeerInfo:
    kernel = Ref("kernel")
    index = [13]

    conduits = [Conduit("kernel.out", "other.in"), Conduit("other.out", "kernel.in")]

    peer_dims = {Ref("other"): []}

    peer_locations = {Ref("other"): ["tcp:other"]}

    ymmsl_ports = [Port(Id("out"), Operator.O_I), Port(Id("in"), Operator.S)]

    return PeerInfo(kernel, index, conduits, peer_dims, peer_locations, ymmsl_ports)





@pytest.fixture
def peer_info2() -> PeerInfo:
    kernel = Ref("other")
    index = []

    conduits = [Conduit("kernel.out", "other.in"), Conduit("other.out", "kernel.in")]

    peer_dims = {Ref("kernel"): [20]}

    peer_locations = {Ref("kernel"): ["tcp:kernel"]}

    ymmsl_ports = [Port(Id("out"), Operator.O_F), Port(Id("in"), Operator.F_INIT)]

    return PeerInfo(kernel, index, conduits, peer_dims, peer_locations, ymmsl_ports)





@pytest.fixture
def peer_info3() -> PeerInfo:
    kernel = Ref("kernel")
    index = []

    conduits = [Conduit("kernel.out", "other.in"), Conduit("other.out", "kernel.in")]

    peer_dims = {Ref("other"): []}

    peer_locations = {Ref("other"): ["tcp:other"]}

    ymmsl_ports = [Port(Id("out"), Operator.O_I), Port(Id("in"), Operator.S)]

    return PeerInfo(kernel, index, conduits, peer_dims, peer_locations, ymmsl_ports)





def test_create_peer_info(peer_info) -> None:
    assert True





def test_is_connected(peer_info) -> None:
    assert peer_info.is_connected(Id("out"))
    assert peer_info.is_connected(Id("in"))
    assert not peer_info.is_connected(Id("not_connected"))





def test_get_peer_port(peer_info, peer_info2, peer_info3) -> None:
    assert peer_info.get_peer_ports(Id("out")) == [Ref("other.in")]
    assert peer_info.get_peer_ports(Id("in")) == [Ref("other.out")]

    assert peer_info2.get_peer_ports(Id("out")) == [Ref("kernel.in")]
    assert peer_info2.get_peer_ports(Id("in")) == [Ref("kernel.out")]

    assert peer_info3.get_peer_ports(Id("out")) == [Ref("other.in")]
    assert peer_info3.get_peer_ports(Id("in")) == [Ref("other.out")]





def test_get_peer_dims(peer_info, peer_info2, peer_info3) -> None:
    assert peer_info.get_peer_dims(Ref("other")) == []
    assert peer_info2.get_peer_dims(Ref("kernel")) == [20]
    assert peer_info3.get_peer_dims(Ref("other")) == []





def test_get_peer_locations(peer_info, peer_info2, peer_info3) -> None:
    assert peer_info.get_peer_locations(Ref("other")) == ["tcp:other"]
    assert peer_info2.get_peer_locations(Ref("kernel")) == ["tcp:kernel"]
    assert peer_info3.get_peer_locations(Ref("other")) == ["tcp:other"]





def test_get_peer_endpoint(peer_info, peer_info2, peer_info3) -> None:
    assert str(peer_info.get_peer_endpoints(Id("out"), [])[0]) == "other.in[13]"
    assert str(peer_info.get_peer_endpoints(Id("in"), [])[0]) == "other.out[13]"

    assert str(peer_info2.get_peer_endpoints(Id("out"), [11])[0]) == "kernel[11].in"
    assert str(peer_info2.get_peer_endpoints(Id("in"), [11])[0]) == "kernel[11].out"

    assert str(peer_info3.get_peer_endpoints(Id("out"), [42])[0]) == "other.in[42]"
    assert str(peer_info3.get_peer_endpoints(Id("in"), [42])[0]) == "other.out[42]"





def test_get_ymmsl_ports(peer_info) -> None:
    assert peer_info.list_ymmsl_ports() == [
        Port(Id("out"), Operator.O_I),
        Port(Id("in"), Operator.S),
    ]