from importlib import import_module import pytest from tiramisu import list_sessions, delete_session from .storage import STORAGE from risotto.context import Context from risotto.services import load_services from risotto.dispatcher import dispatcher def setup_module(module): load_services(['config'], validate=False) config_module = dispatcher.get_service('config') config_module.save_storage = STORAGE dispatcher.set_module('server', import_module(f'.server', 'fake_services'), True) dispatcher.set_module('servermodel', import_module(f'.servermodel', 'fake_services'), True) def setup_function(function): config_module = dispatcher.get_service('config') config_module.server = {} config_module.servermodel = {} def teardown_function(function): # delete all sessions config_module = dispatcher.get_service('config') for session in list_sessions(storage=config_module.save_storage): delete_session(storage=config_module.save_storage, session_id=session) def get_fake_context(module_name): risotto_context = Context() risotto_context.username = 'test' risotto_context.paths.append(f'{module_name}.on_join') risotto_context.type = None return risotto_context @pytest.mark.asyncio async def test_on_join(): config_module = dispatcher.get_service('config') assert config_module.servermodel == {} assert config_module.server == {} # fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) assert list(config_module.servermodel.keys()) == [1, 2] assert list(config_module.server) == [3] assert set(config_module.server[3]) == {'server', 'server_to_deploy', 'funcs_file'} assert config_module.server[3]['funcs_file'] == 'tests/data/1/funcs.py' @pytest.mark.asyncio async def test_server_created(): config_module = dispatcher.get_service('config') fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) # assert list(config_module.server) == [3] await dispatcher.publish('v1', 'server.created', fake_context, server_id=4, server_name='name3', server_description='description3', server_servermodel_id=2) assert list(config_module.server) == [3, 4] assert set(config_module.server[4]) == {'server', 'server_to_deploy', 'funcs_file'} assert config_module.server[4]['funcs_file'] == 'tests/data/2/funcs.py' @pytest.mark.asyncio async def test_server_deleted(): config_module = dispatcher.get_service('config') fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) # assert list(config_module.server) == [3] await dispatcher.publish('v1', 'server.created', fake_context, server_id=4, server_name='name4', server_description='description4', server_servermodel_id=2) assert list(config_module.server) == [3, 4] await dispatcher.publish('v1', 'server.deleted', fake_context, server_id=4) assert list(config_module.server) == [3] @pytest.mark.asyncio async def test_servermodel_created(): config_module = dispatcher.get_service('config') fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) # assert list(config_module.servermodel) == [1, 2] servermodel = {'servermodeid': 3, 'servermodelname': 'name3'} await dispatcher.publish('v1', 'servermodel.created', fake_context, servermodel_id=3, servermodel_description='name3', release_id=1, servermodel_name='name3') assert list(config_module.servermodel) == [1, 2, 3] assert not list(await config_module.servermodel[3].config.parents()) @pytest.mark.asyncio async def test_servermodel_herited_created(): config_module = dispatcher.get_service('config') fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) # assert list(config_module.servermodel) == [1, 2] await dispatcher.publish('v1', 'servermodel.created', fake_context, servermodel_id=3, servermodel_name='name3', release_id=1, servermodel_description='name3', servermodel_parents_id=[1]) assert list(config_module.servermodel) == [1, 2, 3] assert len(list(await config_module.servermodel[3].config.parents())) == 1 @pytest.mark.asyncio async def test_servermodel_multi_herited_created(): config_module = dispatcher.get_service('config') fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) # assert list(config_module.servermodel) == [1, 2] await dispatcher.publish('v1', 'servermodel.created', fake_context, servermodel_id=3, servermodel_name='name3', release_id=1, servermodel_description='name3', servermodel_parents_id=[1, 2]) assert list(config_module.servermodel) == [1, 2, 3] assert len(list(await config_module.servermodel[3].config.parents())) == 2 #@pytest.mark.asyncio #async def test_servermodel_updated_not_exists(): # config_module = dispatcher.get_service('config') # fake_context = get_fake_context('config') # config_module.cache_root_path = 'tests/data' # await config_module.on_join(fake_context) # # # assert list(config_module.servermodel) == [1, 2] # await dispatcher.publish('v1', # 'servermodel.updated', # fake_context, # servermodel_id=3, # servermodel_name='name3', # release_id=1, # servermodel_description='name3', # servermodel_parents_id=[1, 2]) # assert list(config_module.servermodel) == [1, 2, 3] # assert len(list(await config_module.servermodel[3].config.parents())) == 2 # # # @pytest.mark.asyncio # async def test_servermodel_updated1(): # config_module = dispatcher.get_service('config') # fake_context = get_fake_context('config') # config_module.cache_root_path = 'tests/data' # await config_module.on_join(fake_context) # # # assert list(config_module.servermodel) == [1, 2] # metaconfig1 = config_module.servermodel[1] # metaconfig2 = config_module.servermodel[2] # mixconfig1 = (await metaconfig1.config.list())[0] # mixconfig2 = (await metaconfig2.config.list())[0] # assert len(list(await metaconfig1.config.parents())) == 0 # assert len(list(await metaconfig2.config.parents())) == 1 # assert len(list(await mixconfig1.config.list())) == 1 # assert len(list(await mixconfig2.config.list())) == 0 # # # await dispatcher.publish('v1', # 'servermodel.updated', # fake_context, # servermodel_id=1, # servermodel_name='name1-1', # release_id=1, # servermodel_description='name1-1') # assert set(config_module.servermodel) == {1, 2} # assert config_module.servermodel[1].information.get('servermodel_name') == 'name1-1' # assert metaconfig1 != config_module.servermodel[1] # assert metaconfig2 == config_module.servermodel[2] # metaconfig1 = config_module.servermodel[1] # assert mixconfig1 != next(metaconfig1.config.list()) # mixconfig1 = next(metaconfig1.config.list()) # # # assert len(list(await metaconfig1.config.parents())) == 0 # assert len(list(await metaconfig2.config.parents())) == 1 # assert len(list(await mixconfig1.config.list())) == 1 # assert len(list(await mixconfig2.config.list())) == 0 # # # @pytest.mark.asyncio # async def test_servermodel_updated2(): # config_module = dispatcher.get_service('config') # fake_context = get_fake_context('config') # config_module.cache_root_path = 'tests/data' # await config_module.on_join(fake_context) # # create a new servermodel # assert list(config_module.servermodel) == [1, 2] # mixconfig1 = next(config_module.servermodel[1].config.list()) # mixconfig2 = next(config_module.servermodel[2].config.list()) # assert len(list(mixconfig1.config.list())) == 1 # assert len(list(mixconfig2.config.list())) == 0 # await dispatcher.publish('v1', # 'servermodel.created', # fake_context, # servermodel_id=3, # servermodel_name='name3', # release_id=1, # servermodel_description='name3', # servermodel_parents_id=[1]) # assert list(config_module.servermodel) == [1, 2, 3] # assert len(list(await config_module.servermodel[3].config.parents())) == 1 # assert await config_module.servermodel[3].information.get('servermodel_name') == 'name3' # assert len(list(await mixconfig1.config.list())) == 2 # assert len(list(await mixconfig2.config.list())) == 0 # # # await dispatcher.publish('v1', # 'servermodel.updated', # fake_context, # servermodel_id=3, # servermodel_name='name3-1', # release_id=1, # servermodel_description='name3-1', # servermodel_parents_id=[1, 2]) # assert list(config_module.servermodel) == [1, 2, 3] # assert config_module.servermodel[3].information.get('servermodel_name') == 'name3-1' # assert len(list(mixconfig1.config.list())) == 2 # assert len(list(mixconfig2.config.list())) == 1 # # # @pytest.mark.asyncio # async def test_servermodel_updated_config(): # config_module = dispatcher.get_service('config') # fake_context = get_fake_context('config') # config_module.cache_root_path = 'tests/data' # await config_module.on_join(fake_context) # # # config_module.servermodel[1].property.read_write() # assert config_module.servermodel[1].option('creole.general.mode_conteneur_actif').value.get() == 'non' # config_module.servermodel[1].option('creole.general.mode_conteneur_actif').value.set('oui') # assert config_module.servermodel[1].option('creole.general.mode_conteneur_actif').value.get() == 'oui' # # # await dispatcher.publish('v1', # 'servermodel.updated', # fake_context, # servermodel_id=1, # servermodel_name='name1-1', # release_id=1, # servermodel_description='name1-1') # assert config_module.servermodel[1].option('creole.general.mode_conteneur_actif').value.get() == 'oui' @pytest.mark.asyncio async def test_server_configuration_get(): config_module = dispatcher.get_service('config') fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) # await config_module.server[3]['server_to_deploy'].property.read_write() assert await config_module.server[3]['server_to_deploy'].option('creole.general.mode_conteneur_actif').value.get() == 'non' await config_module.server[3]['server_to_deploy'].option('creole.general.mode_conteneur_actif').value.set('oui') assert await config_module.server[3]['server_to_deploy'].option('creole.general.mode_conteneur_actif').value.get() == 'oui' assert await config_module.server[3]['server'].option('creole.general.mode_conteneur_actif').value.get() == 'non' # values = await dispatcher.call('v1', 'config.configuration.server.get', fake_context, server_id=3) configuration = {'configuration': {'creole.general.mode_conteneur_actif': 'non', 'creole.general.master.master': [], 'creole.general.master.slave1': [], 'creole.general.master.slave2': [], 'containers.container0.files.file0.mkdir': False, 'containers.container0.files.file0.name': '/etc/mailname', 'containers.container0.files.file0.rm': False, 'containers.container0.files.file0.source': 'mailname', 'containers.container0.files.file0.activate': True}, 'server_id': 3, 'deployed': True} assert values == configuration # values = await dispatcher.call('v1', 'config.configuration.server.get', fake_context, server_id=3, deployed=False) configuration['configuration']['creole.general.mode_conteneur_actif'] = 'oui' configuration['deployed'] = False assert values == configuration @pytest.mark.asyncio async def test_config_deployed(): config_module = dispatcher.get_service('config') fake_context = get_fake_context('config') config_module.cache_root_path = 'tests/data' await config_module.on_join(fake_context) # await config_module.server[3]['server_to_deploy'].property.read_write() assert await config_module.server[3]['server_to_deploy'].option('creole.general.mode_conteneur_actif').value.get() == 'non' await config_module.server[3]['server_to_deploy'].option('creole.general.mode_conteneur_actif').value.set('oui') assert await config_module.server[3]['server_to_deploy'].option('creole.general.mode_conteneur_actif').value.get() == 'oui' assert await config_module.server[3]['server'].option('creole.general.mode_conteneur_actif').value.get() == 'non' values = await dispatcher.publish('v1', 'config.configuration.server.deploy', fake_context, server_id=3) assert await config_module.server[3]['server_to_deploy'].option('creole.general.mode_conteneur_actif').value.get() == 'oui' assert await config_module.server[3]['server'].option('creole.general.mode_conteneur_actif').value.get() == 'oui'