risotto/tests/test_session.py

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)