← Back to branch summary

~gandelman-a/ubuntu/precise/nova/UCA_2012.2.1

« back to all changes in this revision

Viewing changes to nova/tests/scheduler/test_scheduler.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Chuck Short, Adam Gandelman
  • Date: 2012-06-22 12:39:57 UTC
  • mfrom: (1.1.57)
  • Revision ID: package-import@ubuntu.com-20120622123957-hbzwg84nt9rqwg8r
Tags: 2012.2~f2~20120621.14517-0ubuntu1
[ Chuck Short ]
* New upstream version.

[ Adam Gandelman ]
* debian/rules: Temporarily disable test suite while blocking
  tests are investigated. 
* debian/patches/kombu_tests_timeout.patch: Dropped.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/scheduler/test_scheduler.py
19
19
Tests For Scheduler
20
20
"""
21
21
 
22
 
import json
23
 
 
24
22
from nova.compute import api as compute_api
25
23
from nova.compute import power_state
 
24
from nova.compute import rpcapi as compute_rpcapi
26
25
from nova.compute import vm_states
27
26
from nova import context
28
27
from nova import db
29
28
from nova import exception
30
29
from nova import flags
31
 
from nova import notifications
32
 
from nova import rpc
33
 
from nova.rpc import common as rpc_common
 
30
from nova.openstack.common import jsonutils
 
31
from nova.openstack.common import rpc
 
32
from nova.openstack.common.rpc import common as rpc_common
 
33
from nova.openstack.common import timeutils
34
34
from nova.scheduler import driver
35
35
from nova.scheduler import manager
36
36
from nova import test
468
468
                block_migration=block_migration,
469
469
                disk_over_commit=disk_over_commit)
470
470
 
 
471
    def _check_shared_storage(self, dest, instance, check_result):
 
472
        tmp_filename = 'test-filename'
 
473
        rpc.queue_get_for(self.context, FLAGS.compute_topic,
 
474
                dest).AndReturn('dest_queue')
 
475
        rpc.call(self.context, 'dest_queue',
 
476
                {'method': 'create_shared_storage_test_file',
 
477
                 'args': {},
 
478
                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION}, None
 
479
                ).AndReturn(tmp_filename)
 
480
        rpc.queue_get_for(self.context, FLAGS.compute_topic,
 
481
                instance['host']).AndReturn('src_queue')
 
482
        rpc.call(self.context, 'src_queue',
 
483
                {'method': 'check_shared_storage_test_file',
 
484
                 'args': {'filename': tmp_filename},
 
485
                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION}, None
 
486
                ).AndReturn(check_result)
 
487
        rpc.queue_get_for(self.context, FLAGS.compute_topic,
 
488
                dest).AndReturn('dest_queue')
 
489
        rpc.cast(self.context, 'dest_queue',
 
490
                {'method': 'cleanup_shared_storage_test_file',
 
491
                 'args': {'filename': tmp_filename},
 
492
                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION})
 
493
 
471
494
    def test_live_migration_all_checks_pass(self):
472
495
        """Test live migration when all checks pass."""
473
496
 
476
499
        self.mox.StubOutWithMock(db, 'service_get_all_compute_by_host')
477
500
        self.mox.StubOutWithMock(self.driver, '_get_compute_info')
478
501
        self.mox.StubOutWithMock(db, 'instance_get_all_by_host')
479
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
502
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
480
503
        self.mox.StubOutWithMock(rpc, 'call')
481
504
        self.mox.StubOutWithMock(rpc, 'cast')
482
505
        self.mox.StubOutWithMock(db, 'instance_update_and_get_original')
504
527
        # assert_compute_node_has_enough_disk()
505
528
        self.driver._get_compute_info(self.context, dest,
506
529
                'disk_available_least').AndReturn(1025)
507
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
 
530
        rpc.queue_get_for(self.context, FLAGS.compute_topic,
508
531
                instance['host']).AndReturn('src_queue1')
509
 
        rpc.call(self.context, 'src_queue1',
510
 
                {'method': 'get_instance_disk_info',
511
 
                 'args': {'instance_name': instance['name']}}).AndReturn(
512
 
                        json.dumps([{'disk_size': 1024 * (1024 ** 3)}]))
513
 
 
514
 
        # Common checks (shared storage ok, same hypervisor,e tc)
515
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
516
 
                dest).AndReturn('dest_queue')
517
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
518
 
                instance['host']).AndReturn('src_queue')
519
 
        tmp_filename = 'test-filename'
520
 
        rpc.call(self.context, 'dest_queue',
521
 
                {'method': 'create_shared_storage_test_file'}
522
 
                ).AndReturn(tmp_filename)
523
 
        rpc.call(self.context, 'src_queue',
524
 
                {'method': 'check_shared_storage_test_file',
525
 
                 'args': {'filename': tmp_filename}}).AndReturn(False)
526
 
        rpc.cast(self.context, 'dest_queue',
527
 
                {'method': 'cleanup_shared_storage_test_file',
528
 
                 'args': {'filename': tmp_filename}})
 
532
        instance_disk_info_msg = {
 
533
            'method': 'get_instance_disk_info',
 
534
            'args': {
 
535
                'instance_name': instance['name'],
 
536
            },
 
537
            'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION,
 
538
        }
 
539
        instance_disk_info = [{'disk_size': 1024 * (1024 ** 3)}]
 
540
        rpc.call(self.context,
 
541
                 'src_queue1',
 
542
                 instance_disk_info_msg,
 
543
                 None).AndReturn(jsonutils.dumps(instance_disk_info))
 
544
 
 
545
        # Common checks (shared storage ok, same hypervisor, etc)
 
546
        self._check_shared_storage(dest, instance, False)
 
547
 
529
548
        db.service_get_all_compute_by_host(self.context, dest).AndReturn(
530
549
                [{'compute_node': [{'hypervisor_type': 'xen',
531
550
                                    'hypervisor_version': 1}]}])
535
554
                    [{'compute_node': [{'hypervisor_type': 'xen',
536
555
                                        'hypervisor_version': 1,
537
556
                                        'cpu_info': 'fake_cpu_info'}]}])
538
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
 
557
        rpc.queue_get_for(self.context, FLAGS.compute_topic,
539
558
                dest).AndReturn('dest_queue')
540
559
        rpc.call(self.context, 'dest_queue',
541
560
                {'method': 'compare_cpu',
542
 
                 'args': {'cpu_info': 'fake_cpu_info'}}).AndReturn(True)
 
561
                 'args': {'cpu_info': 'fake_cpu_info'},
 
562
                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION}, None
 
563
                ).AndReturn(True)
543
564
 
544
565
        db.instance_update_and_get_original(self.context, instance['id'],
545
566
                {"vm_state": vm_states.MIGRATING}).AndReturn(
696
717
                'assert_compute_node_has_enough_memory')
697
718
        self.mox.StubOutWithMock(self.driver, '_get_compute_info')
698
719
        self.mox.StubOutWithMock(db, 'instance_get_all_by_host')
699
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
720
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
700
721
        self.mox.StubOutWithMock(rpc, 'call')
701
722
 
702
723
        dest = 'fake_host2'
717
738
        # Not enough disk
718
739
        self.driver._get_compute_info(self.context, dest,
719
740
                'disk_available_least').AndReturn(1023)
720
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
 
741
        rpc.queue_get_for(self.context, FLAGS.compute_topic,
721
742
                instance['host']).AndReturn('src_queue')
722
 
        rpc.call(self.context, 'src_queue',
723
 
                {'method': 'get_instance_disk_info',
724
 
                 'args': {'instance_name': instance['name']}}).AndReturn(
725
 
                        json.dumps([{'disk_size': 1024 * (1024 ** 3)}]))
 
743
        instance_disk_info_msg = {
 
744
            'method': 'get_instance_disk_info',
 
745
            'args': {
 
746
                'instance_name': instance['name'],
 
747
            },
 
748
            'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION,
 
749
        }
 
750
        instance_disk_info = [{'disk_size': 1024 * (1024 ** 3)}]
 
751
        rpc.call(self.context,
 
752
                 'src_queue',
 
753
                 instance_disk_info_msg,
 
754
                 None).AndReturn(jsonutils.dumps(instance_disk_info))
726
755
 
727
756
        self.mox.ReplayAll()
728
757
        self.assertRaises(exception.MigrationError,
737
766
        self.mox.StubOutWithMock(db, 'instance_get')
738
767
        self.mox.StubOutWithMock(self.driver, '_live_migration_src_check')
739
768
        self.mox.StubOutWithMock(self.driver, '_live_migration_dest_check')
740
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
769
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
741
770
        self.mox.StubOutWithMock(rpc, 'call')
742
771
        self.mox.StubOutWithMock(rpc, 'cast')
743
772
 
751
780
        self.driver._live_migration_dest_check(self.context, instance,
752
781
                dest, block_migration, disk_over_commit)
753
782
 
754
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
755
 
                dest).AndReturn('dest_queue')
756
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
757
 
                instance['host']).AndReturn('src_queue')
758
 
        tmp_filename = 'test-filename'
759
 
        rpc.call(self.context, 'dest_queue',
760
 
                {'method': 'create_shared_storage_test_file'}
761
 
                ).AndReturn(tmp_filename)
762
 
        rpc.call(self.context, 'src_queue',
763
 
                {'method': 'check_shared_storage_test_file',
764
 
                 'args': {'filename': tmp_filename}}).AndReturn(False)
765
 
        rpc.cast(self.context, 'dest_queue',
766
 
                {'method': 'cleanup_shared_storage_test_file',
767
 
                 'args': {'filename': tmp_filename}})
 
783
        self._check_shared_storage(dest, instance, False)
768
784
 
769
785
        self.mox.ReplayAll()
770
786
        self.assertRaises(exception.InvalidSharedStorage,
779
795
        self.mox.StubOutWithMock(db, 'instance_get')
780
796
        self.mox.StubOutWithMock(self.driver, '_live_migration_src_check')
781
797
        self.mox.StubOutWithMock(self.driver, '_live_migration_dest_check')
782
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
798
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
783
799
        self.mox.StubOutWithMock(rpc, 'call')
784
800
        self.mox.StubOutWithMock(rpc, 'cast')
785
801
 
793
809
        self.driver._live_migration_dest_check(self.context, instance,
794
810
                dest, block_migration, disk_over_commit)
795
811
 
796
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
797
 
                dest).AndReturn('dest_queue')
798
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
799
 
                instance['host']).AndReturn('src_queue')
800
 
        tmp_filename = 'test-filename'
801
 
        rpc.call(self.context, 'dest_queue',
802
 
                {'method': 'create_shared_storage_test_file'}
803
 
                ).AndReturn(tmp_filename)
804
 
        rpc.call(self.context, 'src_queue',
805
 
                {'method': 'check_shared_storage_test_file',
806
 
                 'args': {'filename': tmp_filename}}).AndReturn(False)
807
 
        rpc.cast(self.context, 'dest_queue',
808
 
                {'method': 'cleanup_shared_storage_test_file',
809
 
                 'args': {'filename': tmp_filename}})
 
812
        self._check_shared_storage(dest, instance, False)
810
813
 
811
814
        self.mox.ReplayAll()
812
815
        self.assertRaises(exception.InvalidSharedStorage,
819
822
        self.mox.StubOutWithMock(db, 'instance_get')
820
823
        self.mox.StubOutWithMock(self.driver, '_live_migration_src_check')
821
824
        self.mox.StubOutWithMock(self.driver, '_live_migration_dest_check')
822
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
825
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
823
826
        self.mox.StubOutWithMock(rpc, 'call')
824
827
        self.mox.StubOutWithMock(rpc, 'cast')
825
828
        self.mox.StubOutWithMock(db, 'service_get_all_compute_by_host')
834
837
        self.driver._live_migration_dest_check(self.context, instance,
835
838
                dest, block_migration, disk_over_commit)
836
839
 
837
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
838
 
                dest).AndReturn('dest_queue')
839
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
840
 
                instance['host']).AndReturn('src_queue')
841
 
        tmp_filename = 'test-filename'
842
 
        rpc.call(self.context, 'dest_queue',
843
 
                {'method': 'create_shared_storage_test_file'}
844
 
                ).AndReturn(tmp_filename)
845
 
        rpc.call(self.context, 'src_queue',
846
 
                {'method': 'check_shared_storage_test_file',
847
 
                 'args': {'filename': tmp_filename}}).AndReturn(True)
848
 
        rpc.cast(self.context, 'dest_queue',
849
 
                {'method': 'cleanup_shared_storage_test_file',
850
 
                 'args': {'filename': tmp_filename}})
 
840
        self._check_shared_storage(dest, instance, True)
 
841
 
851
842
        db.service_get_all_compute_by_host(self.context, dest).AndReturn(
852
843
                [{'compute_node': [{'hypervisor_type': 'xen',
853
844
                                    'hypervisor_version': 1}]}])
868
859
        self.mox.StubOutWithMock(db, 'instance_get')
869
860
        self.mox.StubOutWithMock(self.driver, '_live_migration_src_check')
870
861
        self.mox.StubOutWithMock(self.driver, '_live_migration_dest_check')
871
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
862
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
872
863
        self.mox.StubOutWithMock(rpc, 'call')
873
864
        self.mox.StubOutWithMock(rpc, 'cast')
874
865
        self.mox.StubOutWithMock(db, 'service_get_all_compute_by_host')
883
874
        self.driver._live_migration_dest_check(self.context, instance,
884
875
                dest, block_migration, disk_over_commit)
885
876
 
886
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
887
 
                dest).AndReturn('dest_queue')
888
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
889
 
                instance['host']).AndReturn('src_queue')
890
 
        tmp_filename = 'test-filename'
891
 
        rpc.call(self.context, 'dest_queue',
892
 
                {'method': 'create_shared_storage_test_file'}
893
 
                ).AndReturn(tmp_filename)
894
 
        rpc.call(self.context, 'src_queue',
895
 
                {'method': 'check_shared_storage_test_file',
896
 
                 'args': {'filename': tmp_filename}}).AndReturn(True)
897
 
        rpc.cast(self.context, 'dest_queue',
898
 
                {'method': 'cleanup_shared_storage_test_file',
899
 
                 'args': {'filename': tmp_filename}})
 
877
        self._check_shared_storage(dest, instance, True)
 
878
 
900
879
        db.service_get_all_compute_by_host(self.context, dest).AndReturn(
901
880
                [{'compute_node': [{'hypervisor_type': 'xen',
902
881
                                    'hypervisor_version': 1}]}])
916
895
        self.mox.StubOutWithMock(db, 'instance_get')
917
896
        self.mox.StubOutWithMock(self.driver, '_live_migration_src_check')
918
897
        self.mox.StubOutWithMock(self.driver, '_live_migration_dest_check')
919
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
898
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
920
899
        self.mox.StubOutWithMock(rpc, 'call')
921
900
        self.mox.StubOutWithMock(rpc, 'cast')
922
901
        self.mox.StubOutWithMock(db, 'service_get_all_compute_by_host')
931
910
        self.driver._live_migration_dest_check(self.context, instance,
932
911
                dest, block_migration, disk_over_commit)
933
912
 
934
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
935
 
                dest).AndReturn('dest_queue')
936
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
937
 
                instance['host']).AndReturn('src_queue')
938
 
        tmp_filename = 'test-filename'
939
 
        rpc.call(self.context, 'dest_queue',
940
 
                {'method': 'create_shared_storage_test_file'}
941
 
                ).AndReturn(tmp_filename)
942
 
        rpc.call(self.context, 'src_queue',
943
 
                {'method': 'check_shared_storage_test_file',
944
 
                 'args': {'filename': tmp_filename}}).AndReturn(True)
945
 
        rpc.cast(self.context, 'dest_queue',
946
 
                {'method': 'cleanup_shared_storage_test_file',
947
 
                 'args': {'filename': tmp_filename}})
 
913
        self._check_shared_storage(dest, instance, True)
 
914
 
948
915
        db.service_get_all_compute_by_host(self.context, dest).AndReturn(
949
916
                [{'compute_node': [{'hypervisor_type': 'xen',
950
917
                                    'hypervisor_version': 1}]}])
953
920
                    [{'compute_node': [{'hypervisor_type': 'xen',
954
921
                                        'hypervisor_version': 1,
955
922
                                        'cpu_info': 'fake_cpu_info'}]}])
956
 
        db.queue_get_for(self.context, FLAGS.compute_topic,
 
923
        rpc.queue_get_for(self.context, FLAGS.compute_topic,
957
924
                dest).AndReturn('dest_queue')
958
925
        rpc.call(self.context, 'dest_queue',
959
926
                {'method': 'compare_cpu',
960
 
                 'args': {'cpu_info': 'fake_cpu_info'}}).AndRaise(
961
 
                         rpc_common.RemoteError())
 
927
                 'args': {'cpu_info': 'fake_cpu_info'},
 
928
                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION}, None
 
929
                ).AndRaise(rpc_common.RemoteError())
962
930
 
963
931
        self.mox.ReplayAll()
964
932
        self.assertRaises(rpc_common.RemoteError,
1016
984
                       'extra_arg': 'meow'}
1017
985
        queue = 'fake_queue'
1018
986
 
1019
 
        self.mox.StubOutWithMock(utils, 'utcnow')
 
987
        self.mox.StubOutWithMock(timeutils, 'utcnow')
1020
988
        self.mox.StubOutWithMock(db, 'volume_update')
1021
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
989
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1022
990
        self.mox.StubOutWithMock(rpc, 'cast')
1023
991
 
1024
 
        utils.utcnow().AndReturn('fake-now')
 
992
        timeutils.utcnow().AndReturn('fake-now')
1025
993
        db.volume_update(self.context, 31337,
1026
994
                {'host': host, 'scheduled_at': 'fake-now'})
1027
 
        db.queue_get_for(self.context, 'volume', host).AndReturn(queue)
 
995
        rpc.queue_get_for(self.context, 'volume', host).AndReturn(queue)
1028
996
        rpc.cast(self.context, queue,
1029
997
                {'method': method,
1030
998
                 'args': fake_kwargs})
1039
1007
        fake_kwargs = {'extra_arg': 'meow'}
1040
1008
        queue = 'fake_queue'
1041
1009
 
1042
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
1010
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1043
1011
        self.mox.StubOutWithMock(rpc, 'cast')
1044
1012
 
1045
 
        db.queue_get_for(self.context, 'volume', host).AndReturn(queue)
 
1013
        rpc.queue_get_for(self.context, 'volume', host).AndReturn(queue)
1046
1014
        rpc.cast(self.context, queue,
1047
1015
                {'method': method,
1048
1016
                 'args': fake_kwargs})
1057
1025
        fake_kwargs = {'extra_arg': 'meow'}
1058
1026
        queue = 'fake_queue'
1059
1027
 
1060
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
1028
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1061
1029
        self.mox.StubOutWithMock(rpc, 'cast')
1062
1030
 
1063
 
        db.queue_get_for(self.context, 'volume', host).AndReturn(queue)
 
1031
        rpc.queue_get_for(self.context, 'volume', host).AndReturn(queue)
1064
1032
        rpc.cast(self.context, queue,
1065
1033
                {'method': method,
1066
1034
                 'args': fake_kwargs})
1076
1044
                       'extra_arg': 'meow'}
1077
1045
        queue = 'fake_queue'
1078
1046
 
1079
 
        self.mox.StubOutWithMock(utils, 'utcnow')
 
1047
        self.mox.StubOutWithMock(timeutils, 'utcnow')
1080
1048
        self.mox.StubOutWithMock(db, 'instance_update')
1081
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
1049
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1082
1050
        self.mox.StubOutWithMock(rpc, 'cast')
1083
1051
 
1084
 
        utils.utcnow().AndReturn('fake-now')
 
1052
        timeutils.utcnow().AndReturn('fake-now')
1085
1053
        db.instance_update(self.context, 31337,
1086
1054
                {'host': host, 'scheduled_at': 'fake-now'})
1087
 
        db.queue_get_for(self.context, 'compute', host).AndReturn(queue)
 
1055
        rpc.queue_get_for(self.context, 'compute', host).AndReturn(queue)
1088
1056
        rpc.cast(self.context, queue,
1089
1057
                {'method': method,
1090
1058
                 'args': fake_kwargs})
1099
1067
        fake_kwargs = {'extra_arg': 'meow'}
1100
1068
        queue = 'fake_queue'
1101
1069
 
1102
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
1070
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1103
1071
        self.mox.StubOutWithMock(rpc, 'cast')
1104
1072
 
1105
 
        db.queue_get_for(self.context, 'compute', host).AndReturn(queue)
 
1073
        rpc.queue_get_for(self.context, 'compute', host).AndReturn(queue)
1106
1074
        rpc.cast(self.context, queue,
1107
1075
                {'method': method,
1108
1076
                 'args': fake_kwargs})
1117
1085
        fake_kwargs = {'extra_arg': 'meow'}
1118
1086
        queue = 'fake_queue'
1119
1087
 
1120
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
1088
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1121
1089
        self.mox.StubOutWithMock(rpc, 'cast')
1122
1090
 
1123
 
        db.queue_get_for(self.context, 'compute', host).AndReturn(queue)
 
1091
        rpc.queue_get_for(self.context, 'compute', host).AndReturn(queue)
1124
1092
        rpc.cast(self.context, queue,
1125
1093
                {'method': method,
1126
1094
                 'args': fake_kwargs})
1135
1103
        fake_kwargs = {'extra_arg': 'meow'}
1136
1104
        queue = 'fake_queue'
1137
1105
 
1138
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
1106
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1139
1107
        self.mox.StubOutWithMock(rpc, 'cast')
1140
1108
 
1141
 
        db.queue_get_for(self.context, 'network', host).AndReturn(queue)
 
1109
        rpc.queue_get_for(self.context, 'network', host).AndReturn(queue)
1142
1110
        rpc.cast(self.context, queue,
1143
1111
                {'method': method,
1144
1112
                 'args': fake_kwargs})
1193
1161
        topic = 'unknown'
1194
1162
        queue = 'fake_queue'
1195
1163
 
1196
 
        self.mox.StubOutWithMock(db, 'queue_get_for')
 
1164
        self.mox.StubOutWithMock(rpc, 'queue_get_for')
1197
1165
        self.mox.StubOutWithMock(rpc, 'cast')
1198
1166
 
1199
 
        db.queue_get_for(self.context, topic, host).AndReturn(queue)
 
1167
        rpc.queue_get_for(self.context, topic, host).AndReturn(queue)
1200
1168
        rpc.cast(self.context, queue,
1201
1169
                {'method': method,
1202
1170
                 'args': fake_kwargs})