85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
# 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()
|