Source code for libmuscle.manager.test.test_topology_store

from ymmsl.v0_2 import Reference




def test_create_topology_store(topology_store) -> None:
    assert topology_store.model.conduits[0].sender == 'macro.out'
    assert topology_store.model.conduits[0].receiver == 'micro.in'
    assert topology_store.model.conduits[1].sender == 'micro.out'
    assert topology_store.model.conduits[1].receiver == 'macro.in'

    assert topology_store.model.components['macro'].multiplicity == []
    assert topology_store.model.components['micro'].multiplicity == [10, 10]





def test_get_conduits(topology_store2) -> None:
    conduits = topology_store2.get_conduits(Reference('macro'))
    assert conduits[0].sender == 'macro.out'
    assert conduits[0].receiver == 'meso.init'
    assert conduits[1].sender == 'meso.final'
    assert conduits[1].receiver == 'macro.in'

    conduits = topology_store2.get_conduits(Reference('meso'))
    assert conduits[0].sender == 'macro.out'
    assert conduits[0].receiver == 'meso.init'
    assert conduits[1].sender == 'meso.out'
    assert conduits[1].receiver == 'micro.init'
    assert conduits[2].sender == 'micro.final'
    assert conduits[2].receiver == 'meso.in'
    assert conduits[3].sender == 'meso.final'
    assert conduits[3].receiver == 'macro.in'





def test_get_peer_dimensions(topology_store) -> None:
    macro_peer_dims = topology_store.get_peer_dimensions(Reference('macro'))

    assert len(macro_peer_dims) == 1
    assert Reference('micro') in macro_peer_dims
    assert macro_peer_dims[Reference('micro')] == [10, 10]

    micro_peer_dims = topology_store.get_peer_dimensions(Reference('micro'))

    assert len(micro_peer_dims) == 1
    assert Reference('macro') in micro_peer_dims
    assert micro_peer_dims[Reference('macro')] == []