path: root/tests/test_kolla_container_facts.py

#!/usr/bin/env python

# Copyright 2016 NEC Corporation
# 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.

# NOTE(r-krcek): As the format of data from PodmanClient are nearly identical
# to data returned by DockerClient. The tests can be ran only on one of the
# clients. (There are certain exceptions but none apply to these tests)


import copy
from importlib.machinery import SourceFileLoader
import os
import sys
from unittest import mock

from docker import errors as docker_error
from oslotest import base


this_dir = os.path.dirname(sys.modules[__name__].__file__)
ansible_dir = os.path.join(this_dir, '..', 'ansible')
kolla_container_facts_file = os.path.join(
    ansible_dir,
    'library', 'kolla_container_facts.py')
kcf = SourceFileLoader('kolla_container_facts',
                       kolla_container_facts_file).load_module()


FAKE_DATA = {
    'containers': [
        {'Created': '2022-06-23T14:30:35.595194629Z',
         'State': {'Status': 'running'},
         'HostConfig': {'NetworkMode': 'host'},
         'Id': '1663dfafec3bb59386e4a024416c8b0a872ae0984c9806322751d14b9f794c56',      # noqa: E501
         'ImageName': 'myregistrydomain.com:5000/ubuntu:16.04',
         'Image': '7528a4009573fa8c5dbf4b6f5fad9f5b8d3a0fb90e22bb1b217211b553eb22cf',   # noqa: E501
         'Labels': {},
         'Name': 'my_container'},
        {'Created': '2022-06-23T14:32:13.17545575Z',
         'State': {'Status': 'exited'},
         'HostConfig': {'NetworkMode': 'host'},
         'Id': '9404fc5f90118ddbbc31bb4c9462ad06aa7163eac1bc6d74c3e978143f10cc0c',      # noqa: E501
         'ImageName': 'myregistrydomain.com:5000/ubuntu:16.04',
         'Image': '15529c81ae4a83084b076a16bc314e1af0b040a937f585311c87863fecc623a3',   # noqa: E501
         'Labels': {},
         'Name': 'exited_container'},
    ],
    'container_inspect': {
        'Config': {
            'Env': ['KOLLA_BASE_DISTRO=ubuntu',
                    'KOLLA_INSTALL_TYPE=binary',
                    'KOLLA_INSTALL_METATYPE=rdo'],
            'Hostname': 'node2',
            'Volumes': {'/var/lib/kolla/config_files/': {}}},
        'Mounts': {},
        'NetworkSettings': {}
    },
    'volumes': [
        {'CreatedAt': '2024-10-10T12:05:46+02:00',
         'Driver': 'local',
         'Labels': None,
         'Mountpoint': '/var/lib/docker/volumes/my_volume/_data',
         'Name': 'my_volume',
         'Options': None,
         'Scope': 'local'},
        {'CreatedAt': '2023-05-01T14:55:36+02:00',
         'Driver': 'local',
         'Labels': None,
         'Mountpoint': '/var/lib/docker/volumes/another_volume/_data',
         'Name': 'another_volume',
         'Options': None,
         'Scope': 'local'}
    ]
}


@mock.patch('docker.DockerClient')
def get_DockerFactsWorker(mod_param, mock_client):
    module = mock.MagicMock()
    module.params = copy.deepcopy(mod_param)
    dfw = kcf.DockerFactsWorker(module)
    return dfw


def construct_container(cont_dict) -> mock.Mock:
    container = mock.Mock()
    container.name = cont_dict['Name']
    container.attrs = copy.deepcopy(cont_dict)
    container.status = cont_dict['State']['Status']
    return container


def contruct_volume(vol_dict: dict) -> mock.Mock:
    """Creates volume object that mimics the output of a container client."""
    volume = mock.Mock()
    volume.name = vol_dict['Name']
    volume.attrs = copy.deepcopy(vol_dict)
    return volume


def get_containers(override=None, all: bool = False):
    if override:
        cont_dicts = override
    else:
        cont_dicts = copy.deepcopy(FAKE_DATA['containers'])

    containers = []
    for c in cont_dicts:
        # With the option "all", only running containers are returned
        # by the container API
        if not all and c['State']['Status'] != 'running':
            continue
        containers.append(construct_container(c))

    return containers


def get_volumes(override=None):
    if override:
        vol_dicts = override
    else:
        vol_dicts = copy.deepcopy(FAKE_DATA['volumes'])

    return [contruct_volume(v) for v in vol_dicts]


class TestContainerFacts(base.BaseTestCase):
    def setUp(self):
        super(TestContainerFacts, self).setUp()
        self.fake_data = copy.deepcopy(FAKE_DATA)

    def test_get_containers_single(self):
        self.dfw = get_DockerFactsWorker({'name': ['my_container'],
                                          'action': 'get_containers'})
        running_containers = get_containers(self.fake_data['containers'])
        self.dfw.client.containers.list.return_value = running_containers
        self.dfw.get_containers()

        self.assertFalse(self.dfw.result['changed'])
        self.assertEqual(self.dfw.client.containers.list.call_count, 1)
        self.assertIn('my_container', self.dfw.result['containers'])
        self.assertNotIn('my_container', self.dfw.result)
        self.assertDictEqual(
            self.fake_data['containers'][0],
            self.dfw.result['containers']['my_container'])
        self.dfw.client.containers.list.assert_called_once_with(all=False)

    def test_get_containers_multi(self):
        self.dfw = get_DockerFactsWorker(
            {'name': ['my_container', 'exited_container'],
             'action': 'get_containers'})
        running_containers = get_containers(self.fake_data['containers'])
        self.dfw.client.containers.list.return_value = running_containers
        self.dfw.get_containers()

        self.assertFalse(self.dfw.result['changed'])
        self.assertIn('my_container', self.dfw.result['containers'])
        self.assertNotIn('my_container', self.dfw.result)
        self.assertNotIn('exited_container', self.dfw.result['containers'])
        self.dfw.client.containers.list.assert_called_once_with(all=False)

    def test_get_containers_all_running(self):
        self.dfw = get_DockerFactsWorker({'name': [],
                                          'action': 'get_containers'})
        running_containers = get_containers(self.fake_data['containers'])
        self.dfw.client.containers.list.return_value = running_containers
        self.dfw.get_containers()

        self.assertFalse(self.dfw.result['changed'])
        self.assertIn('my_container', self.dfw.result['containers'])
        self.assertNotIn('my_container', self.dfw.result)
        self.assertNotIn('exited_container', self.dfw.result['containers'])
        self.dfw.client.containers.list.assert_called_once_with(all=False)

    def test_get_containers_all_including_stopped(self):
        self.dfw = get_DockerFactsWorker({'name': [],
                                          'action': 'get_containers',
                                          'args': {
                                              'get_all_containers': True}})
        all_containers = get_containers(self.fake_data['containers'], all=True)
        self.dfw.client.containers.list.return_value = all_containers
        self.dfw.get_containers()

        self.assertFalse(self.dfw.result['changed'])
        self.assertIn('my_container', self.dfw.result['containers'])
        self.assertIn('exited_container', self.dfw.result['containers'])
        self.dfw.client.containers.list.assert_called_once_with(all=True)

    def test_get_containers_env(self):
        fake_env = dict(KOLLA_BASE_DISTRO='ubuntu',
                        KOLLA_INSTALL_TYPE='binary',
                        KOLLA_INSTALL_METATYPE='rdo')
        self.dfw = get_DockerFactsWorker({'name': ['my_container'],
                                          'action': 'get_containers_env'})
        self.fake_data['containers'][0].update(
            self.fake_data['container_inspect'])
        self.dfw.client.containers.get.return_value = construct_container(
            self.fake_data['containers'][0])
        self.dfw.get_containers_env()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.containers.get.assert_called_once_with('my_container')
        self.assertIn('my_container', self.dfw.result['envs'])
        self.assertNotIn('my_container', self.dfw.result)
        self.assertEqual(self.dfw.result['envs']['my_container'], fake_env)

    def test_get_containers_env_negative(self):
        self.dfw = get_DockerFactsWorker({'name': ['fake_container'],
                                          'action': 'get_containers_env'})
        not_found_exc = docker_error.NotFound("not found")
        self.dfw.client.containers.get = mock.Mock(side_effect=not_found_exc)
        self.dfw.get_containers_env()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.containers.get.assert_called_once_with(
            'fake_container')
        self.dfw.module.fail_json.assert_called_once_with(
            msg="No such container: fake_container")

    def test_get_containers_state(self):
        state = {'Dead': False,
                 'ExitCode': 0,
                 'Pid': 12475,
                 'StartedAt': '2016-06-07T11:22:37.66876269Z',
                 'Status': 'running'}
        self.fake_data['container_inspect'].update({'State': state})
        self.dfw = get_DockerFactsWorker({'name': ['my_container'],
                                          'action': 'get_containers_state'})
        self.fake_data['containers'][0].update({'State': state})
        self.dfw.client.containers.get.return_value = construct_container(
            self.fake_data['containers'][0])
        self.dfw.get_containers_state()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.containers.get.assert_called_once_with('my_container')
        self.assertIn('my_container', self.dfw.result['states'])
        self.assertNotIn('my_container', self.dfw.result)

    def test_get_containers_state_negative(self):
        self.dfw = get_DockerFactsWorker({'name': ['fake_container'],
                                          'action': 'get_containers_state'})
        not_found_exc = docker_error.NotFound("not found")
        self.dfw.client.containers.get = mock.Mock(side_effect=not_found_exc)
        self.dfw.get_containers_state()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.containers.get.assert_called_once_with(
            'fake_container')
        self.dfw.module.fail_json.assert_called_once_with(
            msg="No such container: fake_container")

    def test_get_volumes_single(self):
        """Test fetching a single volume"""
        self.dfw = get_DockerFactsWorker(
            {'name': ['my_volume'], 'action': 'get_volumes'})
        full_vol_list = get_volumes(self.fake_data['volumes'])
        self.dfw.client.volumes.list.return_value = full_vol_list

        self.dfw.get_volumes()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.volumes.list.assert_called_once_with()
        self.assertIn('my_volume', self.dfw.result['volumes'])
        self.assertNotIn('my_volume', self.dfw.result)
        self.assertNotIn('another_volume', self.dfw.result['volumes'])
        self.assertEqual(len(self.dfw.result['volumes']), 1)
        self.assertDictEqual(
            self.dfw.result['volumes']['my_volume'],
            self.fake_data['volumes'][0])

    def test_get_volumes_multiple(self):
        """Test fetching multiple volumes"""
        self.dfw = get_DockerFactsWorker({
            'name': ['my_volume', 'another_volume'],
            'action': 'get_volumes'})
        full_vol_list = get_volumes(self.fake_data['volumes'])
        self.dfw.client.volumes.list.return_value = full_vol_list

        self.dfw.get_volumes()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.volumes.list.assert_called_once_with()
        self.assertIn('my_volume', self.dfw.result['volumes'])
        self.assertIn('another_volume', self.dfw.result['volumes'])
        self.assertNotIn('my_volume', self.dfw.result)
        self.assertNotIn('another_volume', self.dfw.result)
        self.assertEqual(len(self.dfw.result['volumes']), 2)
        self.assertDictEqual(
            self.dfw.result['volumes']['my_volume'],
            self.fake_data['volumes'][0])
        self.assertDictEqual(
            self.dfw.result['volumes']['another_volume'],
            self.fake_data['volumes'][1])

    def test_get_volumes_all(self):
        """Test fetching all volumes when no specific names are provided"""
        self.dfw = get_DockerFactsWorker({'name': [],
                                          'action': 'get_volumes'})
        full_vol_list = get_volumes(self.fake_data['volumes'])
        self.dfw.client.volumes.list.return_value = full_vol_list

        self.dfw.get_volumes()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.volumes.list.assert_called_once_with()
        self.assertIn('my_volume', self.dfw.result['volumes'])
        self.assertIn('another_volume', self.dfw.result['volumes'])
        self.assertNotIn('my_volume', self.dfw.result)
        self.assertNotIn('another_volume', self.dfw.result)
        self.assertDictEqual(
            self.dfw.result['volumes']['my_volume'],
            self.fake_data['volumes'][0])
        self.assertDictEqual(
            self.dfw.result['volumes']['another_volume'],
            self.fake_data['volumes'][1])

    def test_get_volumes_negative(self):
        """Test fetching a volume that doesn't exist"""
        self.dfw = get_DockerFactsWorker({'name': ['nonexistent_volume'],
                                          'action': 'get_volumes'})
        full_vol_list = get_volumes(self.fake_data['volumes'])
        self.dfw.client.volumes.list.return_value = full_vol_list

        self.dfw.get_volumes()

        self.assertFalse(self.dfw.result['changed'])
        self.dfw.client.volumes.list.assert_called_once_with()
        self.assertIn('volumes', self.dfw.result)
        self.assertEqual(len(self.dfw.result['volumes']), 0)
        self.assertNotIn('nonexistent_volume', self.dfw.result['volumes'])

    def test_get_containers_names(self):
        """Test fetching container names when containers exist"""
        self.dfw = get_DockerFactsWorker({'action': 'get_containers_names'})
        full_cont_list = get_containers(self.fake_data['containers'])
        self.dfw.client.containers.list.return_value = full_cont_list
        expected_names = ["my_container"]

        self.dfw.get_containers_names()

        self.assertFalse(self.dfw.result['changed'])
        self.assertIn('container_names', self.dfw.result)
        self.assertEqual(self.dfw.result['container_names'], expected_names)

    def test_get_containers_names_no_containers(self):
        """Test fetching container names when no containers exist"""
        self.dfw = get_DockerFactsWorker({'action': 'get_containers_names'})
        self.dfw.client.containers.list.return_value = []

        self.dfw.get_containers_names()

        self.assertFalse(self.dfw.result['changed'])
        self.assertIn('container_names', self.dfw.result)
        self.assertEqual(self.dfw.result['container_names'], [])