tiramisu/tests/test_config_api.py

"configuration objects global API"
import pytest

from .autopath import do_autopath
do_autopath()
from .config import config_type, get_config, value_list, global_owner, event_loop

from tiramisu import Config, IntOption, FloatOption, StrOption, ChoiceOption, \
    BoolOption, FilenameOption, SymLinkOption, IPOption, \
    PortOption, NetworkOption, NetmaskOption, BroadcastOption, \
    DomainnameOption, OptionDescription
from tiramisu.error import PropertiesOptionError, ValueWarning
from tiramisu.storage import list_sessions
import warnings


#def teardown_function(function):
#    # test_od_not_list emit a warnings because of doesn't create a Config
#    with warnings.catch_warnings(record=True) as w:
#        assert list_sessions() == [], 'session list is not empty when leaving "{}"'.format(function.__name__)


def make_description():
    gcoption = ChoiceOption('name', 'GC name', ('ref', 'framework'), 'ref')
    gcdummy = BoolOption('dummy', 'dummy', default=False)
    prop = BoolOption('prop', 'prop 1', properties=('disabled',))
    prop2 = StrOption('prop', 'prop 2', properties=('hidden',))
    objspaceoption = ChoiceOption('objspace', 'Object space',
                                  ('std', 'thunk'), 'std')
    booloption = BoolOption('bool', 'Test boolean option', default=True)
    booloption2 = BoolOption('bool', 'Test boolean option', default=False)
    intoption = IntOption('int', 'Test int option', default=0)
    floatoption2 = FloatOption('float', 'Test float option', default=2.3)
    floatoption = FloatOption('float', 'Test float option', default=2.3)
    stroption = StrOption('str', 'Test string option', default="abc")
    boolop = BoolOption('boolop', 'Test boolean option op', default=True)
    wantref_option = BoolOption('wantref', 'Tests', default=False)
    wantframework_option = BoolOption('wantframework', 'Test', default=False)
    gcgroup2 = OptionDescription('gc2', '', [booloption2, prop])
    gcgroup = OptionDescription('gc', '', [gcgroup2, gcoption, gcdummy, floatoption, prop2])
    descr = OptionDescription('tiramisu', '', [gcgroup, booloption, objspaceoption,
                                               wantref_option, stroption,
                                               wantframework_option,
                                               intoption, boolop, floatoption2])
    return descr


def _is_same_opt(opt1, opt2):
    if "id" in dir(opt1):
        assert opt1.id == opt2.id
    else:
        assert opt1 == opt2


@pytest.mark.asyncio
async def test_od_not_list():
    b = BoolOption('bool', '', multi=True)
    with pytest.raises(AssertionError):
        OptionDescription('od', '', b)
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_str():
    descr = make_description()
    async with await Config(descr) as cfg:
        cfg  # does not crash
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_make_dict(config_type):
    "serialization of the whole config to a dict"
    descr = OptionDescription("opt", "", [
        OptionDescription("s1", "", [
            BoolOption("a", "", default=False),
            BoolOption("b", "", default=False, properties=('hidden',))]),
        IntOption("int", "", default=42)])
    async with await Config(descr) as cfg:
        await cfg.property.read_write()
        await cfg.permissive.add('hidden')
        cfg = await get_config(cfg, config_type)
        d = await cfg.value.dict()
        assert d == {"s1.a": False, "int": 42}
        await cfg.option('int').value.set(43)
        await cfg.option('s1.a').value.set(True)
        d = await cfg.value.dict()
        assert d == {"s1.a": True, "int": 43}
        d2 = await cfg.value.dict(flatten=True)
        assert d2 == {'a': True, 'int': 43}
        if config_type == 'tiramisu':
            assert await cfg.forcepermissive.value.dict() == {"s1.a": True, "s1.b": False, "int": 43}
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_make_dict_with_disabled(config_type):
    descr = OptionDescription("opt", "", [
        OptionDescription("s1", "", [
            BoolOption("a", "", default=False),
            BoolOption("b", "", default=False, properties=('disabled',))]),
        OptionDescription("s2", "", [
            BoolOption("a", "", default=False),
            BoolOption("b", "", default=False)], properties=('disabled',)),
        IntOption("int", "", default=42)])
    async with await Config(descr) as cfg:
        await cfg.property.read_only()
        cfg = await get_config(cfg, config_type)
        assert await cfg.value.dict() == {"s1.a": False, "int": 42}
        if config_type == 'tiramisu':
            assert await cfg.forcepermissive.value.dict() == {"s1.a": False, "int": 42}
            assert await cfg.unrestraint.value.dict() == {"int": 42, "s1.a": False, "s1.b": False, "s2.a": False, "s2.b": False}
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_make_dict_with_disabled_in_callback(config_type):
    descr = OptionDescription("opt", "", [
        OptionDescription("s1", "", [
            BoolOption("a", "", default=False),
            BoolOption("b", "", default=False, properties=('disabled',))]),
        OptionDescription("s2", "", [
            BoolOption("a", "", default=False),
            BoolOption("b", "", default=False)], properties=('disabled',)),
        IntOption("int", "", default=42)])
    async with await Config(descr) as cfg:
        await cfg.property.read_only()
        cfg = await get_config(cfg, config_type)
        d = await cfg.value.dict()
        assert d == {"s1.a": False, "int": 42}
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_make_dict_fullpath(config_type):
    descr = OptionDescription("root", "", [
        OptionDescription("opt", "", [
            OptionDescription("s1", "", [
                BoolOption("a", "", default=False),
                BoolOption("b", "", default=False, properties=('disabled',))]),
            OptionDescription("s2", "", [
                BoolOption("a", "", default=False),
                BoolOption("b", "", default=False)], properties=('disabled',)),
            IntOption("int", "", default=42)]),
        IntOption("introot", "", default=42)])
    async with await Config(descr) as cfg:
        await cfg.property.read_only()
        cfg = await get_config(cfg, config_type)
        assert await cfg.value.dict() == {"opt.s1.a": False, "opt.int": 42, "introot": 42}
        if config_type == 'tiramisu':
            # FIXME
            assert await cfg.option('opt').value.dict() == {"s1.a": False, "int": 42}
        assert await cfg.value.dict(fullpath=True) == {"opt.s1.a": False, "opt.int": 42, "introot": 42}
        if config_type == 'tiramisu':
            # FIXME
            assert await cfg.option('opt').value.dict(fullpath=True) == {"opt.s1.a": False, "opt.int": 42}
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_find_in_config():
    "finds option in config"
    descr = make_description()
    async with await Config(descr) as cfg:
        await cfg.property.read_only()
        await cfg.permissive.add('hidden')
        ret = list(await cfg.option.find('dummy'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.dummy').option.get())
        #
        ret_find = await cfg.option.find('dummy', first=True)
        ret = await ret_find.option.get()
        _is_same_opt(ret, await cfg.option('gc.dummy').option.get())
        #
        ret = list(await cfg.option.find('float'))
        assert len(ret) == 2
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.float').option.get())
        _is_same_opt(await ret[1].option.get(), await cfg.option('float').option.get())
        #
        ret = await cfg.option.find('bool', first=True)
        _is_same_opt(await ret.option.get(), await cfg.option('gc.gc2.bool').option.get())
        ret = await cfg.option.find('bool', value=True, first=True)
        _is_same_opt(await ret.option.get(), await cfg.option('bool').option.get())
        ret = await cfg.option.find('dummy', first=True)
        _is_same_opt(await ret.option.get(), await cfg.option('gc.dummy').option.get())
        ret = await cfg.option.find('float', first=True)
        _is_same_opt(await ret.option.get(), await cfg.option('gc.float').option.get())
        #FIXME cannot find an option without name
        #ret = await cfg.find(bytype=ChoiceOption)
        #assert len(ret) == 2
        #_is_same_opt(ret[0], await cfg.unwrap_from_path('gc.name'))
        #_is_same_opt(ret[1], await cfg.unwrap_from_path('objspace'))
        #
        #_is_same_opt(await cfg.find_first(bytype=ChoiceOption), await cfg.unwrap_from_path('gc.name'))
        #ret = await cfg.find(byvalue='ref')
        #assert len(ret) == 1
        #_is_same_opt(ret[0], await cfg.unwrap_from_path('gc.name'))
        #_is_same_opt(await cfg.find_first(byvalue='ref'), await cfg.unwrap_from_path('gc.name'))
        #
        ret = list(await cfg.option.find('prop'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.prop').option.get())
        #
        ret = list(await cfg.option.find('prop', value=None))
        assert len(ret) == 1
        ret = list(await cfg.option.find('prop'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.prop').option.get())
        #
        await cfg.property.read_write()
        with pytest.raises(AttributeError):
            ret = await cfg.option.find('prop')
            assert await ret.option.get()
        ret = list(await cfg.unrestraint.option.find(name='prop'))
        assert len(ret) == 2
        _is_same_opt(await ret[0].option.get(), await cfg.unrestraint.option('gc.gc2.prop').option.get())
        _is_same_opt(await ret[1].option.get(), await cfg.forcepermissive.option('gc.prop').option.get())
        #
        ret = list(await cfg.forcepermissive.option.find('prop'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.forcepermissive.option('gc.prop').option.get())
        #
        ret = await cfg.forcepermissive.option.find('prop', first=True)
        _is_same_opt(await ret.option.get(), await cfg.forcepermissive.option('gc.prop').option.get())
        # combinaison of filters
        ret = list(await cfg.unrestraint.option.find('prop', type=BoolOption))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.unrestraint.option('gc.gc2.prop').option.get())
        ret = await cfg.unrestraint.option.find('prop', type=BoolOption, first=True)
        _is_same_opt(await ret.option.get(), await cfg.unrestraint.option('gc.gc2.prop').option.get())
        #
        ret = list(await cfg.option.find('dummy', value=False))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.dummy').option.get())
        #
        ret = await cfg.option.find('dummy', value=False, first=True)
        _is_same_opt(await ret.option.get(), await cfg.option('gc.dummy').option.get())
        #subcfgig
        ret = list(await cfg.option('gc').find('dummy'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.dummy').option.get())
        #
        ret = list(await cfg.option('gc').find('float'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.float').option.get())
        #
        ret = list(await cfg.option('gc').find('bool'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.gc2.bool').option.get())
        ret = await cfg.option('gc').find('bool', value=False, first=True)
        _is_same_opt(await ret.option.get(), await cfg.option('gc.gc2.bool').option.get())
        #
        with pytest.raises(AttributeError):
            ret = await cfg.option('gc').find('bool', value=True, first=True)
            assert await ret.option.get()
        #
        with pytest.raises(AttributeError):
            ret = await cfg.option('gc').find('wantref')
            await ret.option.get()
        #
        ret = list(await cfg.unrestraint.option('gc').find('prop'))
        assert len(ret) == 2
        _is_same_opt(await ret[0].option.get(), await cfg.unrestraint.option('gc.gc2.prop').option.get())
        _is_same_opt(await ret[1].option.get(), await cfg.forcepermissive.option('gc.prop').option.get())
        #
        await cfg.property.read_only()
        ret = list(await cfg.option('gc').find('prop'))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), await cfg.option('gc.prop').option.get())
        # not OptionDescription
        with pytest.raises(AttributeError):
            await cfg.option.find('gc', first=True)
        with pytest.raises(AttributeError):
            await cfg.option.find('gc2', first=True)
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_find_multi():
    b = BoolOption('bool', '', multi=True, properties=('notunique',))
    o = OptionDescription('od', '', [b])
    async with await Config(o) as cfg:
        #
        with pytest.raises(AttributeError):
            list(await cfg.option.find('bool', value=True))
        with pytest.raises(AttributeError):
            list(await cfg.option.find('bool', value=True, first=True))
        await cfg.option('bool').value.set([False])
        with pytest.raises(AttributeError):
            list(await cfg.option.find('bool', value=True))
        with pytest.raises(AttributeError):
            list(await cfg.option.find('bool', value=True, first=True))
        await cfg.option('bool').value.set([False, False])
        with pytest.raises(AttributeError):
            list(await cfg.option.find('bool', value=True))
        with pytest.raises(AttributeError):
            list(await cfg.option.find('bool', value=True, first=True))
        await cfg.option('bool').value.set([False, False, True])
        ret = list(await cfg.option.find('bool', value=True))
        assert len(ret) == 1
        _is_same_opt(await ret[0].option.get(), b)
        ret = await cfg.option.find('bool', value=True, first=True)
        _is_same_opt(await ret.option.get(), b)
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_does_not_find_in_config():
    descr = make_description()
    async with await Config(descr) as cfg:
        with pytest.raises(AttributeError):
            list(await cfg.option.find('IDontExist'))
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_filename(config_type):
    a = FilenameOption('a', '')
    o = OptionDescription('o', '', [a])
    async with await Config(o) as cfg:
        # FIXME cfg = await get_config(cfg, config_type)
        await cfg.option('a').value.set('/')
        await cfg.option('a').value.set('/tmp')
        await cfg.option('a').value.set('/tmp/')
        await cfg.option('a').value.set('/tmp/text.txt')
        await cfg.option('a').value.set('/tmp/with space.txt')
        await cfg.option('a').value.set('/tmp/with$.txt')
        with pytest.raises(ValueError):
            await cfg.option('a').value.set('not starts with /')
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_invalid_option():
    ChoiceOption('a', '', ('1', '2'))
    with pytest.raises(TypeError):
        ChoiceOption('a', '', [1, 2])
    with pytest.raises(TypeError):
        ChoiceOption('a', '', 1)
    with pytest.raises(ValueError):
        ChoiceOption('a', '', (1,), 3)
    FloatOption('a', '')
    with pytest.raises(ValueError):
        FloatOption('a', '', 'string')
    StrOption('a', '')
    with pytest.raises(ValueError):
        StrOption('a', '', 1)
    u = StrOption('a', '')
    SymLinkOption('a', u)
    with pytest.raises(ValueError):
        SymLinkOption('a', 'string')
    IPOption('a', '')
    with pytest.raises(ValueError):
        IPOption('a', '', 1)
    with pytest.raises(ValueError):
        IPOption('a', '', 'string')
    PortOption('a', '')
    with pytest.raises(ValueError):
        PortOption('a', '', 'string')
    with pytest.raises(ValueError):
        PortOption('a', '', '11:12:13', allow_range=True)
    with pytest.raises(ValueError):
        PortOption('a', '', 11111111111111111111)
    with pytest.raises(ValueError):
        PortOption('a', '', allow_zero=True, allow_wellknown=False, allow_registred=True, allow_private=False)
    with pytest.raises(ValueError):
        PortOption('a', '', allow_zero=True, allow_wellknown=True, allow_registred=False, allow_private=True)
    with pytest.raises(ValueError):
        PortOption('a', '', allow_zero=True, allow_wellknown=False, allow_registred=False, allow_private=True)
    with pytest.raises(ValueError):
        PortOption('a', '', allow_zero=True, allow_wellknown=False, allow_registred=True, allow_private=True)
    with pytest.raises(ValueError):
        PortOption('a', '', allow_zero=False, allow_wellknown=False, allow_registred=False, allow_private=False)
    NetworkOption('a', '')
    with pytest.raises(ValueError):
        NetworkOption('a', '', 'string')
    NetmaskOption('a', '')
    with pytest.raises(ValueError):
        NetmaskOption('a', '', 'string')
    BroadcastOption('a', '')
    with pytest.raises(ValueError):
        BroadcastOption('a', '', 'string')
    DomainnameOption('a', '')
    with pytest.raises(ValueError):
        DomainnameOption('a', '', 'string')
    with pytest.raises(ValueError):
        DomainnameOption('a', '', type='string')
    with pytest.raises(ValueError):
        DomainnameOption('a', '', allow_ip='string')
    with pytest.raises(ValueError):
        DomainnameOption('a', '', allow_without_dot='string')
    with pytest.raises(ValueError):
        DomainnameOption('a', '', 1)
    #
    ChoiceOption('a', '', (1,), multi=True, default_multi=1)
    with pytest.raises(ValueError):
        ChoiceOption('a', '', (1,), default_multi=1)
    with pytest.raises(ValueError):
        ChoiceOption('a', '', (1,), multi=True, default=[1,], default_multi=2)
    with pytest.raises(ValueError):
        FloatOption('a', '', multi=True, default_multi='string')
    with pytest.raises(ValueError):
        StrOption('a', '', multi=True, default_multi=1)
    with pytest.raises(ValueError):
        IPOption('a', '', multi=True, default_multi=1)
    with pytest.raises(ValueError):
        IPOption('a', '', multi=True, default_multi='string')
    with pytest.raises(ValueError):
        PortOption('a', '', multi=True, default_multi='string')
    with pytest.raises(ValueError):
        PortOption('a', '', multi=True, default_multi='11:12:13', allow_range=True)
    with pytest.raises(ValueError):
        PortOption('a', '', multi=True, default_multi=11111111111111111111)
    with pytest.raises(ValueError):
        NetworkOption('a', '', multi=True, default_multi='string')
    with pytest.raises(ValueError):
        NetmaskOption('a', '', multi=True, default_multi='string')
    with pytest.raises(ValueError):
        BroadcastOption('a', '', multi=True, default_multi='string')
    with pytest.raises(ValueError):
        DomainnameOption('a', '', multi=True, default_multi='string')
    with pytest.raises(ValueError):
        DomainnameOption('a', '', multi=True, default_multi=1)
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_help():
    stro = StrOption('s', '', multi=True)
    od1 = OptionDescription('o', '', [stro])
    od2 = OptionDescription('o', '', [od1])
    async with await Config(od2) as cfg:
        cfg.help(_display=False)
        cfg.config.help(_display=False)
        cfg.option.help(_display=False)
        cfg.option('o').help(_display=False)
        cfg.option('o.s').help(_display=False)
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_config_reset():
    descr = make_description()
    async with await Config(descr) as cfg:
        await cfg.owner.set('test')
        assert await cfg.owner.get() == 'test'
        assert not await cfg.option('gc.gc2.bool').value.get()
        assert not await cfg.option('boolop').property.get()
        assert not await cfg.option('boolop').permissive.get()
        assert not await cfg.option('wantref').information.get('info', None)
        #
        await cfg.option('gc.gc2.bool').value.set(True)
        await cfg.option('boolop').property.add('test')
        await cfg.option('float').permissive.set(frozenset(['test']))
        await cfg.option('wantref').information.set('info', 'info')
        assert await cfg.option('gc.gc2.bool').value.get()
        assert await cfg.option('boolop').property.get()
        assert await cfg.option('float').permissive.get()
        assert await cfg.option('wantref').information.get('info', None)
        #
        assert await cfg.owner.get() == 'test'
        await cfg.config.reset()
        assert await cfg.owner.get() == 'test'
        assert not await cfg.option('gc.gc2.bool').value.get()
        assert not await cfg.option('boolop').property.get()
        assert not await cfg.option('float').permissive.get()
        assert not await cfg.option('wantref').information.get('info', None)
    assert not await list_sessions()