risotto/tests/test_session.py

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)