tiramisu/tests/config.py

86 lines
2.8 KiB
Python
Raw Normal View History

# from json import dumps, loads
2020-01-22 20:46:18 +01:00
import asyncio
from os import environ
try:
2019-08-03 21:03:23 +02:00
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
2019-12-24 15:24:20 +01:00
async def get_config(config, type, error=False):
if type == 'tiramisu':
return config
if error:
2019-12-24 15:24:20 +01:00
await config.property.add('demoting_error_warning')
return TestConfig(config)
def value_list(values):
if values[0] == '':
del values[0]
return tuple(values)
2019-12-24 15:24:20 +01:00
async def global_owner(config, config_type):
return await config.owner.get()
@pytest.fixture(params=PARAMS)
def config_type(request):
return request.param
2020-01-22 20:46:18 +01:00
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()