import pytest
from ymmsl.v0_2 import Component, Conduit, Configuration, Model, Ports, Reference
from libmuscle.manager.deadlock_detector import DeadlockDetector
from libmuscle.manager.instance_registry import InstanceRegistry
from libmuscle.manager.logger import Logger
from libmuscle.manager.mmp_server import MMPRequestHandler
from libmuscle.manager.profile_store import ProfileStore
from libmuscle.manager.run_dir import RunDir
from libmuscle.manager.snapshot_registry import SnapshotRegistry
from libmuscle.manager.topology_store import TopologyStore
[docs]
@pytest.fixture
def logger(tmp_path):
test_logger = Logger(tmp_path)
yield test_logger
test_logger.close()
[docs]
@pytest.fixture
def mmp_configuration():
return Configuration(
'mmp_configuration', [], [Model(
'test_model', None, '', None,
[
Component(
'macro', Ports(o_i=['out'], s=['in']), '',
'macro_implementation'),
Component(
'micro', Ports(f_init=['in'], o_f=['out']), '',
'micro_implementation', False, [10, 10])],
[
Conduit('macro.out', 'micro.in'),
Conduit('micro.out', 'macro.in')
])])
[docs]
@pytest.fixture
def profile_store(tmp_path):
test_profile_store = ProfileStore(tmp_path)
yield test_profile_store
test_profile_store.shutdown()
[docs]
@pytest.fixture
def instance_registry():
return InstanceRegistry()
[docs]
@pytest.fixture
def topology_store(mmp_configuration) -> TopologyStore:
return TopologyStore(mmp_configuration)
[docs]
@pytest.fixture
def snapshot_registry(mmp_configuration, topology_store) -> SnapshotRegistry:
return SnapshotRegistry(mmp_configuration, None, topology_store)
[docs]
@pytest.fixture
def deadlock_detector() -> DeadlockDetector:
return DeadlockDetector()
[docs]
@pytest.fixture
def mmp_request_handler(
logger, profile_store, mmp_configuration, instance_registry,
topology_store, snapshot_registry, deadlock_detector):
return MMPRequestHandler(
logger, profile_store, mmp_configuration, instance_registry,
topology_store, snapshot_registry, deadlock_detector, None)
[docs]
@pytest.fixture
def loaded_instance_registry(instance_registry):
instance_registry.add(Reference('macro'), ['direct:macro'], [])
for j in range(10):
for i in range(10):
name = Reference('micro') + j + i
location = f'direct:{name}'
instance_registry.add(name, [location], [])
return instance_registry
[docs]
@pytest.fixture
def registered_mmp_request_handler(
logger, profile_store, mmp_configuration, loaded_instance_registry,
topology_store, snapshot_registry, deadlock_detector):
return MMPRequestHandler(
logger, profile_store, mmp_configuration, loaded_instance_registry,
topology_store, snapshot_registry, deadlock_detector, None)
[docs]
@pytest.fixture
def mmp_configuration2():
return Configuration(
'mmp_configuration2', [], [Model(
'test_model', None, '', None,
[
Component(
'macro', Ports(o_i=['out'], s=['in']), '',
'macro_implementation'),
Component(
'meso',
Ports(f_init=['init'], o_i=['out'], s=['in'], o_f=['final']),
'', 'meso_implementation', False, [5]),
Component(
'micro', Ports(f_init=['init'], o_f=['final']), '',
'micro_implementation', False, [5, 10])
],
[
Conduit('macro.out', 'meso.init'),
Conduit('meso.out', 'micro.init'),
Conduit('micro.final', 'meso.in'),
Conduit('meso.final', 'macro.in')
])])
[docs]
@pytest.fixture
def topology_store2(mmp_configuration2) -> TopologyStore:
return TopologyStore(mmp_configuration2)
[docs]
@pytest.fixture
def snapshot_registry2(mmp_configuration2, topology_store) -> SnapshotRegistry:
return SnapshotRegistry(mmp_configuration2, None, topology_store)
[docs]
@pytest.fixture
def loaded_instance_registry2():
instance_registry = InstanceRegistry()
instance_registry.add(Reference('macro'), ['direct:macro'], [])
for j in range(5):
name = Reference('meso') + j
location = f'direct:{name}'
instance_registry.add(name, [location], [])
for j in range(5):
for i in range(10):
name = Reference('micro') + j + i
location = f'direct:{name}'
instance_registry.add(name, [location], [])
return instance_registry
[docs]
@pytest.fixture
def registered_mmp_request_handler2(
logger, profile_store, mmp_configuration, loaded_instance_registry2,
topology_store2, snapshot_registry2, deadlock_detector, tmp_path):
return MMPRequestHandler(
logger, profile_store, mmp_configuration,
loaded_instance_registry2, topology_store2, snapshot_registry2,
deadlock_detector, RunDir(tmp_path))