salt/tests/unit/utils/test_pyobjects.py

# -*- coding: utf-8 -*-

# Import Pytohn libs
from __future__ import absolute_import
import jinja2
import logging
import os
import shutil
import tempfile
import textwrap
import uuid

# Import Salt Testing libs
from tests.support.unit import TestCase

# Import Salt libs
import tests.integration as integration
import salt.config
import salt.state
import salt.utils.files
from salt.template import compile_template
from salt.utils.odict import OrderedDict
from salt.utils.pyobjects import (StateFactory, State, Registry,
                                  SaltObject, InvalidFunction, DuplicateState)

log = logging.getLogger(__name__)

File = StateFactory('file')
Service = StateFactory('service')

pydmesg_expected = {
    'file.managed': [
        {'group': 'root'},
        {'mode': '0755'},
        {'require': [{'file': '/usr/local/bin'}]},
        {'source': 'salt://debian/files/pydmesg.py'},
        {'user': 'root'},
    ]
}
pydmesg_salt_expected = OrderedDict([
    ('/usr/local/bin/pydmesg', pydmesg_expected)
])
pydmesg_kwargs = dict(user='root', group='root', mode='0755',
                      source='salt://debian/files/pydmesg.py')

basic_template = '''#!pyobjects
File.directory('/tmp', mode='1777', owner='root', group='root')
'''

invalid_template = '''#!pyobjects
File.fail('/tmp')
'''

include_template = '''#!pyobjects
include('http')
'''

extend_template = '''#!pyobjects
include('http')

from salt.utils.pyobjects import StateFactory
Service = StateFactory('service')

Service.running(extend('apache'), watch=[{'file': '/etc/file'}])
'''

map_prefix = '''\
#!pyobjects
from salt.utils.pyobjects import StateFactory
Service = StateFactory('service')

{% macro priority(value) %}
    priority = {{ value }}
{% endmacro %}
class Samba(Map):
'''

map_suffix = '''
with Pkg.installed("samba", names=[Samba.server, Samba.client]):
    Service.running("samba", name=Samba.service)
'''

map_data = {
    'debian': "    class Debian:\n"
              "        server = 'samba'\n"
              "        client = 'samba-client'\n"
              "        service = 'samba'\n",
    'centos': "    class RougeChapeau:\n"
              "        __match__ = 'RedHat'\n"
              "        server = 'samba'\n"
              "        client = 'samba'\n"
              "        service = 'smb'\n",
    'ubuntu': "    class Ubuntu:\n"
              "        __grain__ = 'os'\n"
              "        service = 'smbd'\n"
}

import_template = '''#!pyobjects
import salt://map.sls

Pkg.removed("samba-imported", names=[map.Samba.server, map.Samba.client])
'''

recursive_map_template = '''#!pyobjects
from salt://map.sls import Samba

class CustomSamba(Samba):
    pass
'''

recursive_import_template = '''#!pyobjects
from salt://recursive_map.sls import CustomSamba

Pkg.removed("samba-imported", names=[CustomSamba.server, CustomSamba.client])'''

scope_test_import_template = '''#!pyobjects
from salt://recursive_map.sls import CustomSamba

# since we import CustomSamba we should shouldn't be able to see Samba
Pkg.removed("samba-imported", names=[Samba.server, Samba.client])'''

from_import_template = '''#!pyobjects
# this spacing is like this on purpose to ensure it's stripped properly
from   salt://map.sls  import     Samba

Pkg.removed("samba-imported", names=[Samba.server, Samba.client])
'''

import_as_template = '''#!pyobjects
from salt://map.sls import Samba as Other
Pkg.removed("samba-imported", names=[Other.server, Other.client])
'''

random_password_template = '''#!pyobjects
import random, string
password = ''.join([random.SystemRandom().choice(
        string.ascii_letters + string.digits) for _ in range(20)])
'''

random_password_import_template = '''#!pyobjects
from salt://password.sls import password
'''

requisite_implicit_list_template = '''#!pyobjects
from salt.utils.pyobjects import StateFactory
Service = StateFactory('service')

with Pkg.installed("pkg"):
    Service.running("service", watch=File("file"), require=Cmd("cmd"))
'''


class MapBuilder(object):
    def build_map(self, template=None):
        '''
        Build from a specific template or just use a default if no template
        is passed to this function.
        '''
        if template is None:
            template = textwrap.dedent('''\
                {{ ubuntu }}
                {{ centos }}
                {{ debian }}
                ''')
        full_template = map_prefix + template + map_suffix
        ret = jinja2.Template(full_template).render(**map_data)
        log.debug('built map: \n%s', ret)
        return ret


class StateTests(TestCase):
    def setUp(self):
        Registry.empty()

    def test_serialization(self):
        f = State('/usr/local/bin/pydmesg', 'file', 'managed',
                  require=File('/usr/local/bin'),
                  **pydmesg_kwargs)

        self.assertEqual(f(), pydmesg_expected)

    def test_factory_serialization(self):
        File.managed('/usr/local/bin/pydmesg',
                     require=File('/usr/local/bin'),
                     **pydmesg_kwargs)

        self.assertEqual(
            Registry.states['/usr/local/bin/pydmesg'],
            pydmesg_expected
        )

    def test_context_manager(self):
        with File('/usr/local/bin'):
            pydmesg = File.managed('/usr/local/bin/pydmesg', **pydmesg_kwargs)

            self.assertEqual(
                Registry.states['/usr/local/bin/pydmesg'],
                pydmesg_expected
            )

            with pydmesg:
                File.managed('/tmp/something', owner='root')

                self.assertEqual(
                    Registry.states['/tmp/something'],
                    {
                        'file.managed': [
                            {'owner': 'root'},
                            {'require': [
                                {'file': '/usr/local/bin'},
                                {'file': '/usr/local/bin/pydmesg'}
                            ]},
                        ]
                    }
                )

    def test_salt_data(self):
        File.managed('/usr/local/bin/pydmesg',
                     require=File('/usr/local/bin'),
                     **pydmesg_kwargs)

        self.assertEqual(
            Registry.states['/usr/local/bin/pydmesg'],
            pydmesg_expected
        )

        self.assertEqual(
            Registry.salt_data(),
            pydmesg_salt_expected
        )

        self.assertEqual(
            Registry.states,
            OrderedDict()
        )

    def test_duplicates(self):
        def add_dup():
            File.managed('dup', name='/dup')

        add_dup()
        self.assertRaises(DuplicateState, add_dup)

        Service.running('dup', name='dup-service')

        self.assertEqual(
            Registry.states,
            OrderedDict([
                ('dup',
                 OrderedDict([
                     ('file.managed', [
                         {'name': '/dup'}
                     ]),
                     ('service.running', [
                         {'name': 'dup-service'}
                     ])
                 ]))
            ])
        )


class RendererMixin(object):
    '''
    This is a mixin that adds a ``.render()`` method to render a template

    It must come BEFORE ``TestCase`` in the declaration of your test case
    class so that our setUp & tearDown get invoked first, and super can
    trigger the methods in the ``TestCase`` class.
    '''
    def setUp(self, *args, **kwargs):
        super(RendererMixin, self).setUp(*args, **kwargs)

        self.root_dir = tempfile.mkdtemp('pyobjects_test_root', dir=integration.TMP)
        self.state_tree_dir = os.path.join(self.root_dir, 'state_tree')
        self.cache_dir = os.path.join(self.root_dir, 'cachedir')
        if not os.path.isdir(self.root_dir):
            os.makedirs(self.root_dir)

        if not os.path.isdir(self.state_tree_dir):
            os.makedirs(self.state_tree_dir)

        if not os.path.isdir(self.cache_dir):
            os.makedirs(self.cache_dir)
        self.config = salt.config.minion_config(None)
        self.config['root_dir'] = self.root_dir
        self.config['state_events'] = False
        self.config['id'] = 'match'
        self.config['file_client'] = 'local'
        self.config['file_roots'] = dict(base=[self.state_tree_dir])
        self.config['cachedir'] = self.cache_dir
        self.config['test'] = False

    def tearDown(self, *args, **kwargs):
        shutil.rmtree(self.root_dir)
        del self.config
        super(RendererMixin, self).tearDown(*args, **kwargs)

    def write_template_file(self, filename, content):
        full_path = os.path.join(self.state_tree_dir, filename)
        with salt.utils.files.fopen(full_path, 'w') as f:
            f.write(content)
        return full_path

    def render(self, template, opts=None, filename=None):
        if opts:
            self.config.update(opts)

        if not filename:
            filename = ".".join([
                str(uuid.uuid4()),
                "sls"
            ])
        full_path = self.write_template_file(filename, template)

        state = salt.state.State(self.config)
        return compile_template(full_path,
                                state.rend,
                                state.opts['renderer'],
                                state.opts['renderer_blacklist'],
                                state.opts['renderer_whitelist'])


class RendererTests(RendererMixin, StateTests, MapBuilder):
    def test_basic(self):
        ret = self.render(basic_template)
        self.assertEqual(ret, OrderedDict([
            ('/tmp', {
                'file.directory': [
                    {'group': 'root'},
                    {'mode': '1777'},
                    {'owner': 'root'}
                ]
            }),
        ]))
        self.assertEqual(Registry.states, OrderedDict())

    def test_invalid_function(self):
        def _test():
            self.render(invalid_template)
        self.assertRaises(InvalidFunction, _test)

    def test_include(self):
        ret = self.render(include_template)
        self.assertEqual(ret, OrderedDict([
            ('include', ['http']),
        ]))

    def test_extend(self):
        ret = self.render(extend_template,
                          {'grains': {
                              'os_family': 'Debian',
                              'os': 'Debian'
                          }})
        self.assertEqual(ret, OrderedDict([
            ('include', ['http']),
            ('extend', OrderedDict([
                ('apache', {
                    'service.running': [
                        {'watch': [{'file': '/etc/file'}]}
                    ]
                }),
            ])),
        ]))

    def test_sls_imports(self):
        def render_and_assert(template):
            ret = self.render(template,
                              {'grains': {
                                  'os_family': 'Debian',
                                  'os': 'Debian'
                              }})

            self.assertEqual(ret, OrderedDict([
                ('samba-imported', {
                    'pkg.removed': [
                        {'names': ['samba', 'samba-client']},
                    ]
                })
            ]))

        self.write_template_file("map.sls", self.build_map())
        render_and_assert(import_template)
        render_and_assert(from_import_template)
        render_and_assert(import_as_template)

        self.write_template_file("recursive_map.sls", recursive_map_template)
        render_and_assert(recursive_import_template)

    def test_import_scope(self):
        self.write_template_file("map.sls", self.build_map())
        self.write_template_file("recursive_map.sls", recursive_map_template)

        def do_render():
            ret = self.render(scope_test_import_template,
                              {'grains': {
                                  'os_family': 'Debian',
                                  'os': 'Debian'
                              }})

        self.assertRaises(NameError, do_render)

    def test_random_password(self):
        '''Test for https://github.com/saltstack/salt/issues/21796'''
        ret = self.render(random_password_template)

    def test_import_random_password(self):
        '''Import test for https://github.com/saltstack/salt/issues/21796'''
        self.write_template_file("password.sls", random_password_template)
        ret = self.render(random_password_import_template)

    def test_requisite_implicit_list(self):
        '''Ensure that the implicit list characteristic works as expected'''
        ret = self.render(requisite_implicit_list_template,
                          {'grains': {
                              'os_family': 'Debian',
                              'os': 'Debian'
                          }})

        self.assertEqual(ret, OrderedDict([
            ('pkg', OrderedDict([
                ('pkg.installed', [])
            ])),
            ('service', OrderedDict([
                ('service.running', [
                    {'require': [{'cmd': 'cmd'}, {'pkg': 'pkg'}]},
                    {'watch': [{'file': 'file'}]},
                ])
            ]))
        ]))


class MapTests(RendererMixin, TestCase, MapBuilder):
    maxDiff = None

    debian_grains = {'os_family': 'Debian', 'os': 'Debian'}
    ubuntu_grains = {'os_family': 'Debian', 'os': 'Ubuntu'}
    centos_grains = {'os_family': 'RedHat', 'os': 'CentOS'}

    debian_attrs = ('samba', 'samba-client', 'samba')
    ubuntu_attrs = ('samba', 'samba-client', 'smbd')
    centos_attrs = ('samba', 'samba', 'smb')

    def samba_with_grains(self, template, grains):
        return self.render(template, {'grains': grains})

    def assert_equal(self, ret, server, client, service):
        self.assertDictEqual(ret, OrderedDict([
            ('samba', OrderedDict([
                ('pkg.installed', [
                    {'names': [server, client]}
                ]),
                ('service.running', [
                    {'name': service},
                    {'require': [{'pkg': 'samba'}]}
                ])
            ]))
        ]))

    def assert_not_equal(self, ret, server, client, service):
        try:
            self.assert_equal(ret, server, client, service)
        except AssertionError:
            pass
        else:
            raise AssertionError('both dicts are equal')

    def test_map(self):
        '''
        Test declarative ordering
        '''
        # With declarative ordering, the ubuntu-specfic service name should
        # override the one inherited from debian.
        template = self.build_map(textwrap.dedent('''\
            {{ debian }}
            {{ centos }}
            {{ ubuntu }}
            '''))

        ret = self.samba_with_grains(template, self.debian_grains)
        self.assert_equal(ret, *self.debian_attrs)

        ret = self.samba_with_grains(template, self.ubuntu_grains)
        self.assert_equal(ret, *self.ubuntu_attrs)

        ret = self.samba_with_grains(template, self.centos_grains)
        self.assert_equal(ret, *self.centos_attrs)

        # Switching the order, debian should still work fine but ubuntu should
        # no longer match, since the debian service name should override the
        # ubuntu one.
        template = self.build_map(textwrap.dedent('''\
            {{ ubuntu }}
            {{ debian }}
            '''))

        ret = self.samba_with_grains(template, self.debian_grains)
        self.assert_equal(ret, *self.debian_attrs)

        ret = self.samba_with_grains(template, self.ubuntu_grains)
        self.assert_not_equal(ret, *self.ubuntu_attrs)

    def test_map_with_priority(self):
        '''
        With declarative ordering, the debian service name would override the
        ubuntu one since debian comes second. This will test overriding this
        behavior using the priority attribute.
        '''
        template = self.build_map(textwrap.dedent('''\
            {{ priority(('os_family', 'os')) }}
            {{ ubuntu }}
            {{ centos }}
            {{ debian }}
            '''))

        ret = self.samba_with_grains(template, self.debian_grains)
        self.assert_equal(ret, *self.debian_attrs)

        ret = self.samba_with_grains(template, self.ubuntu_grains)
        self.assert_equal(ret, *self.ubuntu_attrs)

        ret = self.samba_with_grains(template, self.centos_grains)
        self.assert_equal(ret, *self.centos_attrs)


class SaltObjectTests(TestCase):
    def test_salt_object(self):
        def attr_fail():
            Salt.fail.blah()

        def times2(x):
            return x*2

        __salt__ = {
            'math.times2': times2
        }

        Salt = SaltObject(__salt__)

        self.assertRaises(AttributeError, attr_fail)
        self.assertEqual(Salt.math.times2, times2)
        self.assertEqual(Salt.math.times2(2), 4)