~ubuntu-branches/ubuntu/quantal/nova/quantal-proposed

« back to all changes in this revision

Viewing changes to nova/tests/fake_network.py

  • Committer: Package Import Robot
  • Author(s): Adam Gandelman, Adam Gandelman, Chuck Short
  • Date: 2012-08-27 15:37:18 UTC
  • mfrom: (1.1.60)
  • Revision ID: package-import@ubuntu.com-20120827153718-lj8er44eqqz1gsrj
Tags: 2012.2~rc1~20120827.15815-0ubuntu1
[ Adam Gandelman ]
* New upstream release.

[ Chuck Short ]
* debian/patches/0001-Update-tools-hacking-for-pep8-1.2-and-
  beyond.patch: Dropped we dont run pep8 tests anymore.
* debian/control: Drop pep8 build depends
* debian/*.upstart.in: Make sure we transition correctly from runlevel
  1 to 2. (LP: #820694)

Show diffs side-by-side

added added

removed removed

Lines of Context:
15
15
# License for the specific language governing permissions and limitations
16
16
# under the License.
17
17
 
 
18
from nova.compute import api as compute_api
 
19
from nova.compute import manager as compute_manager
18
20
import nova.context
19
21
from nova import db
20
22
from nova import exception
21
23
from nova import flags
22
24
from nova.network import manager as network_manager
 
25
from nova.network import model as network_model
23
26
from nova.network import nova_ipam_lib
24
27
from nova import utils
25
28
 
133
136
        def virtual_interface_get_all(self, context):
134
137
            return self.vifs
135
138
 
136
 
        def instance_get_id_to_uuid_mapping(self, context, ids):
137
 
            # NOTE(jkoelker): This is just here until we can rely on UUIDs
138
 
            mapping = {}
139
 
            for id in ids:
140
 
                mapping[id] = str(utils.gen_uuid())
141
 
            return mapping
142
 
 
143
139
        def fixed_ips_by_virtual_interface(self, context, vif_id):
144
140
            return [ip for ip in self.fixed_ips
145
141
                    if ip['virtual_interface_id'] == vif_id]
373
369
    if func is None:
374
370
        func = get_instance_nw_info
375
371
    stubs.Set(nova.network.API, 'get_instance_nw_info', func)
 
372
 
 
373
 
 
374
_real_functions = {}
 
375
 
 
376
 
 
377
def set_stub_network_methods(stubs):
 
378
    global _real_functions
 
379
    cm = compute_manager.ComputeManager
 
380
    if not _real_functions:
 
381
        _real_functions = {
 
382
                '_get_instance_nw_info': cm._get_instance_nw_info,
 
383
                '_allocate_network': cm._allocate_network,
 
384
                '_deallocate_network': cm._deallocate_network}
 
385
 
 
386
    def fake_networkinfo(*args, **kwargs):
 
387
        return network_model.NetworkInfo()
 
388
 
 
389
    stubs.Set(cm, '_get_instance_nw_info', fake_networkinfo)
 
390
    stubs.Set(cm, '_allocate_network', fake_networkinfo)
 
391
    stubs.Set(cm, '_deallocate_network', lambda *args, **kwargs: None)
 
392
 
 
393
 
 
394
def unset_stub_network_methods(stubs):
 
395
    global _real_functions
 
396
    if _real_functions:
 
397
        cm = compute_manager.ComputeManager
 
398
        for name in _real_functions:
 
399
            stubs.Set(cm, name, _real_functions[name])
 
400
 
 
401
 
 
402
def stub_compute_with_ips(stubs):
 
403
    orig_get = compute_api.API.get
 
404
    orig_get_all = compute_api.API.get_all
 
405
 
 
406
    def fake_get(*args, **kwargs):
 
407
        return _get_instances_with_cached_ips(orig_get, *args, **kwargs)
 
408
 
 
409
    def fake_get_all(*args, **kwargs):
 
410
        return _get_instances_with_cached_ips(orig_get_all, *args, **kwargs)
 
411
 
 
412
    stubs.Set(compute_api.API, 'get', fake_get)
 
413
    stubs.Set(compute_api.API, 'get_all', fake_get_all)
 
414
 
 
415
 
 
416
def _get_fake_cache():
 
417
    def _ip(ip, fixed=True, floats=None):
 
418
        ip_dict = {'address': ip, 'type': 'fixed'}
 
419
        if not fixed:
 
420
            ip_dict['type'] = 'floating'
 
421
        if fixed and floats:
 
422
            ip_dict['floating_ips'] = [_ip(f, fixed=False) for f in floats]
 
423
        return ip_dict
 
424
 
 
425
    info = [{'address': 'aa:bb:cc:dd:ee:ff',
 
426
             'id': 1,
 
427
             'network': {'bridge': 'br0',
 
428
                         'id': 1,
 
429
                         'label': 'private',
 
430
                         'subnets': [{'cidr': '192.168.0.0/24',
 
431
                                      'ips': [_ip('192.168.0.3')]}]}}]
 
432
    if FLAGS.use_ipv6:
 
433
        ipv6_addr = 'fe80:b33f::a8bb:ccff:fedd:eeff'
 
434
        info[0]['network']['subnets'].append({'cidr': 'fe80:b33f::/64',
 
435
                                              'ips': [_ip(ipv6_addr)]})
 
436
    return info
 
437
 
 
438
 
 
439
def _get_instances_with_cached_ips(orig_func, *args, **kwargs):
 
440
    """Kludge the cache into instance(s) without having to create DB
 
441
    entries
 
442
    """
 
443
    instances = orig_func(*args, **kwargs)
 
444
    if isinstance(instances, list):
 
445
        for instance in instances:
 
446
            instance['info_cache'] = {'network_info': _get_fake_cache()}
 
447
    else:
 
448
        instances['info_cache'] = {'network_info': _get_fake_cache()}
 
449
    return instances