path: root/nova/tests/db/fakes.py

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

# Copyright (c) 2011 X.commerce, a business unit of eBay Inc.
# Copyright 2010 OpenStack Foundation
# All Rights Reserved.
#
#    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."""

from nova import db
from nova import exception


class FakeModel(object):
    """Stubs out for model."""
    def __init__(self, values):
        self.values = values

    def __getattr__(self, name):
        return self.values[name]

    def __getitem__(self, key):
        if key in self.values:
            return self.values[key]
        else:
            raise NotImplementedError()

    def __repr__(self):
        return '<FakeModel: %s>' % self.values

    def get(self, name):
        return self.values[name]


def stub_out(stubs, funcs):
    """Set the stubs in mapping in the db api."""
    for func in funcs:
        func_name = '_'.join(func.__name__.split('_')[1:])
        stubs.Set(db, func_name, func)
        stubs.Set(db.sqlalchemy.api, func_name, func)


fixed_ip_fields = {'id': 0,
                   'network_id': 0,
                   'address': '192.168.0.100',
                   'instance': False,
                   'instance_uuid': 'eb57d790-fc60-4119-a51a-f2b0913bdc93',
                   'allocated': False,
                   'virtual_interface_id': 0,
                   'virtual_interface': None,
                   'floating_ips': []}


def stub_out_db_network_api(stubs):
    network_fields = {'id': 0,
                      'cidr': '192.168.0.0/24',
                      'netmask': '255.255.255.0',
                      'cidr_v6': 'dead:beef::/64',
                      'netmask_v6': '64',
                      'project_id': 'fake',
                      'label': 'fake',
                      'gateway': '192.168.0.1',
                      'bridge': 'fa0',
                      'bridge_interface': 'fake_fa0',
                      'broadcast': '192.168.0.255',
                      'gateway_v6': 'dead:beef::1',
                      'dns': '192.168.0.1',
                      'vlan': None,
                      'host': None,
                      'injected': False,
                      'vpn_public_address': '192.168.0.2'}

    flavor_fields = {'id': 0,
                     'rxtx_cap': 3}

    floating_ip_fields = {'id': 0,
                          'address': '192.168.1.100',
                          'fixed_ip_id': None,
                          'fixed_ip': None,
                          'project_id': None,
                          'pool': 'nova',
                          'auto_assigned': False}

    virtual_interface_fields = {'id': 0,
                                'address': 'DE:AD:BE:EF:00:00',
                                'network_id': 0,
                                'instance_id': 0,
                                'network': FakeModel(network_fields)}

    fixed_ips = [fixed_ip_fields]
    floating_ips = [floating_ip_fields]
    virtual_interfacees = [virtual_interface_fields]
    networks = [network_fields]

    def fake_floating_ip_allocate_address(context, project_id, pool):
        ips = filter(lambda i: i['fixed_ip_id'] is None and
                               i['project_id'] is None and
                               i['pool'] == pool,
                     floating_ips)
        if not ips:
            raise exception.NoMoreFloatingIps()
        ips[0]['project_id'] = project_id
        return FakeModel(ips[0])

    def fake_floating_ip_deallocate(context, address):
        ips = filter(lambda i: i['address'] == address,
                     floating_ips)
        if ips:
            ips[0]['project_id'] = None
            ips[0]['auto_assigned'] = False

    def fake_floating_ip_disassociate(context, address):
        ips = filter(lambda i: i['address'] == address,
                     floating_ips)
        if ips:
            fixed_ip_address = None
            if ips[0]['fixed_ip']:
                fixed_ip_address = ips[0]['fixed_ip']['address']
            ips[0]['fixed_ip'] = None
            ips[0]['host'] = None
            return fixed_ip_address

    def fake_floating_ip_fixed_ip_associate(context, floating_address,
                                            fixed_address, host):
        float = filter(lambda i: i['address'] == floating_address,
                       floating_ips)
        fixed = filter(lambda i: i['address'] == fixed_address,
                       fixed_ips)
        if float and fixed:
            float[0]['fixed_ip'] = fixed[0]
            float[0]['fixed_ip_id'] = fixed[0]['id']
            float[0]['host'] = host

    def fake_floating_ip_get_all_by_host(context, host):
        # TODO(jkoelker): Once we get the patches that remove host from
        #                 the floating_ip table, we'll need to stub
        #                 this out
        pass

    def fake_floating_ip_get_by_address(context, address):
        if isinstance(address, FakeModel):
            # NOTE(tr3buchet): yo dawg, i heard you like addresses
            address = address['address']
        ips = filter(lambda i: i['address'] == address,
                     floating_ips)
        if not ips:
            raise exception.FloatingIpNotFoundForAddress(address=address)
        return FakeModel(ips[0])

    def fake_floating_ip_set_auto_assigned(contex, address):
        ips = filter(lambda i: i['address'] == address,
                     floating_ips)
        if ips:
            ips[0]['auto_assigned'] = True

    def fake_fixed_ip_associate(context, address, instance_id):
        ips = filter(lambda i: i['address'] == address,
                     fixed_ips)
        if not ips:
            raise exception.NoMoreFixedIps()
        ips[0]['instance'] = True
        ips[0]['instance_id'] = instance_id

    def fake_fixed_ip_associate_pool(context, network_id, instance_id):
        ips = filter(lambda i: (i['network_id'] == network_id or
                                i['network_id'] is None) and not i['instance'],
                     fixed_ips)
        if not ips:
            raise exception.NoMoreFixedIps()
        ips[0]['instance'] = True
        ips[0]['instance_id'] = instance_id
        return ips[0]['address']

    def fake_fixed_ip_create(context, values):
        ip = dict(fixed_ip_fields)
        ip['id'] = max([i['id'] for i in fixed_ips] or [-1]) + 1
        for key in values:
            ip[key] = values[key]
        return ip

    def fake_fixed_ip_disassociate(context, address):
        ips = filter(lambda i: i['address'] == address,
                     fixed_ips)
        if ips:
            ips[0]['instance_id'] = None
            ips[0]['instance'] = None
            ips[0]['virtual_interface'] = None
            ips[0]['virtual_interface_id'] = None

    def fake_fixed_ip_disassociate_all_by_timeout(context, host, time):
        return 0

    def fake_fixed_ip_get_all(context):
        return [FakeModel(i) for i in fixed_ips]

    def fake_fixed_ip_get_by_instance(context, instance_uuid):
        ips = filter(lambda i: i['instance_uuid'] == instance_uuid,
                     fixed_ips)
        return [FakeModel(i) for i in ips]

    def fake_fixed_ip_get_by_address(context, address):
        ips = filter(lambda i: i['address'] == address,
                     fixed_ips)
        if ips:
            return FakeModel(ips[0])

    def fake_fixed_ip_update(context, address, values):
        ips = filter(lambda i: i['address'] == address,
                     fixed_ips)
        if ips:
            for key in values:
                ips[0][key] = values[key]
                if key == 'virtual_interface_id':
                    vif = filter(lambda x: x['id'] == values[key],
                                 virtual_interfacees)
                    if not vif:
                        continue
                    fixed_ip_fields['virtual_interface'] = FakeModel(vif[0])

    def fake_instance_type_get(context, id):
        if flavor_fields['id'] == id:
            return FakeModel(flavor_fields)

    def fake_virtual_interface_create(context, values):
        vif = dict(virtual_interface_fields)
        vif['id'] = max([m['id'] for m in virtual_interfacees] or [-1]) + 1
        for key in values:
            vif[key] = values[key]
        return FakeModel(vif)

    def fake_virtual_interface_delete_by_instance(context, instance_id):
        addresses = [m for m in virtual_interfacees
                     if m['instance_id'] == instance_id]
        try:
            for address in addresses:
                virtual_interfacees.remove(address)
        except ValueError:
            pass

    def fake_virtual_interface_get_by_instance(context, instance_id):
        return [FakeModel(m) for m in virtual_interfacees
                if m['instance_id'] == instance_id]

    def fake_virtual_interface_get_by_instance_and_network(context,
                                                           instance_id,
                                                           network_id):
        vif = filter(lambda m: m['instance_id'] == instance_id and
                               m['network_id'] == network_id,
                     virtual_interfacees)
        if not vif:
            return None
        return FakeModel(vif[0])

    def fake_network_create_safe(context, values):
        net = dict(network_fields)
        net['id'] = max([n['id'] for n in networks] or [-1]) + 1
        for key in values:
            net[key] = values[key]
        return FakeModel(net)

    def fake_network_get(context, network_id):
        net = filter(lambda n: n['id'] == network_id, networks)
        if not net:
            return None
        return FakeModel(net[0])

    def fake_network_get_all(context):
        return [FakeModel(n) for n in networks]

    def fake_network_get_all_by_host(context, host):
        nets = filter(lambda n: n['host'] == host, networks)
        return [FakeModel(n) for n in nets]

    def fake_network_set_host(context, network_id, host_id):
        nets = filter(lambda n: n['id'] == network_id, networks)
        for net in nets:
            net['host'] = host_id
        return host_id

    def fake_network_update(context, network_id, values):
        nets = filter(lambda n: n['id'] == network_id, networks)
        for net in nets:
            for key in values:
                net[key] = values[key]

    def fake_project_get_networks(context, project_id):
        return [FakeModel(n) for n in networks
                if n['project_id'] == project_id]

    funcs = [fake_floating_ip_allocate_address,
             fake_floating_ip_deallocate,
             fake_floating_ip_disassociate,
             fake_floating_ip_fixed_ip_associate,
             fake_floating_ip_get_all_by_host,
             fake_floating_ip_get_by_address,
             fake_floating_ip_set_auto_assigned,
             fake_fixed_ip_associate,
             fake_fixed_ip_associate_pool,
             fake_fixed_ip_create,
             fake_fixed_ip_disassociate,
             fake_fixed_ip_disassociate_all_by_timeout,
             fake_fixed_ip_get_all,
             fake_fixed_ip_get_by_instance,
             fake_fixed_ip_get_by_address,
             fake_fixed_ip_update,
             fake_instance_type_get,
             fake_virtual_interface_create,
             fake_virtual_interface_delete_by_instance,
             fake_virtual_interface_get_by_instance,
             fake_virtual_interface_get_by_instance_and_network,
             fake_network_create_safe,
             fake_network_get,
             fake_network_get_all,
             fake_network_get_all_by_host,
             fake_network_set_host,
             fake_network_update,
             fake_project_get_networks]

    stub_out(stubs, funcs)


def stub_out_db_instance_api(stubs, injected=True):
    """Stubs out the db API for creating Instances."""

    INSTANCE_TYPES = {
        'm1.tiny': dict(id=2,
                        name='m1.tiny',
                        memory_mb=512,
                        vcpus=1,
                        vcpu_weight=None,
                        root_gb=0,
                        ephemeral_gb=10,
                        flavorid=1,
                        rxtx_factor=1.0,
                        swap=0),
        'm1.small': dict(id=5,
                         name='m1.small',
                         memory_mb=2048,
                         vcpus=1,
                         vcpu_weight=None,
                         root_gb=20,
                         ephemeral_gb=0,
                         flavorid=2,
                         rxtx_factor=1.0,
                         swap=1024),
        'm1.medium':
            dict(id=1,
                 name='m1.medium',
                 memory_mb=4096,
                 vcpus=2,
                 vcpu_weight=None,
                 root_gb=40,
                 ephemeral_gb=40,
                 flavorid=3,
                 rxtx_factor=1.0,
                 swap=0),
        'm1.large': dict(id=3,
                         name='m1.large',
                         memory_mb=8192,
                         vcpus=4,
                         vcpu_weight=None,
                         root_gb=80,
                         ephemeral_gb=80,
                         flavorid=4,
                         rxtx_factor=1.0,
                         swap=0),
        'm1.xlarge':
            dict(id=4,
                 name='m1.xlarge',
                 memory_mb=16384,
                 vcpus=8,
                 vcpu_weight=None,
                 root_gb=160,
                 ephemeral_gb=160,
                 flavorid=5,
                 rxtx_factor=1.0,
                 swap=0)}

    flat_network_fields = {'id': 'fake_flat',
                           'bridge': 'xenbr0',
                           'label': 'fake_flat_network',
                           'netmask': '255.255.255.0',
                           'cidr_v6': 'fe80::a00:0/120',
                           'netmask_v6': '120',
                           'gateway': '10.0.0.1',
                           'gateway_v6': 'fe80::a00:1',
                           'broadcast': '10.0.0.255',
                           'dns': '10.0.0.2',
                           'ra_server': None,
                           'injected': injected}

    vlan_network_fields = {'id': 'fake_vlan',
                           'bridge': 'br111',
                           'label': 'fake_vlan_network',
                           'netmask': '255.255.255.0',
                           'cidr_v6': 'fe80::a00:0/120',
                           'netmask_v6': '120',
                           'gateway': '10.0.0.1',
                           'gateway_v6': 'fe80::a00:1',
                           'broadcast': '10.0.0.255',
                           'dns': '10.0.0.2',
                           'ra_server': None,
                           'vlan': 111,
                           'injected': False}

    fixed_ip_fields = {'address': '10.0.0.3',
                       'address_v6': 'fe80::a00:3',
                       'network_id': 'fake_flat'}

    def fake_instance_type_get_all(context, inactive=0, filters=None):
        return INSTANCE_TYPES.values()

    def fake_instance_type_get_by_name(context, name):
        return INSTANCE_TYPES[name]

    def fake_instance_type_get(context, id):
        for name, inst_type in INSTANCE_TYPES.iteritems():
            if str(inst_type['id']) == str(id):
                return inst_type
        return None

    def fake_fixed_ip_get_by_instance(context, instance_id):
        return [FakeModel(fixed_ip_fields)]

    funcs = [fake_instance_type_get_all,
             fake_instance_type_get_by_name,
             fake_instance_type_get,
             fake_fixed_ip_get_by_instance]
    stub_out(stubs, funcs)