tiramisu/tests/config.py

# from json import dumps, loads
import asyncio
from os import environ
try:
    from tiramisu_api import Config
    class TestConfig(Config):
        def __init__(self,
                     config):
            self.test_option = config.option
            json = self.test_option.dict()
            # assert json == loads(dumps(json))
            super().__init__(json)

        def send_data(self,
                      updates):
            return self.updates_data(self.test_option.updates(updates))
    PARAMS = ['tiramisu', 'tiramisu-api']
except:
    PARAMS = ['tiramisu']

import pytest


async def get_config(config, type, error=False):
    if type == 'tiramisu':
        return config
    if error:
        await config.property.add('demoting_error_warning')
    return TestConfig(config)


def value_list(values):
    if values[0] == '':
        del values[0]
    return tuple(values)


async def global_owner(config, config_type):
    return await config.owner.get()


@pytest.fixture(params=PARAMS)
def config_type(request):
    return request.param


LOOP = None
@pytest.fixture(scope='session')
def event_loop(request):
    """Create an instance of the default event loop for each test case."""
    global LOOP
    if LOOP is None:
        LOOP = asyncio.get_event_loop_policy().new_event_loop()
    return LOOP


async def _delete_sessions(meta):
    if await meta.config.type() != 'config':
        for conf in await meta.config.list():
            await _delete_sessions(conf)
    await meta.session.reset()


async def delete_sessions(confs):
    if not isinstance(confs, list):
        confs = [confs]
    for conf in confs:
        await _delete_sessions(conf)
    if environ.get('TIRAMISU_STORAGE') == 'postgres':
        async with confs[0]._config_bag.context.getconnection() as connection:
            assert await connection.fetchrow('SELECT * FROM session') is None
            assert await connection.fetchrow('SELECT * FROM value') is None
            assert await connection.fetchrow('SELECT * FROM information') is None
            assert await connection.fetchrow('SELECT * FROM property') is None
            assert await connection.fetchrow('SELECT * FROM permissive') is None
    elif environ.get('TIRAMISU_STORAGE') == 'sqlite3':
        async with confs[0]._config_bag.context.getconnection() as connection:
            assert  await connection.select('SELECT * FROM session') is None
            assert  await connection.select('SELECT * FROM value') is None
            assert  await connection.select('SELECT * FROM information') is None
            assert  await connection.select('SELECT * FROM property') is None
            assert  await connection.select('SELECT * FROM permissive') is None
    else:
        from tiramisu import list_sessions
        assert not await list_sessions()