ara/ara/tests/unit/test_ara_record.py

#  Copyright (c) 2018 Red Hat, Inc.
#
#  This file is part of ARA Records Ansible.
#
#  ARA is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  ARA is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with ARA.  If not, see <http://www.gnu.org/licenses/>.

from collections import defaultdict
import random

import ara.models as m
import ara.plugins.callbacks.log_ara as l
import ara.plugins.actions.ara_record as ara_record

from ara.tests.unit.common import TestAra

from mock import Mock
from mock import MagicMock


class Playbook(object):
    def __init__(self, path):
        self._file_name = path
        self.path = path
        self.options = {'fake': 'yes'}


class Play(object):
    def __init__(self, name):
        self.name = name


class Task(object):
    def __init__(self, name, path, lineno=1, action='fakeaction'):
        self.name = name
        self.action = action
        self.path = '%s:%d' % (path, lineno)
        self._attributes = {'tags': []}

    def get_path(self):
        """ Callback specific method """
        return self.path

    def get_name(self):
        """ Callback specific method """
        return self.name


class TaskResult(object):
    def __init__(self, task, host, status, changed=False):
        assert status in ['ok', 'failed', 'skipped', 'unreachable']

        self.task = task
        self.status = status
        self._host = MagicMock()
        self._host.get_name.return_value = host
        self._result = {
            'changed': changed,
            'failed': status == 'failed',
            'skipped': status == 'skipped',
            'unreachable': status == 'unreachable',
        }


class Stats(object):
    def __init__(self, processed):
        self.processed = processed

    def summarize(self, name):
        return {
            'failures': self.processed[name]['failed'],
            'ok': self.processed[name]['ok'],
            'changed': self.processed[name]['changed'],
            'skipped': self.processed[name]['skipped'],
            'unreachable': self.processed[name]['unreachable'],
        }


class TestRecord(TestAra):
    '''Tests for the Ansible ara_record module'''
    def setUp(self):
        super(TestRecord, self).setUp()

        self.cb = l.CallbackModule()
        self.tag = '%04d' % random.randint(0, 9999)

        self.ansible_run()

        self.play_context = Mock()
        self.play_context.check_mode = False
        self.connection = Mock()

    def tearDown(self):
        super(TestRecord, self).tearDown()

    def ansible_run(self):
        '''Simulates an ansible run by creating stub versions of the
        information that Ansible passes to the callback, and then
        calling the various callback methods.'''

        self.playbook = self._test_playbook()
        self.play = self._test_play()
        self.task = self._test_task(self.playbook)
        self.results = [
            self._test_result(self.task, 'host1', 'ok', changed=True),
        ]

        self.stats = self._test_stats()

    def _test_stats(self):
        stats = Stats({
            'host1': defaultdict(int, ok=1, changed=1),
            'host2': defaultdict(int, failed=1),
        })

        self.cb.v2_playbook_on_stats(stats)
        return stats

    def _test_result(self, task, host, status='ok', changed=False):
        result = TaskResult(task, host, status, changed)
        func = getattr(self.cb, 'v2_runner_on_%s' % status)
        func(result)
        return result

    def _test_playbook(self):
        path = '/test-playbook-%s.yml' % self.tag
        playbook = Playbook(path)
        self.cb.v2_playbook_on_start(playbook)
        return playbook

    def _test_play(self):
        name = 'test-play-%s' % self.tag
        play = Play(name)
        self.cb.v2_playbook_on_play_start(play)
        return play

    def _test_task(self, playbook):
        name = 'test-task-%s' % self.tag
        task = Task(name, playbook.path)
        self.cb.v2_playbook_on_task_start(task, False)
        return task

    def test_create_text_record_with_playbook(self):
        """
        Create a new record with ara_record on a specified playbook
        """
        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'playbook': r_playbook.id,
            'key': 'test-text',
            'value': 'test-with-playbook',
            'type': 'text'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-text').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-text')
        self.assertEqual(r_data.value, 'test-with-playbook')
        self.assertEqual(r_data.type, 'text')

    def test_create_text_record(self):
        """
        Create a new record with ara_record.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-text',
            'value': 'test-value',
            'type': 'text'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-text').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-text')
        self.assertEqual(r_data.value, 'test-value')
        self.assertEqual(r_data.type, 'text')

    def test_create_url_record(self):
        """
        Create a new record with ara_record.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-url',
            'value': 'http://url',
            'type': 'url'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-url').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-url')
        self.assertEqual(r_data.value, 'http://url')
        self.assertEqual(r_data.type, 'url')

    def test_create_json_record(self):
        """
        Create a new record with ara_record.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-json',
            'value': '{"foo": "bar"}',
            'type': 'json'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-json').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-json')
        self.assertEqual(r_data.value, '{"foo": "bar"}')
        self.assertEqual(r_data.type, 'json')

    def test_create_list_record(self):
        """
        Create a new record with ara_record.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-list',
            'value': ['foo', 'bar'],
            'type': 'list'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-list').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-list')
        self.assertEqual(r_data.value, ['foo', 'bar'])
        self.assertEqual(r_data.type, 'list')

    def test_create_dict_record(self):
        """
        Create a new record with ara_record.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-dict',
            'value': {'foo': 'bar'},
            'type': 'dict'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-dict').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-dict')
        self.assertEqual(r_data.value, {'foo': 'bar'})
        self.assertEqual(r_data.type, 'dict')

    def test_create_record_with_no_type(self):
        """
        Create a new record with ara_record with no type specified.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-notype',
            'value': 'test-value'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-notype').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-notype')
        self.assertEqual(r_data.value, 'test-value')
        self.assertEqual(r_data.type, 'text')

    def test_create_record_as_wrong_type(self):
        """
        Create a new record with ara_record.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-wrongtype',
            'value': ['foo', 'bar'],
            'type': 'text'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-wrongtype').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-wrongtype')
        self.assertEqual(r_data.value, ['foo', 'bar'])
        self.assertEqual(r_data.type, 'text')

    def test_update_record(self):
        """
        Update an existing record by running ara_record a second time on the
        same key.
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-update',
            'value': 'test-value',
            'type': 'text'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-update').one()
        self.assertIsNotNone(r_data)
        self.assertEqual(r_data.playbook_id, r_playbook.id)
        self.assertEqual(r_data.key, 'test-update')
        self.assertEqual(r_data.value, 'test-value')
        self.assertEqual(r_data.type, 'text')

        task.args = {
            'key': 'test-update',
            'value': 'http://another-value',
            'type': 'url'
        }
        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_data = m.Data.query.filter_by(playbook_id=r_playbook.id,
                                        key='test-update').one()

        self.assertEqual(r_data.value, 'http://another-value')
        self.assertEqual(r_data.type, 'url')

    def test_record_with_no_key(self):
        """
        Trying to use ara_record with no key parameter should properly fail
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'value': 'test-value'
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        # There is no exception raised in the action module, we instead
        # properly return a failure status to Ansible.
        # If there is a failure, no data will be recorded so we can catch this.
        with self.assertRaises(Exception):
            m.Data.query.filter_by(playbook_id=r_playbook.id).one()

    def test_record_with_no_value(self):
        """
        Trying to use ara_record with no value parameter should properly fail
        """
        task = MagicMock(Task)
        task.async_ = 0
        task.async_val = False
        task.args = {
            'key': 'test-key',
        }

        action = ara_record.ActionModule(task, self.connection,
                                         self.play_context, loader=None,
                                         templar=None, shared_loader_obj=None)
        action.run()

        r_playbook = m.Playbook.query.first()
        self.assertIsNotNone(r_playbook)

        # There is no exception raised in the action module, we instead
        # properly return a failure status to Ansible.
        # If there is a failure, no data will be recorded so we can catch this.
        with self.assertRaises(Exception):
            m.Data.query.filter_by(playbook_id=r_playbook.id).one()