python-openstackclient/openstackclient/tests/unit/image/v2/fakes.py

#   Copyright 2013 Nebula Inc.
#
#   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 random
from unittest import mock
import uuid

from openstack.image.v2 import image
from openstack.image.v2 import member
from openstack.image.v2 import metadef_namespace
from openstack.image.v2 import metadef_resource_type
from openstack.image.v2 import service_info as _service_info
from openstack.image.v2 import task

from openstackclient.tests.unit import fakes
from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes
from openstackclient.tests.unit import utils


class FakeImagev2Client:
    def __init__(self, **kwargs):
        self.images = mock.Mock()
        self.create_image = mock.Mock()
        self.delete_image = mock.Mock()
        self.update_image = mock.Mock()
        self.find_image = mock.Mock()
        self.get_image = mock.Mock()
        self.download_image = mock.Mock()
        self.reactivate_image = mock.Mock()
        self.deactivate_image = mock.Mock()
        self.stage_image = mock.Mock()
        self.import_image = mock.Mock()

        self.members = mock.Mock()
        self.add_member = mock.Mock()
        self.remove_member = mock.Mock()
        self.update_member = mock.Mock()

        self.remove_tag = mock.Mock()
        self.metadef_namespaces = mock.Mock()
        self.metadef_resource_types = mock.Mock()

        self.tasks = mock.Mock()
        self.tasks.resource_class = fakes.FakeResource(None, {})
        self.get_task = mock.Mock()

        self.get_import_info = mock.Mock()

        self.auth_token = kwargs['token']
        self.management_url = kwargs['endpoint']
        self.version = 2.0


class TestImagev2(utils.TestCommand):
    def setUp(self):
        super().setUp()

        self.app.client_manager.image = FakeImagev2Client(
            endpoint=fakes.AUTH_URL,
            token=fakes.AUTH_TOKEN,
        )

        self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client(
            endpoint=fakes.AUTH_URL,
            token=fakes.AUTH_TOKEN,
        )


def create_one_image(attrs=None):
    """Create a fake image.

    :param attrs: A dictionary with all attributes of image
    :type attrs: dict
    :return: A fake Image object.
    :rtype: `openstack.image.v2.image.Image`
    """
    attrs = attrs or {}

    # Set default attribute
    image_info = {
        'id': str(uuid.uuid4()),
        'name': 'image-name' + uuid.uuid4().hex,
        'owner_id': 'image-owner' + uuid.uuid4().hex,
        'is_protected': bool(random.choice([0, 1])),
        'visibility': random.choice(['public', 'private']),
        'tags': [uuid.uuid4().hex for r in range(2)],
    }

    # Overwrite default attributes if there are some attributes set
    image_info.update(attrs)

    return image.Image(**image_info)


def create_images(attrs=None, count=2):
    """Create multiple fake images.

    :param attrs: A dictionary with all attributes of image
    :type attrs: dict
    :param count: The number of images to be faked
    :type count: int
    :return: A list of fake Image objects
    :rtype: list
    """
    images = []
    for n in range(0, count):
        images.append(create_one_image(attrs))

    return images


def create_one_image_member(attrs=None):
    """Create a fake image member.

    :param attrs: A dictionary with all attributes of image member
    :type attrs: dict
    :return: A fake Member object.
    :rtype: `openstack.image.v2.member.Member`
    """
    attrs = attrs or {}

    # Set default attribute
    image_member_info = {
        'member_id': 'member-id-' + uuid.uuid4().hex,
        'image_id': 'image-id-' + uuid.uuid4().hex,
        'status': 'pending',
    }

    # Overwrite default attributes if there are some attributes set
    image_member_info.update(attrs)

    return member.Member(**image_member_info)


def create_one_import_info(attrs=None):
    """Create a fake import info.

    :param attrs: A dictionary with all attributes of import info
    :type attrs: dict
    :return: A fake Import object.
    :rtype: `openstack.image.v2.service_info.Import`
    """
    attrs = attrs or {}

    import_info = {
        'import-methods': {
            'description': 'Import methods available.',
            'type': 'array',
            'value': [
                'glance-direct',
                'web-download',
                'glance-download',
                'copy-image',
            ],
        }
    }
    import_info.update(attrs)

    return _service_info.Import(**import_info)


def create_one_task(attrs=None):
    """Create a fake task.

    :param attrs: A dictionary with all attributes of task
    :type attrs: dict
    :return: A fake Task object.
    :rtype: `openstack.image.v2.task.Task`
    """
    attrs = attrs or {}

    # Set default attribute
    task_info = {
        'created_at': '2016-06-29T16:13:07Z',
        'expires_at': '2016-07-01T16:13:07Z',
        'id': str(uuid.uuid4()),
        'input': {
            'image_properties': {
                'container_format': 'ovf',
                'disk_format': 'vhd',
            },
            'import_from': 'https://apps.openstack.org/excellent-image',
            'import_from_format': 'qcow2',
        },
        'message': '',
        'owner': str(uuid.uuid4()),
        'result': {
            'image_id': str(uuid.uuid4()),
        },
        'schema': '/v2/schemas/task',
        'status': random.choice(
            [
                'pending',
                'processing',
                'success',
                'failure',
            ]
        ),
        # though not documented, the API only allows 'import'
        # https://github.com/openstack/glance/blob/24.0.0/glance/api/v2/tasks.py#L186-L190
        'type': 'import',
        'updated_at': '2016-06-29T16:13:07Z',
    }

    # Overwrite default attributes if there are some attributes set
    task_info.update(attrs)

    return task.Task(**task_info)


def create_tasks(attrs=None, count=2):
    """Create multiple fake tasks.

    :param attrs: A dictionary with all attributes of Task
    :type attrs: dict
    :param count: The number of tasks to be faked
    :type count: int
    :return: A list of fake Task objects
    :rtype: list
    """
    tasks = []
    for n in range(0, count):
        tasks.append(create_one_task(attrs))

    return tasks


class FakeMetadefNamespaceClient:
    def __init__(self, **kwargs):
        self.create_metadef_namespace = mock.Mock()
        self.delete_metadef_namespace = mock.Mock()
        self.metadef_namespaces = mock.Mock()
        self.get_metadef_namespace = mock.Mock()
        self.update_metadef_namespace = mock.Mock()

        self.auth_token = kwargs['token']
        self.management_url = kwargs['endpoint']
        self.version = 2.0


class TestMetadefNamespaces(utils.TestCommand):
    def setUp(self):
        super().setUp()

        self.app.client_manager.image = FakeMetadefNamespaceClient(
            endpoint=fakes.AUTH_URL,
            token=fakes.AUTH_TOKEN,
        )

        self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client(
            endpoint=fakes.AUTH_URL,
            token=fakes.AUTH_TOKEN,
        )


def create_one_metadef_namespace(attrs=None):
    """Create a fake MetadefNamespace member.

    :param attrs: A dictionary with all attributes of metadef_namespace member
    :type attrs: dict
    :return: a list of MetadefNamespace objects
    :rtype: list of `metadef_namespace.MetadefNamespace`
    """
    attrs = attrs or {}

    metadef_namespace_list = {
        'created_at': '2022-08-17T11:30:22Z',
        'display_name': 'Flavor Quota',
        'namespace': 'OS::Compute::Quota',
        'owner': 'admin',
        # 'resource_type_associations': ['OS::Nova::Flavor'],
        # The part that receives the list type factor is not implemented.
        'visibility': 'public',
    }

    # Overwrite default attributes if there are some attributes set
    metadef_namespace_list.update(attrs)
    return metadef_namespace.MetadefNamespace(**metadef_namespace_list)


def create_one_resource_type(attrs=None):
    """Create a fake MetadefResourceType member.

    :param attrs: A dictionary with all attributes of
        metadef_resource_type member
    :type attrs: dict
    :return: a fake MetadefResourceType object
    :rtype: A `metadef_resource_type.MetadefResourceType`
    """
    attrs = attrs or {}

    metadef_resource_type_info = {
        'name': 'OS::Compute::Quota',
        'properties_target': 'image',
    }

    metadef_resource_type_info.update(attrs)
    return metadef_resource_type.MetadefResourceType(
        **metadef_resource_type_info
    )


def create_resource_types(attrs=None, count=2):
    """Create multiple fake resource types.

    :param attrs: A dictionary with all attributes of
        metadef_resource_type member
    :type attrs: dict
    :return: A list of fake MetadefResourceType objects
    :rtype: list
    """
    metadef_resource_types = []
    for n in range(0, count):
        metadef_resource_types.append(create_one_resource_type(attrs))

    return metadef_resource_types