tiramisu/tests/test_dereference.py

# 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()