291 lines
7.4 KiB
Python
291 lines
7.4 KiB
Python
# coding: utf-8
|
|
from .autopath import do_autopath
|
|
do_autopath()
|
|
import weakref
|
|
import pytest
|
|
|
|
from tiramisu import BoolOption, IntOption, StrOption, IPOption, NetmaskOption, \
|
|
SymLinkOption, OptionDescription, DynOptionDescription, submulti, \
|
|
Config, GroupConfig, MetaConfig, Params, ParamOption, Calculation
|
|
from tiramisu.storage import list_sessions
|
|
from .config import event_loop
|
|
|
|
|
|
IS_DEREFABLE = True
|
|
|
|
|
|
def funcname(*args, **kwargs):
|
|
return value
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_storage():
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
async with await Config(o) as cfg:
|
|
w = weakref.ref(cfg._config_bag.context.cfgimpl_get_values()._p_)
|
|
del cfg
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_value():
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
async with await Config(o) as cfg:
|
|
w = weakref.ref(cfg._config_bag.context.cfgimpl_get_values())
|
|
del cfg
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_setting():
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
async with await Config(o) as cfg:
|
|
w = weakref.ref(cfg._config_bag.context.cfgimpl_get_settings())
|
|
del cfg
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_config():
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
async with await Config(o) as cfg:
|
|
w = weakref.ref(cfg)
|
|
del cfg
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_option():
|
|
global IS_DEREFABLE
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
w = weakref.ref(b)
|
|
del(b)
|
|
try:
|
|
assert w() is not None
|
|
except AssertionError:
|
|
IS_DEREFABLE = False
|
|
return
|
|
del(o)
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_optiondescription():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
w = weakref.ref(o)
|
|
del(b)
|
|
assert w() is not None
|
|
del(o)
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_option_cache():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
await o._build_cache()
|
|
w = weakref.ref(b)
|
|
del(b)
|
|
assert w() is not None
|
|
del(o)
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_optiondescription_cache():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
await o._build_cache()
|
|
w = weakref.ref(o)
|
|
del(b)
|
|
assert w() is not None
|
|
del(o)
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_option_config():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
async with await Config(o) as cfg:
|
|
w = weakref.ref(b)
|
|
del(b)
|
|
assert w() is not None
|
|
del(o)
|
|
assert w() is not None
|
|
del cfg
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_optiondescription_config():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
b = BoolOption('b', '')
|
|
o = OptionDescription('od', '', [b])
|
|
async with await Config(o) as cfg:
|
|
w = weakref.ref(o)
|
|
del(b)
|
|
assert w() is not None
|
|
del(o)
|
|
assert w() is not None
|
|
del cfg
|
|
assert w() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_validator():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
a = StrOption('a', '', default='yes')
|
|
b = StrOption('b', '', validators=[Calculation(funcname, Params(ParamOption(a)))], default='val')
|
|
od = OptionDescription('root', '', [a, b])
|
|
async with await Config(od) as cfg:
|
|
w = weakref.ref(a)
|
|
x = weakref.ref(b)
|
|
y = weakref.ref(od)
|
|
z = weakref.ref(cfg)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(a)
|
|
del(b)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(od)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del cfg
|
|
assert y() is None
|
|
assert z() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_callback():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
a = StrOption('a', "", 'val')
|
|
b = StrOption('b', "", Calculation(funcname, Params((ParamOption(a),))))
|
|
od = OptionDescription('root', '', [a, b])
|
|
async with await Config(od) as cfg:
|
|
w = weakref.ref(a)
|
|
x = weakref.ref(b)
|
|
y = weakref.ref(od)
|
|
z = weakref.ref(cfg)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(a)
|
|
del(b)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(od)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del cfg
|
|
assert y() is None
|
|
assert z() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_symlink():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
a = BoolOption("a", "", default=False)
|
|
b = SymLinkOption("b", a)
|
|
od = OptionDescription('root', '', [a, b])
|
|
async with await Config(od) as cfg:
|
|
w = weakref.ref(a)
|
|
x = weakref.ref(b)
|
|
y = weakref.ref(od)
|
|
z = weakref.ref(cfg)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(a)
|
|
del(b)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(od)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del cfg
|
|
assert y() is None
|
|
assert z() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deref_dyn():
|
|
if not IS_DEREFABLE:
|
|
return
|
|
a = StrOption('a', '', ['val1', 'val2'], multi=True)
|
|
b = StrOption('b', '')
|
|
dod = DynOptionDescription('dod', '', [b], suffixes=Calculation(funcname, Params((ParamOption(a),))))
|
|
od = OptionDescription('od', '', [dod, a])
|
|
async with await Config(od) as cfg:
|
|
w = weakref.ref(a)
|
|
x = weakref.ref(b)
|
|
y = weakref.ref(od)
|
|
z = weakref.ref(cfg)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(a)
|
|
del(b)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del(od)
|
|
del(dod)
|
|
assert w() is not None
|
|
assert x() is not None
|
|
assert w() is not None
|
|
assert x() is not None
|
|
del cfg
|
|
assert y() is None
|
|
assert z() is None
|
|
assert not await list_sessions()
|