Source code for libmuscle.manager.test.test_instance_registry

import pytest
from ymmsl.v0_2 import Operator

from libmuscle.manager.instance_registry import InstanceRegistry, Port




@pytest.fixture
def port():
    return Port('test_port', Operator.F_INIT)





@pytest.fixture
def registry():
    return InstanceRegistry()





def test_port(port):
    assert port.name == 'test_port'
    assert port.operator == Operator.F_INIT





def test_registry_add(registry, port):
    registry.add('instance1', 'tcp://localhost:6253', [port])
    assert (registry._locations['instance1'] ==
            'tcp://localhost:6253')
    assert registry._ports['instance1'] == [port]





def test_registry_get(registry, port):
    registry._locations['instance1'] = [
            'tcp://localhost:6253']
    registry._ports['instance1'] = [port]
    assert registry.get_locations('instance1') == ['tcp://localhost:6253']
    assert registry.get_ports('instance1') == [port]

    with pytest.raises(KeyError):
        registry.get_locations('non-existant-instance')

    with pytest.raises(KeyError):
        registry.get_ports('non-existant-instance')





def test_registry_remove(registry, port):
    registry._locations['instance1'] = [
            'tcp://localhost:6253']
    registry._ports['instance1'] = [port]
    registry.remove('instance1')
    assert 'instance1' not in registry._locations
    assert 'instance1' not in registry._ports

    with pytest.raises(KeyError):
        registry.remove('non-existant-instance')