from importlib import import_module import pytest from .storage import STORAGE from risotto.context import Context #from risotto.services import load_services from risotto.dispatcher import dispatcher from risotto.services.session.storage import storage_server, storage_servermodel 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 def setup_module(module): #load_services(['config', 'session'], # validate=False, # test=True) 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) dispatcher.injected_self['servermodel'].cache_root_path = 'tests/data' def teardown_function(function): config_module = dispatcher.get_service('session') storage_server.sessions = {} storage_servermodel.sessions = {} @pytest.mark.asyncio async def test_server_start(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) await dispatcher.call('v1', 'session.server.start', fake_context, id=3) @pytest.mark.asyncio async def test_server_list(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) assert not await dispatcher.call('v1', 'session.server.list', fake_context) await dispatcher.call('v1', 'session.server.start', fake_context, id=3) assert await dispatcher.call('v1', 'session.server.list', fake_context) @pytest.mark.asyncio async def test_server_filter_namespace(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = session['session_id'] namespace = 'containers' await dispatcher.call('v1', 'session.server.filter', fake_context, session_id=session_id, namespace=namespace) list_result = await dispatcher.call('v1', 'session.server.list', fake_context) assert list_result[0]['namespace'] == namespace @pytest.mark.asyncio async def test_server_filter_mode(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = session['session_id'] assert session['mode'] == 'normal' mode = 'expert' await dispatcher.call('v1', 'session.server.filter', fake_context, session_id=session_id, mode=mode) list_result = await dispatcher.call('v1', 'session.server.list', fake_context) assert list_result[0]['mode'] == mode @pytest.mark.asyncio async def test_server_filter_debug(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = session['session_id'] assert session['debug'] == False debug = True await dispatcher.call('v1', 'session.server.filter', fake_context, session_id=session_id, debug=debug) list_result = await dispatcher.call('v1', 'session.server.list', fake_context) assert list_result[0]['debug'] == debug #FIXME @pytest.mark.asyncio async def test_server_filter_get(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = session['session_id'] values = await dispatcher.call('v1', 'session.server.get', fake_context, session_id=session_id) assert values == {'content': {"creole.general.mode_conteneur_actif": "non", "creole.general.master.master": []}, 'debug': False, 'id': 3, 'mode': 'normal', 'namespace': 'creole', 'session_id': session_id, 'timestamp': values['timestamp'], 'username': 'test'} @pytest.mark.asyncio async def test_server_filter_get_one_value(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = session['session_id'] values = await dispatcher.call('v1', 'session.server.get', fake_context, session_id=session_id, name="creole.general.mode_conteneur_actif") assert values == {'content': {"creole.general.mode_conteneur_actif": "non"}, 'debug': False, 'id': 3, 'mode': 'normal', 'namespace': 'creole', 'session_id': session_id, 'timestamp': values['timestamp'], 'username': 'test'} @pytest.mark.asyncio async def test_server_filter_configure(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = session['session_id'] await dispatcher.call('v1', 'session.server.configure', fake_context, session_id=session_id, action='modify', name='creole.general.mode_conteneur_actif', value='oui') list_result = await dispatcher.call('v1', 'session.server.list', fake_context) values = await dispatcher.call('v1', 'session.server.get', fake_context, session_id=session_id, name="creole.general.mode_conteneur_actif") assert values == {'content': {"creole.general.mode_conteneur_actif": "oui"}, 'debug': False, 'id': 3, 'mode': 'normal', 'namespace': 'creole', 'session_id': session_id, 'timestamp': values['timestamp'], 'username': 'test'} @pytest.mark.asyncio async def test_server_filter_validate(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = session['session_id'] await dispatcher.call('v1', 'session.server.validate', fake_context, session_id=session_id) @pytest.mark.asyncio async def test_server_stop(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.server: await config_module.on_join(fake_context) assert not await dispatcher.call('v1', 'session.server.list', fake_context) start = await dispatcher.call('v1', 'session.server.start', fake_context, id=3) session_id = start['session_id'] assert await dispatcher.call('v1', 'session.server.list', fake_context) await dispatcher.call('v1', 'session.server.stop', fake_context, session_id=session_id) assert not await dispatcher.call('v1', 'session.server.list', fake_context) # servermodel @pytest.mark.asyncio async def test_servermodel_start(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) @pytest.mark.asyncio async def test_servermodel_list(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) assert not await dispatcher.call('v1', 'session.servermodel.list', fake_context) await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) assert await dispatcher.call('v1', 'session.servermodel.list', fake_context) @pytest.mark.asyncio async def test_servermodel_filter_namespace(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = session['session_id'] namespace = 'containers' await dispatcher.call('v1', 'session.servermodel.filter', fake_context, session_id=session_id, namespace=namespace) list_result = await dispatcher.call('v1', 'session.servermodel.list', fake_context) assert list_result[0]['namespace'] == namespace @pytest.mark.asyncio async def test_servermodel_filter_mode(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = session['session_id'] assert session['mode'] == 'normal' mode = 'expert' await dispatcher.call('v1', 'session.servermodel.filter', fake_context, session_id=session_id, mode=mode) list_result = await dispatcher.call('v1', 'session.servermodel.list', fake_context) assert list_result[0]['mode'] == mode @pytest.mark.asyncio async def test_servermodel_filter_debug(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = session['session_id'] assert session['debug'] == False debug = True await dispatcher.call('v1', 'session.servermodel.filter', fake_context, session_id=session_id, debug=debug) list_result = await dispatcher.call('v1', 'session.servermodel.list', fake_context) assert list_result[0]['debug'] == debug @pytest.mark.asyncio async def test_servermodel_filter_get(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = session['session_id'] values = await dispatcher.call('v1', 'session.servermodel.get', fake_context, session_id=session_id) assert values == {'content': {"creole.general.mode_conteneur_actif": "non", "creole.general.master.master": []}, 'debug': False, 'id': 1, 'mode': 'normal', 'namespace': 'creole', 'session_id': session_id, 'timestamp': values['timestamp'], 'username': 'test'} @pytest.mark.asyncio async def test_servermodel_filter_get_one_value(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = session['session_id'] values = await dispatcher.call('v1', 'session.servermodel.get', fake_context, session_id=session_id, name="creole.general.mode_conteneur_actif") assert values == {'content': {"creole.general.mode_conteneur_actif": "non"}, 'debug': False, 'id': 1, 'mode': 'normal', 'namespace': 'creole', 'session_id': session_id, 'timestamp': values['timestamp'], 'username': 'test'} @pytest.mark.asyncio async def test_servermodel_filter_configure(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = session['session_id'] await dispatcher.call('v1', 'session.servermodel.configure', fake_context, session_id=session_id, action='modify', name='creole.general.mode_conteneur_actif', value='oui') list_result = await dispatcher.call('v1', 'session.servermodel.list', fake_context) values = await dispatcher.call('v1', 'session.servermodel.get', fake_context, session_id=session_id, name="creole.general.mode_conteneur_actif") assert values == {'content': {"creole.general.mode_conteneur_actif": "oui"}, 'debug': False, 'id': 1, 'mode': 'normal', 'namespace': 'creole', 'session_id': session_id, 'timestamp': values['timestamp'], 'username': 'test'} @pytest.mark.asyncio async def test_servermodel_filter_validate(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) session = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = session['session_id'] await dispatcher.call('v1', 'session.servermodel.validate', fake_context, session_id=session_id) @pytest.mark.asyncio async def test_servermodel_stop(): fake_context = get_fake_context('session') config_module = dispatcher.get_service('config') config_module.cache_root_path = 'tests/data' if not config_module.servermodel: await config_module.on_join(fake_context) assert not await dispatcher.call('v1', 'session.servermodel.list', fake_context) start = await dispatcher.call('v1', 'session.servermodel.start', fake_context, id=1) session_id = start['session_id'] assert await dispatcher.call('v1', 'session.servermodel.list', fake_context) await dispatcher.call('v1', 'session.servermodel.stop', fake_context, session_id=session_id) assert not await dispatcher.call('v1', 'session.servermodel.list', fake_context)