path: root/nova/tests/xenapi/stubs.py

# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright (c) 2010 Citrix Systems, 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.

"""Stubouts, mocks and fixtures for the test suite"""

import contextlib
import random
import sys

from nova.openstack.common import jsonutils
from nova import test
import nova.tests.image.fake
from nova.virt.xenapi import driver as xenapi_conn
from nova.virt.xenapi import fake
from nova.virt.xenapi import vm_utils
from nova.virt.xenapi import vmops


def stubout_firewall_driver(stubs, conn):

    def fake_none(self, *args):
        return

    vmops = conn._vmops
    stubs.Set(vmops.firewall_driver, 'prepare_instance_filter', fake_none)
    stubs.Set(vmops.firewall_driver, 'instance_filter_exists', fake_none)


def stubout_instance_snapshot(stubs):
    def fake_fetch_image(context, session, instance, image, type):
        return {'root': dict(uuid=_make_fake_vdi(), file=None),
                'kernel': dict(uuid=_make_fake_vdi(), file=None),
                'ramdisk': dict(uuid=_make_fake_vdi(), file=None)}

    stubs.Set(vm_utils, '_fetch_image', fake_fetch_image)

    def fake_wait_for_vhd_coalesce(*args):
        #TODO(sirp): Should we actually fake out the data here
        return "fakeparent", "fakebase"

    stubs.Set(vm_utils, '_wait_for_vhd_coalesce', fake_wait_for_vhd_coalesce)


def stubout_session(stubs, cls, product_version=(5, 6, 2), **opt_args):
    """Stubs out methods from XenAPISession"""
    stubs.Set(xenapi_conn.XenAPISession, '_create_session',
              lambda s, url: cls(url, **opt_args))
    stubs.Set(xenapi_conn.XenAPISession, '_get_product_version',
              lambda s: product_version)


def stubout_get_this_vm_uuid(stubs):
    def f():
        vms = [rec['uuid'] for ref, rec
               in fake.get_all_records('VM').iteritems()
               if rec['is_control_domain']]
        return vms[0]
    stubs.Set(vm_utils, 'get_this_vm_uuid', f)


def stubout_image_service_download(stubs):
    def fake_download(*args, **kwargs):
        pass
    stubs.Set(nova.tests.image.fake._FakeImageService,
        'download', fake_download)


def stubout_stream_disk(stubs):
    def fake_stream_disk(*args, **kwargs):
        pass
    stubs.Set(vm_utils, '_stream_disk', fake_stream_disk)


def stubout_is_vdi_pv(stubs):
    def f(_1):
        return False
    stubs.Set(vm_utils, '_is_vdi_pv', f)


def stubout_determine_is_pv_objectstore(stubs):
    """Assumes VMs stu have PV kernels"""

    def f(*args):
        return False
    stubs.Set(vm_utils, '_determine_is_pv_objectstore', f)


def stubout_is_snapshot(stubs):
    """ Always returns true
        xenapi fake driver does not create vmrefs for snapshots """

    def f(*args):
        return True
    stubs.Set(vm_utils, 'is_snapshot', f)


def stubout_lookup_image(stubs):
    """Simulates a failure in lookup image."""
    def f(_1, _2, _3, _4):
        raise Exception("Test Exception raised by fake lookup_image")
    stubs.Set(vm_utils, 'lookup_image', f)


def stubout_fetch_disk_image(stubs, raise_failure=False):
    """Simulates a failure in fetch image_glance_disk."""

    def _fake_fetch_disk_image(context, session, instance, image,
                                      image_type):
        if raise_failure:
            raise fake.Failure("Test Exception raised by "
                               "fake fetch_image_glance_disk")
        elif image_type == vm_utils.ImageType.KERNEL:
            filename = "kernel"
        elif image_type == vm_utils.ImageType.RAMDISK:
            filename = "ramdisk"
        else:
            filename = "unknown"

        vdi_type = vm_utils.ImageType.to_string(image_type)
        return {vdi_type: dict(uuid=None, file=filename)}

    stubs.Set(vm_utils, '_fetch_disk_image', _fake_fetch_disk_image)


def stubout_create_vm(stubs):
    """Simulates a failure in create_vm."""

    def f(*args):
        raise fake.Failure("Test Exception raised by " +
                           "fake create_vm")
    stubs.Set(vm_utils, 'create_vm', f)


def _make_fake_vdi():
    sr_ref = fake.get_all('SR')[0]
    vdi_ref = fake.create_vdi('', sr_ref)
    vdi_rec = fake.get_record('VDI', vdi_ref)
    return vdi_rec['uuid']


class FakeSessionForVMTests(fake.SessionBase):
    """ Stubs out a XenAPISession for VM tests """

    _fake_iptables_save_output = ("# Generated by iptables-save v1.4.10 on "
                                  "Sun Nov  6 22:49:02 2011\n"
                                  "*filter\n"
                                  ":INPUT ACCEPT [0:0]\n"
                                  ":FORWARD ACCEPT [0:0]\n"
                                  ":OUTPUT ACCEPT [0:0]\n"
                                  "COMMIT\n"
                                  "# Completed on Sun Nov  6 22:49:02 2011\n")

    def __init__(self, uri):
        super(FakeSessionForVMTests, self).__init__(uri)

    def host_call_plugin(self, _1, _2, plugin, method, _5):
        if (plugin, method) == ('glance', 'download_vhd'):
            root_uuid = _make_fake_vdi()
            return jsonutils.dumps(dict(root=dict(uuid=root_uuid)))
        elif (plugin, method) == ("xenhost", "iptables_config"):
            return fake.as_json(out=self._fake_iptables_save_output,
                                err='')
        else:
            return (super(FakeSessionForVMTests, self).
                    host_call_plugin(_1, _2, plugin, method, _5))

    def host_call_plugin_swap(self, _1, _2, plugin, method, _5):
        if (plugin, method) == ('glance', 'download_vhd'):
            root_uuid = _make_fake_vdi()
            swap_uuid = _make_fake_vdi()
            return jsonutils.dumps(dict(root=dict(uuid=root_uuid),
                                        swap=dict(uuid=swap_uuid)))
        else:
            return (super(FakeSessionForVMTests, self).
                    host_call_plugin(_1, _2, plugin, method, _5))

    def VM_start(self, _1, ref, _2, _3):
        vm = fake.get_record('VM', ref)
        if vm['power_state'] != 'Halted':
            raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted',
                                vm['power_state']])
        vm['power_state'] = 'Running'
        vm['is_a_template'] = False
        vm['is_control_domain'] = False
        vm['domid'] = random.randrange(1, 1 << 16)
        return vm

    def VM_start_on(self, _1, vm_ref, host_ref, _2, _3):
        vm_rec = self.VM_start(_1, vm_ref, _2, _3)
        vm_rec['resident_on'] = host_ref

    def VM_snapshot(self, session_ref, vm_ref, label):
        status = "Running"
        template_vm_ref = fake.create_vm(label, status, is_a_template=True,
            is_control_domain=False)

        sr_ref = "fakesr"
        template_vdi_ref = fake.create_vdi(label, sr_ref, read_only=True)

        template_vbd_ref = fake.create_vbd(template_vm_ref, template_vdi_ref)
        return template_vm_ref

    def SR_scan(self, session_ref, sr_ref):
        pass


class FakeSessionForFirewallTests(FakeSessionForVMTests):
    """ Stubs out a XenApi Session for doing IPTable Firewall tests """

    def __init__(self, uri, test_case=None):
        super(FakeSessionForFirewallTests, self).__init__(uri)
        if hasattr(test_case, '_in_filter_rules'):
            self._in_filter_rules = test_case._in_filter_rules
        if hasattr(test_case, '_in6_filter_rules'):
            self._in6_filter_rules = test_case._in6_filter_rules
        if hasattr(test_case, '_in_nat_rules'):
            self._in_nat_rules = test_case._in_nat_rules
        self._test_case = test_case

    def host_call_plugin(self, _1, _2, plugin, method, args):
        """Mock method four host_call_plugin to be used in unit tests
           for the dom0 iptables Firewall drivers for XenAPI

        """
        if plugin == "xenhost" and method == "iptables_config":
            # The command to execute is a json-encoded list
            cmd_args = args.get('cmd_args', None)
            cmd = jsonutils.loads(cmd_args)
            if not cmd:
                ret_str = ''
            else:
                output = ''
                process_input = args.get('process_input', None)
                if cmd == ['ip6tables-save', '-t', 'filter']:
                    output = '\n'.join(self._in6_filter_rules)
                if cmd == ['iptables-save', '-t', 'filter']:
                    output = '\n'.join(self._in_filter_rules)
                if cmd == ['iptables-save', '-t', 'nat']:
                    output = '\n'.join(self._in_nat_rules)
                if cmd == ['iptables-restore', ]:
                    lines = process_input.split('\n')
                    if '*filter' in lines:
                        if self._test_case is not None:
                            self._test_case._out_rules = lines
                        output = '\n'.join(lines)
                if cmd == ['ip6tables-restore', ]:
                    lines = process_input.split('\n')
                    if '*filter' in lines:
                        output = '\n'.join(lines)
                ret_str = fake.as_json(out=output, err='')
        return ret_str


def stub_out_vm_methods(stubs):
    def fake_acquire_bootlock(self, vm):
        pass

    def fake_release_bootlock(self, vm):
        pass

    def fake_generate_ephemeral(*args):
        pass

    def fake_wait_for_device(dev):
        pass

    stubs.Set(vmops.VMOps, "_acquire_bootlock", fake_acquire_bootlock)
    stubs.Set(vmops.VMOps, "_release_bootlock", fake_release_bootlock)
    stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)
    stubs.Set(vm_utils, '_wait_for_device', fake_wait_for_device)


class FakeSessionForVolumeTests(fake.SessionBase):
    """ Stubs out a XenAPISession for Volume tests """
    def __init__(self, uri):
        super(FakeSessionForVolumeTests, self).__init__(uri)

    def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
                      _6, _7, _8, _9, _10, _11):
        valid_vdi = False
        refs = fake.get_all('VDI')
        for ref in refs:
            rec = fake.get_record('VDI', ref)
            if rec['uuid'] == uuid:
                valid_vdi = True
        if not valid_vdi:
            raise fake.Failure([['INVALID_VDI', 'session', self._session]])


class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests):
    """ Stubs out a XenAPISession for Volume tests: it injects failures """
    def __init__(self, uri):
        super(FakeSessionForVolumeFailedTests, self).__init__(uri)

    def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
                      _6, _7, _8, _9, _10, _11):
        # This is for testing failure
        raise fake.Failure([['INVALID_VDI', 'session', self._session]])

    def PBD_unplug(self, _1, ref):
        rec = fake.get_record('PBD', ref)
        rec['currently-attached'] = False

    def SR_forget(self, _1, ref):
        pass


def stub_out_migration_methods(stubs):
    @contextlib.contextmanager
    def fake_snapshot_attached_here(session, instance, vm_ref, label):
        yield ['bar', 'foo']

    def fake_move_disks(self, instance, disk_info):
        vdi_ref = fake.create_vdi(instance['name'], 'fake')
        vdi_rec = fake.get_record('VDI', vdi_ref)
        vdi_rec['other_config']['nova_disk_type'] = 'root'
        return {'uuid': vdi_rec['uuid'], 'ref': vdi_ref}

    def fake_get_vdi(session, vm_ref):
        vdi_ref = fake.create_vdi('derp', 'herp')
        vdi_rec = session.call_xenapi("VDI.get_record", vdi_ref)
        return vdi_ref, {'uuid': vdi_rec['uuid'], }

    def fake_sr(session, *args):
        pass

    def fake_get_sr_path(*args):
        return "fake"

    def fake_destroy(*args, **kwargs):
        pass

    def fake_generate_ephemeral(*args):
        pass

    stubs.Set(vmops.VMOps, '_destroy', fake_destroy)
    stubs.Set(vm_utils, 'move_disks', fake_move_disks)
    stubs.Set(vm_utils, 'scan_default_sr', fake_sr)
    stubs.Set(vm_utils, '_scan_sr', fake_sr)
    stubs.Set(vm_utils, 'snapshot_attached_here', fake_snapshot_attached_here)
    stubs.Set(vm_utils, 'get_vdi_for_vm_safely', fake_get_vdi)
    stubs.Set(vm_utils, 'get_sr_path', fake_get_sr_path)
    stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)


class XenAPITestBase(test.TestCase):
    def setUp(self):
        super(XenAPITestBase, self).setUp()

        self.orig_XenAPI = sys.modules.get('XenAPI')
        sys.modules['XenAPI'] = fake

        fake.reset()

    def tearDown(self):
        if self.orig_XenAPI is not None:
            sys.modules['XenAPI'] = self.orig_XenAPI
            self.orig_XenAPI = None
        else:
            sys.modules.pop('XenAPI')

        super(XenAPITestBase, self).tearDown()