kolla-ansible/tests/test_kolla_toolbox.py

# Copyright 2024 Tietoevry
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import builtins
import json
import os
import sys

from ansible.module_utils import basic
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_bytes
from importlib.machinery import SourceFileLoader
from oslotest import base
from unittest import mock

# Import kolla_toolbox module using SourceFileLoader
this_dir = os.path.dirname(sys.modules[__name__].__file__)
ansible_dir = os.path.join(this_dir, '..', 'ansible')
kolla_toolbox_file = os.path.join(ansible_dir, 'library', 'kolla_toolbox.py')

kolla_toolbox = SourceFileLoader('kolla_toolbox',
                                 kolla_toolbox_file).load_module()


def set_module_args(args):
    """Prepare arguments so they will be picked up during module creation."""

    args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
    basic._ANSIBLE_ARGS = to_bytes(args)


class AnsibleExitJson(BaseException):
    """Exception to be raised by module.exit_json and caught by a test case."""

    def __init__(self, kwargs):
        super().__init__(kwargs)
        self.result = kwargs


class AnsibleFailJson(BaseException):
    """Exception to be raised by module.fail_json and caught by a test case."""

    def __init__(self, kwargs):
        super().__init__(kwargs)
        self.result = kwargs


class MockAPIError(Exception):
    """Mock exception to be raised to simulate engine client APIError."""

    def __init__(self, message, explanation=None):
        super().__init__(message)
        self.explanation = explanation


class TestKollaToolboxModule(base.BaseTestCase):
    """Base class for the module's tests.

    Sets up methods that patch over the module's fail_json and exit_json,
    so that they dont just call sys.exit() and instead they return
    value of the result.
    """

    def setUp(self):
        super().setUp()

        self.fail_json_patch = mock.patch(
            'ansible.module_utils.basic.AnsibleModule.fail_json',
            side_effect=self.fail_json)
        self.exit_json_patch = mock.patch(
            'ansible.module_utils.basic.AnsibleModule.exit_json',
            side_effect=self.exit_json)

        self.fail_json_mock = self.fail_json_patch.start()
        self.exit_json_mock = self.exit_json_patch.start()

    def tearDown(self):
        super().tearDown()
        self.fail_json_patch.stop()
        self.exit_json_patch.stop()

    def exit_json(self, *args, **kwargs):
        raise AnsibleExitJson(kwargs)

    def fail_json(self, *args, **kwargs):
        raise AnsibleFailJson(kwargs)


class TestKollaToolboxMethods(TestKollaToolboxModule):
    """Class focused on testing the methods of KollaToolboxWorker."""

    def setUp(self):
        super().setUp()

        # Mock container client
        self.mock_container_client = mock.MagicMock()
        self.mock_container_errors = mock.MagicMock()
        self.mock_container_errors.APIError = MockAPIError

        # Mock Ansible module
        self.mock_ansible_module = mock.MagicMock()
        self.mock_ansible_module.fail_json.side_effect = self.fail_json
        self.mock_ansible_module.exit_json.side_effect = self.exit_json

        # Fake Kolla Toolbox Worker
        self.fake_ktbw = kolla_toolbox.KollaToolboxWorker(
            self.mock_ansible_module,
            self.mock_container_client,
            self.mock_container_errors)

    def test_ktb_container_missing_or_not_running(self):
        self.mock_container_client.containers.list.return_value = []

        error = self.assertRaises(AnsibleFailJson,
                                  self.fake_ktbw._get_toolbox_container)
        self.assertIn("kolla_toolbox container is missing or not running!",
                      error.result["msg"])

    def test_get_ktb_container_success(self):
        ktb_container = mock.MagicMock()
        other_container = mock.MagicMock()
        self.mock_container_client.containers.list.return_value = [
            ktb_container, other_container]

        ktb_container_returned = self.fake_ktbw._get_toolbox_container()

        self.assertEqual(ktb_container, ktb_container_returned)

    def test_format_module_args(self):
        module_args = [
            {
                'module_args': {},
                'expected_output': []
            },
            {
                'module_args': {
                    'path': '/some/folder',
                    'state': 'absent'},
                'expected_output': ["path='/some/folder'", "state='absent'"]
            }
        ]

        for args in module_args:
            formatted_args = self.fake_ktbw._format_module_args(
                args['module_args'])

            self.assertEqual(args['expected_output'], formatted_args)

    @mock.patch('kolla_toolbox.KollaToolboxWorker._format_module_args')
    def test_generate_correct_ktb_command(self, mock_formatter):
        fake_module_params = {
            'module_args': {
                'path': '/some/folder',
                'state': 'absent'
            },
            'module_extra_vars': {
                'variable': {
                    'key': 'pair',
                    'list': ['item1', 'item2']
                }
            },
            'user': 'root',
            'module_name': 'file'
        }

        mock_params = mock.MagicMock()
        mock_params.get.side_effect = lambda key: fake_module_params.get(key)
        self.mock_ansible_module.params = mock_params

        mock_formatter.side_effect = [
            ["path='/some/folder'", "state='absent'"],
            ['variable=\'{"key": "pair", "list": ["item1", "item2"]}\'']
        ]

        expected_command = ['ansible', 'localhost', '-m', 'file',
                            '-a', "path='/some/folder' state='absent'",
                            '-e', 'variable=\'{"key": "pair", '
                            '"list": ["item1", "item2"]}\'',
                            '--check']

        generated_command = self.fake_ktbw._generate_command()

        self.assertEqual(expected_command, generated_command)
        mock_formatter.assert_has_calls([
            mock.call(fake_module_params['module_args']),
            mock.call(fake_module_params['module_extra_vars'])
        ])

    def test_run_command_raises_apierror(self):
        ktb_container = mock.MagicMock()
        api_error = self.mock_container_errors.APIError(
            'API error occurred', explanation='Error explanation')
        ktb_container.exec_run.side_effect = api_error

        error = self.assertRaises(AnsibleFailJson,
                                  self.fake_ktbw._run_command,
                                  ktb_container,
                                  'some_command')
        self.assertIn('Container engine client encountered API error',
                      error.result['msg'])

    def test_run_command_success(self):
        exec_return_value = (0, b'data')
        ktb_container = mock.MagicMock()
        ktb_container.exec_run.return_value = exec_return_value
        self.mock_container_client.containers.list.return_value = [
            ktb_container]

        command_output = self.fake_ktbw._run_command(
            ktb_container, 'some_command')

        self.assertEqual(exec_return_value[1], command_output)
        self.assertIsInstance(command_output, bytes)
        ktb_container.exec_run.assert_called_once_with('some_command')

    def test_process_container_output_invalid_json(self):
        invalid_json = b'this is no json'

        error = self.assertRaises(AnsibleFailJson,
                                  self.fake_ktbw._process_container_output,
                                  invalid_json)
        self.assertIn('Parsing kolla_toolbox JSON output failed',
                      error.result['msg'])

    def test_process_container_output_invalid_structure(self):
        wrong_output_json = {
            'plays': [
                {
                    'tasks': [
                        {
                            'wrong': {
                                'control_node': {
                                    'pong': 'ping'
                                }
                            }
                        }
                    ]
                }
            ]
        }
        encoded_json = json.dumps(wrong_output_json).encode('utf-8')

        error = self.assertRaises(AnsibleFailJson,
                                  self.fake_ktbw._process_container_output,
                                  encoded_json)
        self.assertIn('Ansible JSON output has unexpected format',
                      error.result['msg'])

    def test_process_container_output_success(self):
        container_output_json = {
            'custom_stats': {},
            'global_custom_stats': {},
            'plays': [
                {
                    'tasks': [
                        {
                            'hosts': {
                                'localhost': {
                                    '_ansible_no_log': False,
                                    'action': 'ping',
                                    'changed': False,
                                    'invocation': {
                                        'module_args': {
                                            'data': 'pong'
                                        }
                                    },
                                    'ping': 'pong'
                                }
                            },
                        }
                    ]
                }
            ],
        }
        container_encoded_json = json.dumps(
            container_output_json).encode('utf-8')

        expected_output = {
            'action': 'ping',
            'changed': False,
            'invocation': {
                'module_args': {
                    'data': 'pong'
                }
            },
            'ping': 'pong'
        }
        generated_module_output = self.fake_ktbw._process_container_output(
            container_encoded_json)

        self.assertNotIn('_ansible_no_log', generated_module_output)
        self.assertEqual(expected_output, generated_module_output)


class TestModuleInteraction(TestKollaToolboxModule):
    """Class focused on testing user input data from playbook."""

    def test_create_ansible_module_missing_required_module_name(self):
        set_module_args({
            'container_engine': 'docker'
        })

        error = self.assertRaises(AnsibleFailJson,
                                  kolla_toolbox.create_ansible_module)
        self.assertIn('missing required arguments: module_name',
                      error.result['msg'])

    def test_create_ansible_module_missing_required_container_engine(self):
        set_module_args({
            'module_name': 'url'
        })

        error = self.assertRaises(AnsibleFailJson,
                                  kolla_toolbox.create_ansible_module)
        self.assertIn('missing required arguments: container_engine',
                      error.result['msg'])

    def test_create_ansible_module_invalid_container_engine(self):
        set_module_args({
            'module_name': 'url',
            'container_engine': 'podmano'
        })

        error = self.assertRaises(AnsibleFailJson,
                                  kolla_toolbox.create_ansible_module)
        self.assertIn(
            'value of container_engine must be one of: podman, docker',
            error.result['msg']
        )

    def test_create_ansible_module_success(self):
        args = {
            'container_engine': 'docker',
            'module_name': 'file',
            'module_args': {
                'path': '/some/folder',
                'state': 'absent'
            },
            'module_extra_vars': {
                'variable': {
                    'key': 'pair',
                    'list': ['item1', 'item2']
                }
            },
            'user': 'root',
            'timeout': 180,
            'api_version': '1.5'
        }
        set_module_args(args)

        module = kolla_toolbox.create_ansible_module()

        self.assertIsInstance(module, AnsibleModule)
        self.assertEqual(args, module.params)


class TestContainerEngineClientIntraction(TestKollaToolboxModule):
    """Class focused on testing container engine client creation."""

    def setUp(self):
        super().setUp()
        self.module_to_mock_import = ''
        self.original_import = builtins.__import__

    def mock_import_error(self, name, globals, locals, fromlist, level):
        """Mock import function to raise ImportError for a specific module."""

        if name == self.module_to_mock_import:
            raise ImportError(f'No module named {name}')
        return self.original_import(name, globals, locals, fromlist, level)

    def test_podman_client_params(self):
        set_module_args({
            'module_name': 'ping',
            'container_engine': 'podman',
            'api_version': '1.47',
            'timeout': 155
        })

        module = kolla_toolbox.create_ansible_module()
        mock_podman = mock.MagicMock()
        mock_podman_errors = mock.MagicMock()
        import_dict = {'podman': mock_podman,
                       'podman.errors': mock_podman_errors}

        with mock.patch.dict('sys.modules', import_dict):
            kolla_toolbox.create_container_client(module)
            mock_podman.PodmanClient.assert_called_with(
                base_url='http+unix:/run/podman/podman.sock',
                version='1.47',
                timeout=155
            )

    def test_docker_client_params(self):
        set_module_args({
            'module_name': 'ping',
            'container_engine': 'docker',
            'api_version': '1.47',
            'timeout': 155
        })

        module = kolla_toolbox.create_ansible_module()
        mock_docker = mock.MagicMock()
        mock_docker_errors = mock.MagicMock()
        import_dict = {'docker': mock_docker,
                       'docker.errors': mock_docker_errors}

        with mock.patch.dict('sys.modules', import_dict):
            kolla_toolbox.create_container_client(module)
            mock_docker.DockerClient.assert_called_with(
                base_url='http+unix:/var/run/docker.sock',
                version='1.47',
                timeout=155
            )

    def test_create_container_client_podman_not_called_with_auto(self):
        set_module_args({
            'module_name': 'ping',
            'container_engine': 'podman',
            'api_version': 'auto',
            'timeout': 90
        })

        module = kolla_toolbox.create_ansible_module()
        mock_podman = mock.MagicMock()
        mock_podman_errors = mock.MagicMock()
        import_dict = {'podman': mock_podman,
                       'podman.errors': mock_podman_errors}

        with mock.patch.dict('sys.modules', import_dict):
            kolla_toolbox.create_container_client(module)
            mock_podman.PodmanClient.assert_called_with(
                base_url='http+unix:/run/podman/podman.sock',
                timeout=90
            )

    def test_create_container_client_podman_importerror(self):
        set_module_args({
            'module_name': 'ping',
            'container_engine': 'podman'
        })
        self.module_to_mock_import = 'podman'
        module = kolla_toolbox.create_ansible_module()

        with mock.patch('builtins.__import__',
                        side_effect=self.mock_import_error):
            error = self.assertRaises(AnsibleFailJson,
                                      kolla_toolbox.create_container_client,
                                      module)
            self.assertIn('The podman library could not be imported!',
                          error.result['msg'])

    def test_create_container_client_docker_importerror(self):
        set_module_args({
            'module_name': 'ping',
            'container_engine': 'docker'
        })

        self.module_to_mock_import = 'docker'
        module = kolla_toolbox.create_ansible_module()

        with mock.patch('builtins.__import__',
                        side_effect=self.mock_import_error):
            error = self.assertRaises(AnsibleFailJson,
                                      kolla_toolbox.create_container_client,
                                      module)
            self.assertIn('The docker library could not be imported!',
                          error.result['msg'])