← Back to branch summary

~ubuntu-branches/ubuntu/raring/nova/raring-proposed

« back to all changes in this revision

Viewing changes to nova/tests/network/test_quantumv2.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Adam Gandelman, Chuck Short
  • Date: 2012-11-23 09:04:58 UTC
  • mfrom: (1.1.66)
  • Revision ID: package-import@ubuntu.com-20121123090458-91565o7aev1i1h71
Tags: 2013.1~g1-0ubuntu1
https://launchpad.net/bugs/1073289
https://launchpad.net/bugs/1076442
[ Adam Gandelman ]
* debian/control: Ensure novaclient is upgraded with nova,
  require python-keystoneclient >= 1:2.9.0. (LP: #1073289)
* debian/patches/{ubuntu/*, rbd-security.patch}: Dropped, applied
  upstream.
* debian/control: Add python-testtools to Build-Depends.

[ Chuck Short ]
* New upstream version.
* Refreshed debian/patches/avoid_setuptools_git_dependency.patch.
* debian/rules: FTBFS if missing binaries.
* debian/nova-scheudler.install: Add missing rabbit-queues and
  nova-rpc-zmq-receiver.
* Remove nova-volume since it doesnt exist anymore, transition to cinder-*.
* debian/rules: install apport hook in the right place.
* debian/patches/ubuntu-show-tests.patch: Display test failures.
* debian/control: Add depends on genisoimage
* debian/control: Suggest guestmount.
* debian/control: Suggest websockify. (LP: #1076442)
* debian/nova.conf: Disable nova-volume service.
* debian/control: Depend on xen-system-* rather than the hypervisor.
* debian/control, debian/mans/nova-conductor.8, debian/nova-conductor.init,
  debian/nova-conductor.install, debian/nova-conductor.logrotate
  debian/nova-conductor.manpages, debian/nova-conductor.postrm
  debian/nova-conductor.upstart.in: Add nova-conductor service.
* debian/control: Add python-fixtures as a build deps.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/network/test_quantumv2.py
15
15
#
16
16
# vim: tabstop=4 shiftwidth=4 softtabstop=4
17
17
 
 
18
import uuid
 
19
 
18
20
import mox
19
21
 
20
22
from nova import context
24
26
from nova.network.quantumv2 import api as quantumapi
25
27
from nova.openstack.common import cfg
26
28
from nova import test
27
 
from nova import utils
28
29
from quantumclient.v2_0 import client
29
30
 
30
31
 
31
 
FLAGS = cfg.CONF
 
32
CONF = cfg.CONF
 
33
 
32
34
#NOTE: Quantum client raises Exception which is discouraged by HACKING.
33
35
#      We set this variable here and use it for assertions below to avoid
34
36
#      the hacking checks until we can make quantum client throw a custom
84
86
 
85
87
 
86
88
class TestQuantumClient(test.TestCase):
87
 
    def setUp(self):
88
 
        super(TestQuantumClient, self).setUp()
89
 
 
90
89
    def test_withtoken(self):
91
90
        self.flags(quantum_url='http://anyhost/')
92
91
        self.flags(quantum_url_timeout=30)
95
94
                                            auth_token='token')
96
95
        self.mox.StubOutWithMock(client.Client, "__init__")
97
96
        client.Client.__init__(
98
 
            endpoint_url=FLAGS.quantum_url,
 
97
            endpoint_url=CONF.quantum_url,
99
98
            token=my_context.auth_token,
100
 
            timeout=FLAGS.quantum_url_timeout).AndReturn(None)
 
99
            timeout=CONF.quantum_url_timeout).AndReturn(None)
101
100
        self.mox.ReplayAll()
102
101
        quantumv2.get_client(my_context)
103
102
 
110
109
                          my_context)
111
110
 
112
111
    def test_withouttoken_keystone_not_auth(self):
113
 
        # self.flags(quantum_auth_strategy=None) fail to work
114
 
        old_quantum_auth_strategy = FLAGS.quantum_auth_strategy
115
 
        setattr(FLAGS, 'quantum_auth_strategy', None)
 
112
        self.flags(quantum_auth_strategy=None)
116
113
        self.flags(quantum_url='http://anyhost/')
117
114
        self.flags(quantum_url_timeout=30)
118
115
        my_context = context.RequestContext('userid', 'my_tenantid')
119
116
        self.mox.StubOutWithMock(client.Client, "__init__")
120
117
        client.Client.__init__(
121
 
            endpoint_url=FLAGS.quantum_url,
 
118
            endpoint_url=CONF.quantum_url,
122
119
            auth_strategy=None,
123
 
            timeout=FLAGS.quantum_url_timeout).AndReturn(None)
 
120
            timeout=CONF.quantum_url_timeout).AndReturn(None)
124
121
        self.mox.ReplayAll()
125
 
        try:
126
 
            quantumv2.get_client(my_context)
127
 
        finally:
128
 
            setattr(FLAGS, 'quantum_auth_strategy',
129
 
                    old_quantum_auth_strategy)
 
122
        quantumv2.get_client(my_context)
130
123
 
131
124
 
132
125
class TestQuantumv2(test.TestCase):
142
135
                'auth_token',
143
136
                'bff4a5a6b9eb4ea2a6efec6eefb77936')
144
137
        self.instance = {'project_id': '9d049e4b60b64716978ab415e6fbd5c0',
145
 
                         'uuid': str(utils.gen_uuid()),
 
138
                         'uuid': str(uuid.uuid4()),
146
139
                         'display_name': 'test_instance',
147
140
                         'security_groups': []}
148
141
        self.nets1 = [{'id': 'my_netid1',
199
192
                                  'gateway_ip': '10.0.2.1',
200
193
                                  'dns_nameservers': ['8.8.2.1', '8.8.2.2']})
201
194
 
 
195
        self.fip_pool = {'id': '4fdbfd74-eaf8-4884-90d9-00bd6f10c2d3',
 
196
                         'name': 'ext_net',
 
197
                         'router:external': True,
 
198
                         'tenant_id': 'admin_tenantid'}
 
199
        self.fip_pool_nova = {'id': '435e20c3-d9f1-4f1b-bee5-4611a1dd07db',
 
200
                              'name': 'nova',
 
201
                              'router:external': True,
 
202
                              'tenant_id': 'admin_tenantid'}
 
203
        self.fip_unassociated = {'tenant_id': 'my_tenantid',
 
204
                                 'id': 'fip_id1',
 
205
                                 'floating_ip_address': '172.24.4.227',
 
206
                                 'floating_network_id': self.fip_pool['id'],
 
207
                                 'port_id': None,
 
208
                                 'fixed_ip_address': None,
 
209
                                 'router_id': None}
 
210
        fixed_ip_address = self.port_data2[1]['fixed_ips'][0]['ip_address']
 
211
        self.fip_associated = {'tenant_id': 'my_tenantid',
 
212
                               'id': 'fip_id2',
 
213
                               'floating_ip_address': '172.24.4.228',
 
214
                               'floating_network_id': self.fip_pool['id'],
 
215
                               'port_id': self.port_data2[1]['id'],
 
216
                               'fixed_ip_address': fixed_ip_address,
 
217
                               'router_id': 'router_id1'}
 
218
 
202
219
    def tearDown(self):
203
220
        try:
204
221
            self.mox.UnsetStubs()
205
222
            self.mox.VerifyAll()
206
223
        finally:
207
 
            FLAGS.reset()
 
224
            CONF.reset()
 
225
        super(TestQuantumv2, self).tearDown()
208
226
 
209
227
    def _verify_nw_info(self, nw_inf, index=0):
210
228
        id_suffix = index + 1
329
347
                    self.moxed_client.show_port(port_id).AndReturn(
330
348
                        {'port': {'id': 'my_portid1',
331
349
                         'network_id': 'my_netid1'}})
332
 
                    req_net_ids.append('my_netid1')
333
350
                    ports['my_netid1'] = self.port_data1[0]
334
351
                    id = 'my_netid1'
335
352
                else:
336
353
                    fixed_ips[id] = fixed_ip
337
354
                req_net_ids.append(id)
 
355
            expected_network_order = req_net_ids
 
356
        else:
 
357
            expected_network_order = [n['id'] for n in nets]
338
358
        search_ids = [net['id'] for net in nets if net['id'] in req_net_ids]
339
359
 
340
360
        mox_list_network_params = dict(tenant_id=self.instance['project_id'],
341
361
                                       shared=False)
342
362
        if search_ids:
343
 
            mox_list_network_params['id'] = search_ids
 
363
            mox_list_network_params['id'] = mox.SameElementsAs(search_ids)
344
364
        self.moxed_client.list_networks(
345
365
            **mox_list_network_params).AndReturn({'networks': nets})
346
366
 
347
367
        mox_list_network_params = dict(shared=True)
348
368
        if search_ids:
349
 
            mox_list_network_params['id'] = search_ids
 
369
            mox_list_network_params['id'] = mox.SameElementsAs(search_ids)
350
370
        self.moxed_client.list_networks(
351
371
            **mox_list_network_params).AndReturn({'networks': []})
352
372
 
353
 
        for network in nets:
 
373
        for net_id in expected_network_order:
354
374
            port_req_body = {
355
375
                'port': {
356
376
                    'device_id': self.instance['uuid'],
357
377
                    'device_owner': 'compute:nova',
358
378
                },
359
379
            }
360
 
            port = ports.get(network['id'], None)
 
380
            port = ports.get(net_id, None)
361
381
            if port:
362
382
                port_id = port['id']
363
383
                self.moxed_client.update_port(port_id,
365
385
                                              ).AndReturn(
366
386
                                                  {'port': port})
367
387
            else:
368
 
                fixed_ip = fixed_ips.get(network['id'])
 
388
                fixed_ip = fixed_ips.get(net_id)
369
389
                if fixed_ip:
370
390
                    port_req_body['port']['fixed_ip'] = fixed_ip
371
 
                port_req_body['port']['network_id'] = network['id']
 
391
                port_req_body['port']['network_id'] = net_id
372
392
                port_req_body['port']['admin_state_up'] = True
373
393
                port_req_body['port']['tenant_id'] = \
374
394
                    self.instance['project_id']
388
408
 
389
409
    def test_allocate_for_instance_with_requested_networks(self):
390
410
        # specify only first and last network
391
 
        requested_networks = [(net['id'], None, None)
392
 
                              for net in (self.nets3[0], self.nets3[-1])]
 
411
        requested_networks = [
 
412
            (net['id'], None, None)
 
413
            for net in (self.nets3[1], self.nets3[0], self.nets3[2])]
393
414
        self._allocate_for_instance(net_idx=3,
394
415
                                    requested_networks=requested_networks)
395
416
 
621
642
        # specify only first and last network
622
643
        req_ids = [net['id'] for net in (self.nets3[0], self.nets3[-1])]
623
644
        self._get_available_networks(prv_nets, pub_nets, req_ids)
 
645
 
 
646
    def test_get_floating_ip_pools(self):
 
647
        api = quantumapi.API()
 
648
        search_opts = {'router:external': True}
 
649
        self.moxed_client.list_networks(**search_opts).\
 
650
            AndReturn({'networks': [self.fip_pool, self.fip_pool_nova]})
 
651
        self.mox.ReplayAll()
 
652
        pools = api.get_floating_ip_pools(self.context)
 
653
        expected = [{'name': self.fip_pool['name']},
 
654
                    {'name': self.fip_pool_nova['name']}]
 
655
        self.assertEqual(expected, pools)
 
656
 
 
657
    def _get_expected_fip_model(self, fip_data, idx=0):
 
658
        expected = {'id': fip_data['id'],
 
659
                    'address': fip_data['floating_ip_address'],
 
660
                    'pool': self.fip_pool['name'],
 
661
                    'project_id': fip_data['tenant_id'],
 
662
                    'fixed_ip_id': fip_data['port_id'],
 
663
                    'fixed_ip':
 
664
                        {'address': fip_data['fixed_ip_address']},
 
665
                    'instance': ({'uuid': self.port_data2[idx]['device_id']}
 
666
                                 if fip_data['port_id']
 
667
                                 else None)}
 
668
        return expected
 
669
 
 
670
    def _test_get_floating_ip(self, fip_data, idx=0, by_address=False):
 
671
        api = quantumapi.API()
 
672
        fip_id = fip_data['id']
 
673
        net_id = fip_data['floating_network_id']
 
674
        address = fip_data['floating_ip_address']
 
675
        if by_address:
 
676
            self.moxed_client.list_floatingips(floating_ip_address=address).\
 
677
                AndReturn({'floatingips': [fip_data]})
 
678
        else:
 
679
            self.moxed_client.show_floatingip(fip_id).\
 
680
                AndReturn({'floatingip': fip_data})
 
681
        self.moxed_client.show_network(net_id).\
 
682
            AndReturn({'network': self.fip_pool})
 
683
        if fip_data['port_id']:
 
684
            self.moxed_client.show_port(fip_data['port_id']).\
 
685
                AndReturn({'port': self.port_data2[idx]})
 
686
        self.mox.ReplayAll()
 
687
 
 
688
        expected = self._get_expected_fip_model(fip_data, idx)
 
689
 
 
690
        if by_address:
 
691
            fip = api.get_floating_ip_by_address(self.context, address)
 
692
        else:
 
693
            fip = api.get_floating_ip(self.context, fip_id)
 
694
        self.assertEqual(expected, fip)
 
695
 
 
696
    def test_get_floating_ip_unassociated(self):
 
697
        self._test_get_floating_ip(self.fip_unassociated, idx=0)
 
698
 
 
699
    def test_get_floating_ip_associated(self):
 
700
        self._test_get_floating_ip(self.fip_associated, idx=1)
 
701
 
 
702
    def test_get_floating_ip_by_address(self):
 
703
        self._test_get_floating_ip(self.fip_unassociated, idx=0,
 
704
                                   by_address=True)
 
705
 
 
706
    def test_get_floating_ip_by_address_associated(self):
 
707
        self._test_get_floating_ip(self.fip_associated, idx=1,
 
708
                                   by_address=True)
 
709
 
 
710
    def test_get_floating_ip_by_address_not_found(self):
 
711
        api = quantumapi.API()
 
712
        address = self.fip_unassociated['floating_ip_address']
 
713
        self.moxed_client.list_floatingips(floating_ip_address=address).\
 
714
            AndReturn({'floatingips': []})
 
715
        self.mox.ReplayAll()
 
716
        self.assertRaises(exception.FloatingIpNotFoundForAddress,
 
717
                          api.get_floating_ip_by_address,
 
718
                          self.context, address)
 
719
 
 
720
    def test_get_floating_ip_by_address_multiple_found(self):
 
721
        api = quantumapi.API()
 
722
        address = self.fip_unassociated['floating_ip_address']
 
723
        self.moxed_client.list_floatingips(floating_ip_address=address).\
 
724
            AndReturn({'floatingips': [self.fip_unassociated] * 2})
 
725
        self.mox.ReplayAll()
 
726
        self.assertRaises(exception.FloatingIpMultipleFoundForAddress,
 
727
                          api.get_floating_ip_by_address,
 
728
                          self.context, address)
 
729
 
 
730
    def test_get_floating_ips_by_project(self):
 
731
        api = quantumapi.API()
 
732
        project_id = self.context.project_id
 
733
        self.moxed_client.list_floatingips(tenant_id=project_id).\
 
734
            AndReturn({'floatingips': [self.fip_unassociated,
 
735
                                       self.fip_associated]})
 
736
        search_opts = {'router:external': True}
 
737
        self.moxed_client.list_networks(**search_opts).\
 
738
            AndReturn({'networks': [self.fip_pool, self.fip_pool_nova]})
 
739
        self.moxed_client.list_ports(tenant_id=project_id).\
 
740
                AndReturn({'ports': self.port_data2})
 
741
        self.mox.ReplayAll()
 
742
 
 
743
        expected = [self._get_expected_fip_model(self.fip_unassociated),
 
744
                    self._get_expected_fip_model(self.fip_associated, idx=1)]
 
745
        fips = api.get_floating_ips_by_project(self.context)
 
746
        self.assertEqual(expected, fips)
 
747
 
 
748
    def _test_get_instance_id_by_floating_address(self, fip_data,
 
749
                                                  associated=False):
 
750
        api = quantumapi.API()
 
751
        address = fip_data['floating_ip_address']
 
752
        self.moxed_client.list_floatingips(floating_ip_address=address).\
 
753
                AndReturn({'floatingips': [fip_data]})
 
754
        if associated:
 
755
            self.moxed_client.show_port(fip_data['port_id']).\
 
756
                AndReturn({'port': self.port_data2[1]})
 
757
        self.mox.ReplayAll()
 
758
 
 
759
        if associated:
 
760
            expected = self.port_data2[1]['device_id']
 
761
        else:
 
762
            expected = None
 
763
        fip = api.get_instance_id_by_floating_address(self.context, address)
 
764
        self.assertEqual(expected, fip)
 
765
 
 
766
    def test_get_instance_id_by_floating_address(self):
 
767
        self._test_get_instance_id_by_floating_address(self.fip_unassociated)
 
768
 
 
769
    def test_get_instance_id_by_floating_address_associated(self):
 
770
        self._test_get_instance_id_by_floating_address(self.fip_associated,
 
771
                                                       associated=True)
 
772
 
 
773
    def test_allocate_floating_ip(self):
 
774
        api = quantumapi.API()
 
775
        pool_name = self.fip_pool['name']
 
776
        pool_id = self.fip_pool['id']
 
777
        search_opts = {'router:external': True,
 
778
                       'fields': 'id',
 
779
                       'name': pool_name}
 
780
        self.moxed_client.list_networks(**search_opts).\
 
781
            AndReturn({'networks': [self.fip_pool]})
 
782
        self.moxed_client.create_floatingip(
 
783
            {'floatingip': {'floating_network_id': pool_id}}).\
 
784
            AndReturn({'floatingip': self.fip_unassociated})
 
785
        self.mox.ReplayAll()
 
786
        fip = api.allocate_floating_ip(self.context, 'ext_net')
 
787
        self.assertEqual(fip, self.fip_unassociated['floating_ip_address'])
 
788
 
 
789
    def test_allocate_floating_ip_with_pool_id(self):
 
790
        api = quantumapi.API()
 
791
        pool_name = self.fip_pool['name']
 
792
        pool_id = self.fip_pool['id']
 
793
        search_opts = {'router:external': True,
 
794
                       'fields': 'id',
 
795
                       'id': pool_id}
 
796
        self.moxed_client.list_networks(**search_opts).\
 
797
            AndReturn({'networks': [self.fip_pool]})
 
798
        self.moxed_client.create_floatingip(
 
799
            {'floatingip': {'floating_network_id': pool_id}}).\
 
800
            AndReturn({'floatingip': self.fip_unassociated})
 
801
        self.mox.ReplayAll()
 
802
        fip = api.allocate_floating_ip(self.context, pool_id)
 
803
        self.assertEqual(fip, self.fip_unassociated['floating_ip_address'])
 
804
 
 
805
    def test_allocate_floating_ip_with_default_pool(self):
 
806
        api = quantumapi.API()
 
807
        pool_name = self.fip_pool_nova['name']
 
808
        pool_id = self.fip_pool_nova['id']
 
809
        search_opts = {'router:external': True,
 
810
                       'fields': 'id',
 
811
                       'name': pool_name}
 
812
        self.moxed_client.list_networks(**search_opts).\
 
813
            AndReturn({'networks': [self.fip_pool_nova]})
 
814
        self.moxed_client.create_floatingip(
 
815
            {'floatingip': {'floating_network_id': pool_id}}).\
 
816
            AndReturn({'floatingip': self.fip_unassociated})
 
817
        self.mox.ReplayAll()
 
818
        fip = api.allocate_floating_ip(self.context)
 
819
        self.assertEqual(fip, self.fip_unassociated['floating_ip_address'])
 
820
 
 
821
    def test_release_floating_ip(self):
 
822
        api = quantumapi.API()
 
823
        address = self.fip_unassociated['floating_ip_address']
 
824
        fip_id = self.fip_unassociated['id']
 
825
 
 
826
        self.moxed_client.list_floatingips(floating_ip_address=address).\
 
827
            AndReturn({'floatingips': [self.fip_unassociated]})
 
828
        self.moxed_client.delete_floatingip(fip_id)
 
829
        self.mox.ReplayAll()
 
830
        api.release_floating_ip(self.context, address)
 
831
 
 
832
    def test_release_floating_ip_associated(self):
 
833
        api = quantumapi.API()
 
834
        address = self.fip_associated['floating_ip_address']
 
835
        fip_id = self.fip_associated['id']
 
836
 
 
837
        self.moxed_client.list_floatingips(floating_ip_address=address).\
 
838
            AndReturn({'floatingips': [self.fip_associated]})
 
839
        self.mox.ReplayAll()
 
840
        self.assertRaises(exception.FloatingIpAssociated,
 
841
                          api.release_floating_ip, self.context, address)
 
842
 
 
843
    def _setup_mock_for_refresh_cache(self, api):
 
844
        nw_info = self.mox.CreateMock(model.NetworkInfo)
 
845
        nw_info.json()
 
846
        self.mox.StubOutWithMock(api, '_get_instance_nw_info')
 
847
        api._get_instance_nw_info(mox.IgnoreArg(), self.instance).\
 
848
            AndReturn(nw_info)
 
849
        self.mox.StubOutWithMock(api.db, 'instance_info_cache_update')
 
850
        api.db.instance_info_cache_update(mox.IgnoreArg(),
 
851
                                          self.instance['uuid'],
 
852
                                          mox.IgnoreArg())
 
853
 
 
854
    def test_associate_floating_ip(self):
 
855
        api = quantumapi.API()
 
856
        address = self.fip_associated['floating_ip_address']
 
857
        fixed_address = self.fip_associated['fixed_ip_address']
 
858
        fip_id = self.fip_associated['id']
 
859
 
 
860
        search_opts = {'device_owner': 'compute:nova',
 
861
                       'device_id': self.instance['uuid']}
 
862
        self.moxed_client.list_ports(**search_opts).\
 
863
            AndReturn({'ports': [self.port_data2[1]]})
 
864
        self.moxed_client.list_floatingips(floating_ip_address=address).\
 
865
            AndReturn({'floatingips': [self.fip_associated]})
 
866
        self.moxed_client.update_floatingip(
 
867
            fip_id, {'floatingip': {'port_id': self.fip_associated['port_id'],
 
868
                                    'fixed_ip_address': fixed_address}})
 
869
        self._setup_mock_for_refresh_cache(api)
 
870
 
 
871
        self.mox.ReplayAll()
 
872
        api.associate_floating_ip(self.context, self.instance,
 
873
                                  address, fixed_address)
 
874
 
 
875
    def test_associate_floating_ip_not_found_fixed_ip(self):
 
876
        api = quantumapi.API()
 
877
        address = self.fip_associated['floating_ip_address']
 
878
        fixed_address = self.fip_associated['fixed_ip_address']
 
879
        fip_id = self.fip_associated['id']
 
880
 
 
881
        search_opts = {'device_owner': 'compute:nova',
 
882
                       'device_id': self.instance['uuid']}
 
883
        self.moxed_client.list_ports(**search_opts).\
 
884
            AndReturn({'ports': [self.port_data2[0]]})
 
885
 
 
886
        self.mox.ReplayAll()
 
887
        self.assertRaises(exception.FixedIpNotFoundForAddress,
 
888
                          api.associate_floating_ip, self.context,
 
889
                          self.instance, address, fixed_address)
 
890
 
 
891
    def test_disassociate_floating_ip(self):
 
892
        api = quantumapi.API()
 
893
        address = self.fip_associated['floating_ip_address']
 
894
        fip_id = self.fip_associated['id']
 
895
 
 
896
        self.moxed_client.list_floatingips(floating_ip_address=address).\
 
897
            AndReturn({'floatingips': [self.fip_associated]})
 
898
        self.moxed_client.update_floatingip(
 
899
            fip_id, {'floatingip': {'port_id': None}})
 
900
        self._setup_mock_for_refresh_cache(api)
 
901
 
 
902
        self.mox.ReplayAll()
 
903
        api.disassociate_floating_ip(self.context, self.instance, address)
 
904
 
 
905
 
 
906
class TestQuantumv2ModuleMethods(test.TestCase):
 
907
    def test_ensure_requested_network_ordering_no_preference(self):
 
908
        l = [1, 2, 3]
 
909
 
 
910
        quantumapi._ensure_requested_network_ordering(
 
911
            lambda x: x,
 
912
            l,
 
913
            None)
 
914
 
 
915
    def test_ensure_requested_network_ordering_no_preference(self):
 
916
        l = [{'id': 3}, {'id': 1}, {'id': 2}]
 
917
 
 
918
        quantumapi._ensure_requested_network_ordering(
 
919
            lambda x: x['id'],
 
920
            l,
 
921
            None)
 
922
 
 
923
        self.assertEqual(l, [{'id': 3}, {'id': 1}, {'id': 2}])
 
924
 
 
925
    def test_ensure_requested_network_ordering_with_preference(self):
 
926
        l = [{'id': 3}, {'id': 1}, {'id': 2}]
 
927
 
 
928
        quantumapi._ensure_requested_network_ordering(
 
929
            lambda x: x['id'],
 
930
            l,
 
931
            [1, 2, 3])
 
932
 
 
933
        self.assertEqual(l, [{'id': 1}, {'id': 2}, {'id': 3}])