519 lines
21 KiB
Python
519 lines
21 KiB
Python
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)
|