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
from libmuscle.port_manager import PortManager
[docs]
@pytest.fixture
def index() -> list[int]:
return [13]
[docs]
@pytest.fixture
def port_manager(index) -> PortManager:
declared_ports = {Operator.O_I: ["out"], Operator.S: ["in"]}
return PortManager(index, declared_ports)
[docs]
@pytest.fixture
def index2() -> list[int]:
return []
[docs]
@pytest.fixture
def port_manager2(index2) -> PortManager:
declared_ports = {Operator.F_INIT: ["in[]"], Operator.O_F: ["out[]"]}
port_manager = PortManager(index2, declared_ports)
component_id = Ref("other")
conduits = [
Conduit("component.out", "other.in"),
Conduit("other.out", "component.in"),
]
peer_dims = {Ref("component"): [20]}
peer_locations = {Ref("component"): ["direct:test"]}
peer_info = PeerInfo(component_id, index2, conduits, peer_dims, peer_locations, [])
port_manager.connect_ports(peer_info)
return port_manager
[docs]
def test_connect_ports(index, port_manager) -> None:
component_id = Ref("component")
conduits = [
Conduit("component.out", "other.in"),
Conduit("other.settings_out", "component.muscle_settings_in"),
Conduit("other.out", "component.in"),
]
peer_dims = {Ref("other"): []}
peer_locations = {Ref("other"): ["direct:test"]}
peer_info = PeerInfo(component_id, index, conduits, peer_dims, peer_locations, [])
port_manager.connect_ports(peer_info)
# check automatic ports
assert port_manager.settings_in_connected()
assert port_manager._muscle_settings_in.name == Id("muscle_settings_in")
assert port_manager._muscle_settings_in.operator == Operator.F_INIT
assert port_manager._muscle_settings_in._length is None
# check declared ports
ports = port_manager._ports
assert ports["in"].name == Id("in")
assert ports["in"].operator == Operator.S
assert ports["in"]._length is None
assert ports["out"].name == Id("out")
assert ports["out"].operator == Operator.O_I
assert ports["out"]._length is None
[docs]
def test_connect_vector_ports(index) -> None:
declared_ports = {Operator.F_INIT: ["in[]"], Operator.O_F: ["out1", "out2[]"]}
port_manager = PortManager(index, declared_ports)
component_id = Ref("component")
conduits = [
Conduit("other1.out", "component.in"),
Conduit("component.out1", "other.in"),
Conduit("component.out2", "other3.in"),
]
peer_dims = {Ref("other1"): [20, 7], Ref("other"): [25], Ref("other3"): [20]}
peer_locations = {
Ref("other"): ["direct:test"],
Ref("other1"): ["direct:test1"],
Ref("other3"): ["direct:test3"],
}
peer_info = PeerInfo(component_id, index, conduits, peer_dims, peer_locations, [])
port_manager.connect_ports(peer_info)
ports = port_manager._ports
assert ports["in"].name == Id("in")
assert ports["in"].operator == Operator.F_INIT
assert ports["in"]._length == 7
assert ports["in"]._is_resizable is False
assert ports["out1"].name == Id("out1")
assert ports["out1"].operator == Operator.O_F
assert ports["out1"]._length is None
assert ports["out2"].name == Id("out2")
assert ports["out2"].operator == Operator.O_F
assert ports["out2"]._length == 0
assert ports["out2"]._is_resizable is True
[docs]
def test_connect_multidimensional_ports() -> None:
index = [13]
declared_ports = {Operator.F_INIT: ["in[][]"]}
port_manager = PortManager(index, declared_ports)
component_id = Ref("component")
conduits = [Conduit("other.out", "component.in")]
peer_dims = {Ref("other"): [20, 7, 30]}
peer_locations = {Ref("other"): ["direct:test"]}
peer_info = PeerInfo(component_id, index, conduits, peer_dims, peer_locations, [])
with pytest.raises(ValueError):
port_manager.connect_ports(peer_info)
[docs]
def test_connect_inferred_ports() -> None:
index = [13]
port_manager = PortManager(index, None)
component_id = Ref("component")
conduits = [
Conduit("other1.out", "component.in"),
Conduit("component.out1", "other.in"),
Conduit("component.out3", "other2.in"),
]
peer_dims = {Ref("other1"): [20, 7], Ref("other"): [25], Ref("other2"): []}
peer_locations = {
Ref("other"): ["direct:test"],
Ref("other1"): ["direct:test1"],
Ref("other2"): ["direct:test2"],
}
ymmsl_ports = [
Port(Id("in"), Operator.F_INIT),
Port(Id("out1"), Operator.O_F),
Port(Id("out3"), Operator.O_F),
]
peer_info = PeerInfo(
component_id, index, conduits, peer_dims, peer_locations, ymmsl_ports
)
port_manager.connect_ports(peer_info)
ports = port_manager._ports
assert ports["in"].name == Id("in")
assert ports["in"].operator == Operator.F_INIT
assert ports["in"]._length == 7
assert ports["in"]._is_resizable is False
assert ports["out1"].name == Id("out1")
assert ports["out1"].operator == Operator.O_F
assert ports["out1"]._length is None
assert ports["out3"].name == Id("out3")
assert ports["out3"].operator == Operator.O_F
assert ports["out3"]._length is None
[docs]
def test_connect_inferred_ports_oi_s() -> None:
port_manager = PortManager([], None)
peer_info = PeerInfo(
Ref("Component"),
[],
[Conduit("component.oi", "other.init"), Conduit("other.final", "component.s")],
{Ref("other"): []},
{Ref("other"): ["direct:test"]},
[Port(Id("oi"), Operator.O_I), Port(Id("s"), Operator.S)],
)
port_manager.connect_ports(peer_info)
ports = port_manager._ports
assert ports["oi"].name == Id("oi")
assert ports["oi"].operator == Operator.O_I
assert ports["oi"]._length is None
assert ports["s"].name == Id("s")
assert ports["s"].operator == Operator.S
assert ports["s"]._length is None
[docs]
def test_port_message_counts(port_manager) -> None:
port_manager.connect_ports(PeerInfo("component", [], [], {}, {}, []))
msg_counts = port_manager.get_message_counts()
assert msg_counts == {"in": [0], "out": [0], "muscle_settings_in": [0]}
port_manager.get_port("in").increment_num_messages()
msg_counts2 = port_manager.get_message_counts()
assert msg_counts2 == {"in": [1], "out": [0], "muscle_settings_in": [0]}
port_manager.get_port("out").increment_num_messages()
port_manager.get_port("out").increment_num_messages()
msg_counts3 = port_manager.get_message_counts()
assert msg_counts3 == {"in": [1], "out": [2], "muscle_settings_in": [0]}
port_manager._muscle_settings_in.increment_num_messages()
msg_counts4 = port_manager.get_message_counts()
assert msg_counts4 == {"in": [1], "out": [2], "muscle_settings_in": [1]}
port_manager.restore_message_counts(msg_counts)
msg_counts5 = port_manager.get_message_counts()
assert msg_counts5 == {"in": [0], "out": [0], "muscle_settings_in": [0]}
with pytest.raises(RuntimeError):
port_manager.restore_message_counts({"x?invalid_port": 3})
[docs]
def test_vector_port_message_counts(port_manager2) -> None:
msg_counts = port_manager2.get_message_counts()
assert msg_counts == {"out": [0] * 20, "in": [0] * 20, "muscle_settings_in": [0]}
port_manager2.get_port("out").increment_num_messages(13)
msg_counts = port_manager2.get_message_counts()
assert msg_counts == {
"out": [0] * 13 + [1] + [0] * 6,
"in": [0] * 20,
"muscle_settings_in": [0],
}
port_manager2.restore_message_counts(
{"out": list(range(20)), "in": list(range(20)), "muscle_settings_in": [4]}
)
port_manager2.get_port("out").increment_num_messages(13)
msg_counts = port_manager2.get_message_counts()
assert msg_counts == {
"out": list(range(13)) + [14] + list(range(14, 20)),
"in": list(range(20)),
"muscle_settings_in": [4],
}