Merge "Remove redundant topic from rpc calls"

neutron/agent/dhcp_agent.py

@@ -400,8 +400,7 @@ class DhcpPluginApi(n_rpc.RpcProxy):
         """Make a remote process call to retrieve all network info."""
         networks = self.call(self.context,
                              self.make_msg('get_active_networks_info',
-                                           host=self.host),
+                                           host=self.host))
-                             topic=self.topic)
         return [dhcp.NetModel(self.use_namespaces, n) for n in networks]
 
     def get_network_info(self, network_id):
@@ -409,8 +408,7 @@ class DhcpPluginApi(n_rpc.RpcProxy):
         network = self.call(self.context,
                             self.make_msg('get_network_info',
                                           network_id=network_id,
-                                          host=self.host),
+                                          host=self.host))
-                            topic=self.topic)
         if network:
             return dhcp.NetModel(self.use_namespaces, network)
 
@@ -420,8 +418,7 @@ class DhcpPluginApi(n_rpc.RpcProxy):
                          self.make_msg('get_dhcp_port',
                                        network_id=network_id,
                                        device_id=device_id,
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
         if port:
             return dhcp.DictModel(port)
 
@@ -430,8 +427,7 @@ class DhcpPluginApi(n_rpc.RpcProxy):
         port = self.call(self.context,
                          self.make_msg('create_dhcp_port',
                                        port=port,
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
         if port:
             return dhcp.DictModel(port)
 
@@ -441,8 +437,7 @@ class DhcpPluginApi(n_rpc.RpcProxy):
                          self.make_msg('update_dhcp_port',
                                        port_id=port_id,
                                        port=port,
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
         if port:
             return dhcp.DictModel(port)
 
@@ -452,8 +447,7 @@ class DhcpPluginApi(n_rpc.RpcProxy):
                          self.make_msg('release_dhcp_port',
                                        network_id=network_id,
                                        device_id=device_id,
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
 
     def release_port_fixed_ip(self, network_id, device_id, subnet_id):
         """Make a remote process call to release a fixed_ip on the port."""
@@ -462,8 +456,7 @@ class DhcpPluginApi(n_rpc.RpcProxy):
                                        network_id=network_id,
                                        subnet_id=subnet_id,
                                        device_id=device_id,
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
 
 
 class NetworkCache(object):

neutron/agent/l3_agent.py

@@ -96,8 +96,7 @@ class L3PluginApi(n_rpc.RpcProxy):
         """Make a remote process call to retrieve the sync data for routers."""
         return self.call(context,
                          self.make_msg('sync_routers', host=self.host,
-                                       router_ids=router_ids),
+                                       router_ids=router_ids))
-                         topic=self.topic)
 
     def get_external_network_id(self, context):
         """Make a remote process call to retrieve the external network id.
@@ -108,8 +107,7 @@ class L3PluginApi(n_rpc.RpcProxy):
         """
         return self.call(context,
                          self.make_msg('get_external_network_id',
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
 
     def update_floatingip_statuses(self, context, router_id, fip_statuses):
         """Call the plugin update floating IPs's operational status."""
@@ -117,7 +115,6 @@ class L3PluginApi(n_rpc.RpcProxy):
                          self.make_msg('update_floatingip_statuses',
                                        router_id=router_id,
                                        fip_statuses=fip_statuses),
-                         topic=self.topic,
                          version='1.1')
 
     def get_ports_by_subnet(self, context, subnet_id):

neutron/agent/rpc.py

@@ -67,9 +67,9 @@ class PluginReportStateAPI(n_rpc.RpcProxy):
                                          agent_state},
                             time=timeutils.strtime())
         if use_call:
-            return self.call(context, msg, topic=self.topic)
+            return self.call(context, msg)
         else:
-            return self.cast(context, msg, topic=self.topic)
+            return self.cast(context, msg)
 
 
 class PluginApi(n_rpc.RpcProxy):
@@ -91,8 +91,7 @@ class PluginApi(n_rpc.RpcProxy):
     def get_device_details(self, context, device, agent_id, host=None):
         return self.call(context,
                          self.make_msg('get_device_details', device=device,
-                                       agent_id=agent_id, host=host),
+                                       agent_id=agent_id, host=host))
-                         topic=self.topic)
 
     def get_devices_details_list(self, context, devices, agent_id, host=None):
         res = []
@@ -102,7 +101,7 @@ class PluginApi(n_rpc.RpcProxy):
                                           devices=devices,
                                           agent_id=agent_id,
                                           host=host),
-                            topic=self.topic, version='1.3')
+                            version='1.3')
         except messaging.UnsupportedVersion:
             # If the server has not been upgraded yet, a DVR-enabled agent
             # may not work correctly, however it can function in 'degraded'
@@ -112,8 +111,7 @@ class PluginApi(n_rpc.RpcProxy):
             res = [
                 self.call(context,
                           self.make_msg('get_device_details', device=device,
-                                        agent_id=agent_id, host=host),
+                                        agent_id=agent_id, host=host))
-                          topic=self.topic)
                 for device in devices
             ]
         return res
@@ -121,17 +119,14 @@ class PluginApi(n_rpc.RpcProxy):
     def update_device_down(self, context, device, agent_id, host=None):
         return self.call(context,
                          self.make_msg('update_device_down', device=device,
-                                       agent_id=agent_id, host=host),
+                                       agent_id=agent_id, host=host))
-                         topic=self.topic)
 
     def update_device_up(self, context, device, agent_id, host=None):
         return self.call(context,
                          self.make_msg('update_device_up', device=device,
-                                       agent_id=agent_id, host=host),
+                                       agent_id=agent_id, host=host))
-                         topic=self.topic)
 
     def tunnel_sync(self, context, tunnel_ip, tunnel_type=None):
         return self.call(context,
                          self.make_msg('tunnel_sync', tunnel_ip=tunnel_ip,
-                                       tunnel_type=tunnel_type),
+                                       tunnel_type=tunnel_type))
-                         topic=self.topic)

neutron/agent/securitygroups_rpc.py

@@ -80,8 +80,7 @@ class SecurityGroupServerRpcApiMixin(object):
         return self.call(context,
                          self.make_msg('security_group_rules_for_devices',
                                        devices=devices),
-                         version=SG_RPC_VERSION,
+                         version=SG_RPC_VERSION)
-                         topic=self.topic)
 
 
 class SecurityGroupAgentRpcCallbackMixin(object):

neutron/api/rpc/agentnotifiers/metering_rpc_agent_api.py

@@ -68,8 +68,7 @@ class MeteringAgentNotifyAPI(n_rpc.RpcProxy):
                    'router_id': router_id})
         self.fanout_cast(
             context, self.make_msg(method,
-                                   router_id=router_id),
+                                   router_id=router_id))
-            topic=self.topic)
 
     def _notification(self, context, method, routers):
         """Notify all the agents that are hosting the routers."""
@@ -79,8 +78,7 @@ class MeteringAgentNotifyAPI(n_rpc.RpcProxy):
             plugin, constants.L3_AGENT_SCHEDULER_EXT_ALIAS):
             self._agent_notification(context, method, routers)
         else:
-            self.fanout_cast(context, self.make_msg(method, routers=routers),
+            self.fanout_cast(context, self.make_msg(method, routers=routers))
-                             topic=self.topic)
 
     def router_deleted(self, context, router_id):
         self._notification_fanout(context, 'router_deleted', router_id)

neutron/api/rpc/handlers/dvr_rpc.py

@@ -31,15 +31,13 @@ class DVRServerRpcApiMixin(object):
         return self.call(context,
                          self.make_msg('get_dvr_mac_address_by_host',
                                        host=host),
-                         version=self.DVR_RPC_VERSION,
+                         version=self.DVR_RPC_VERSION)
-                         topic=self.topic)
 
     @log.log
     def get_dvr_mac_address_list(self, context):
         return self.call(context,
                          self.make_msg('get_dvr_mac_address_list'),
-                         version=self.DVR_RPC_VERSION,
+                         version=self.DVR_RPC_VERSION)
-                         topic=self.topic)
 
     @log.log
     def get_compute_ports_on_host_by_subnet(self, context, host, subnet):
@@ -47,16 +45,14 @@ class DVRServerRpcApiMixin(object):
                          self.make_msg('get_compute_ports_on_host_by_subnet',
                                        host=host,
                                        subnet=subnet),
-                         version=self.DVR_RPC_VERSION,
+                         version=self.DVR_RPC_VERSION)
-                         topic=self.topic)
 
     @log.log
     def get_subnet_for_dvr(self, context, subnet):
         return self.call(context,
                          self.make_msg('get_subnet_for_dvr',
                                        subnet=subnet),
-                         version=self.DVR_RPC_VERSION,
+                         version=self.DVR_RPC_VERSION)
-                         topic=self.topic)
 
 
 class DVRServerRpcCallbackMixin(object):

neutron/plugins/ibm/agent/sdnve_neutron_agent.py

@@ -48,8 +48,7 @@ class SdnvePluginApi(agent_rpc.PluginApi):
 
     def sdnve_info(self, context, info):
         return self.call(context,
-                         self.make_msg('sdnve_info', info=info),
+                         self.make_msg('sdnve_info', info=info))
-                         topic=self.topic)
 
 
 class SdnveNeutronAgent(n_rpc.RpcCallback):

neutron/plugins/ryu/agent/ryu_neutron_agent.py

@@ -168,8 +168,7 @@ class RyuPluginApi(agent_rpc.PluginApi,
     def get_ofp_rest_api_addr(self, context):
         LOG.debug(_("Get Ryu rest API address"))
         return self.call(context,
-                         self.make_msg('get_ofp_rest_api'),
+                         self.make_msg('get_ofp_rest_api'))
-                         topic=self.topic)
 
 
 class RyuSecurityGroupAgent(sg_rpc.SecurityGroupAgentRpcMixin):

neutron/services/firewall/agents/firewall_agent_api.py

@@ -52,15 +52,13 @@ class FWaaSPluginApiMixin(n_rpc.RpcProxy):
         """Make a RPC to set the status of a firewall."""
         return self.call(context,
                          self.make_msg('set_firewall_status', host=self.host,
-                                       firewall_id=firewall_id, status=status),
+                                       firewall_id=firewall_id, status=status))
-                         topic=self.topic)
 
     def firewall_deleted(self, context, firewall_id):
         """Make a RPC to indicate that the firewall resources are deleted."""
         return self.call(context,
                          self.make_msg('firewall_deleted', host=self.host,
-                                       firewall_id=firewall_id),
+                                       firewall_id=firewall_id))
-                         topic=self.topic)
 
 
 class FWaaSAgentRpcCallbackMixin(object):

neutron/services/firewall/agents/l3reference/firewall_l3_agent.py

@@ -44,8 +44,7 @@ class FWaaSL3PluginApi(api.FWaaSPluginApiMixin):
 
         return self.call(context,
                          self.make_msg('get_firewalls_for_tenant',
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
 
     def get_tenants_with_firewalls(self, context, **kwargs):
         """Get all Tenants that have Firewalls configured from plugin."""
@@ -53,8 +52,7 @@ class FWaaSL3PluginApi(api.FWaaSPluginApiMixin):
 
         return self.call(context,
                          self.make_msg('get_tenants_with_firewalls',
-                                       host=self.host),
+                                       host=self.host))
-                         topic=self.topic)
 
 
 class FWaaSL3AgentRpcCallback(api.FWaaSAgentRpcCallbackMixin):

neutron/services/firewall/fwaas_plugin.py

@@ -114,24 +114,21 @@ class FirewallAgentApi(n_rpc.RpcProxy):
         return self.fanout_cast(
             context,
             self.make_msg('create_firewall', firewall=firewall,
-                          host=self.host),
+                          host=self.host)
-            topic=self.topic
         )
 
     def update_firewall(self, context, firewall):
         return self.fanout_cast(
             context,
             self.make_msg('update_firewall', firewall=firewall,
-                          host=self.host),
+                          host=self.host)
-            topic=self.topic
         )
 
     def delete_firewall(self, context, firewall):
         return self.fanout_cast(
             context,
             self.make_msg('delete_firewall', firewall=firewall,
-                          host=self.host),
+                          host=self.host)
-            topic=self.topic
         )
 
 

neutron/services/loadbalancer/agent/agent_api.py

@@ -35,22 +35,19 @@ class LbaasAgentApi(n_rpc.RpcProxy):
     def get_ready_devices(self):
         return self.call(
             self.context,
-            self.make_msg('get_ready_devices', host=self.host),
+            self.make_msg('get_ready_devices', host=self.host)
-            topic=self.topic
         )
 
     def pool_destroyed(self, pool_id):
         return self.call(
             self.context,
-            self.make_msg('pool_destroyed', pool_id=pool_id),
+            self.make_msg('pool_destroyed', pool_id=pool_id)
-            topic=self.topic
         )
 
     def pool_deployed(self, pool_id):
         return self.call(
             self.context,
-            self.make_msg('pool_deployed', pool_id=pool_id),
+            self.make_msg('pool_deployed', pool_id=pool_id)
-            topic=self.topic
         )
 
     def get_logical_device(self, pool_id):
@@ -59,30 +56,26 @@ class LbaasAgentApi(n_rpc.RpcProxy):
             self.make_msg(
                 'get_logical_device',
                 pool_id=pool_id
-            ),
+            )
-            topic=self.topic
         )
 
     def update_status(self, obj_type, obj_id, status):
         return self.call(
             self.context,
             self.make_msg('update_status', obj_type=obj_type, obj_id=obj_id,
-                          status=status),
+                          status=status)
-            topic=self.topic
         )
 
     def plug_vip_port(self, port_id):
         return self.call(
             self.context,
-            self.make_msg('plug_vip_port', port_id=port_id, host=self.host),
+            self.make_msg('plug_vip_port', port_id=port_id, host=self.host)
-            topic=self.topic
         )
 
     def unplug_vip_port(self, port_id):
         return self.call(
             self.context,
-            self.make_msg('unplug_vip_port', port_id=port_id, host=self.host),
+            self.make_msg('unplug_vip_port', port_id=port_id, host=self.host)
-            topic=self.topic
         )
 
     def update_pool_stats(self, pool_id, stats):
@@ -93,6 +86,5 @@ class LbaasAgentApi(n_rpc.RpcProxy):
                 pool_id=pool_id,
                 stats=stats,
                 host=self.host
-            ),
+            )
-            topic=self.topic
         )

neutron/services/vpn/device_drivers/cisco_ipsec.py

@@ -161,15 +161,13 @@ class CiscoCsrIPsecVpnDriverApi(n_rpc.RpcProxy):
         """
         return self.call(context,
                          self.make_msg('get_vpn_services_on_host',
-                                       host=host),
+                                       host=host))
-                         topic=self.topic)
 
     def update_status(self, context, status):
         """Update status for all VPN services and connections."""
         return self.cast(context,
                          self.make_msg('update_status',
-                                       status=status),
+                                       status=status))
-                         topic=self.topic)
 
 
 @six.add_metaclass(abc.ABCMeta)

neutron/services/vpn/device_drivers/ipsec.py

@@ -454,8 +454,7 @@ class IPsecVpnDriverApi(n_rpc.RpcProxy):
         return self.call(context,
                          self.make_msg('get_vpn_services_on_host',
                                        host=host),
-                         version=self.IPSEC_PLUGIN_VERSION,
+                         version=self.IPSEC_PLUGIN_VERSION)
-                         topic=self.topic)
 
     def update_status(self, context, status):
         """Update local status.
@@ -466,8 +465,7 @@ class IPsecVpnDriverApi(n_rpc.RpcProxy):
         return self.cast(context,
                          self.make_msg('update_status',
                                        status=status),
-                         version=self.IPSEC_PLUGIN_VERSION,
+                         version=self.IPSEC_PLUGIN_VERSION)
-                         topic=self.topic)
 
 
 @six.add_metaclass(abc.ABCMeta)

neutron/tests/unit/hyperv/test_hyperv_rpcapi.py

@@ -46,14 +46,14 @@ class rpcHyperVApiTestCase(base.BaseTestCase):
             retval = getattr(rpcapi, method)(ctxt, **kwargs)
 
         self.assertEqual(retval, expected_retval)
+        additional_args = {}
+        if topic:
+            additional_args['topic'] = topic
         if expected_version:
-            expected = [
+            additional_args['version'] = expected_version
-                mock.call(ctxt, expected_msg, topic=topic,
+        expected = [
-                          version=expected_version)]
+            mock.call(ctxt, expected_msg, **additional_args)
-        else:
+        ]
-            expected = [
-                mock.call(ctxt, expected_msg, topic=topic)
-            ]
         rpc_method_mock.assert_has_calls(expected)
 
     def test_delete_network(self):
@@ -106,7 +106,7 @@ class rpcHyperVApiTestCase(base.BaseTestCase):
     def test_device_details(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
         self._test_hyperv_neutron_api(
-            rpcapi, topics.PLUGIN,
+            rpcapi, None,
             'get_device_details', rpc_method='call',
             device='fake_device',
             agent_id='fake_agent_id',
@@ -115,7 +115,7 @@ class rpcHyperVApiTestCase(base.BaseTestCase):
     def test_devices_details_list(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
         self._test_hyperv_neutron_api(
-            rpcapi, topics.PLUGIN,
+            rpcapi, None,
             'get_devices_details_list', rpc_method='call',
             devices=['fake_device1', 'fake_device2'],
             agent_id='fake_agent_id', host='fake_host',
@@ -124,7 +124,7 @@ class rpcHyperVApiTestCase(base.BaseTestCase):
     def test_update_device_down(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
         self._test_hyperv_neutron_api(
-            rpcapi, topics.PLUGIN,
+            rpcapi, None,
             'update_device_down', rpc_method='call',
             device='fake_device',
             agent_id='fake_agent_id',
@@ -133,7 +133,7 @@ class rpcHyperVApiTestCase(base.BaseTestCase):
     def test_tunnel_sync(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
         self._test_hyperv_neutron_api(
-            rpcapi, topics.PLUGIN,
+            rpcapi, None,
             'tunnel_sync', rpc_method='call',
             tunnel_ip='fake_tunnel_ip',
             tunnel_type=None)

neutron/tests/unit/linuxbridge/test_rpcapi.py

@@ -31,7 +31,9 @@ class rpcApiTestCase(base.BaseTestCase):
                      expected_msg=None, **kwargs):
         ctxt = context.RequestContext('fake_user', 'fake_project')
         expected_retval = 'foo' if method == 'call' else None
-        expected_kwargs = {'topic': topic}
+        expected_kwargs = {}
+        if topic:
+            expected_kwargs['topic'] = topic
         if 'version' in kwargs:
             expected_kwargs['version'] = kwargs.pop('version')
         if not expected_msg:
@@ -110,7 +112,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_device_details(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_lb_api(rpcapi, topics.PLUGIN,
+        self._test_lb_api(rpcapi, None,
                           'get_device_details', rpc_method='call',
                           device='fake_device',
                           agent_id='fake_agent_id',
@@ -118,7 +120,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_devices_details_list(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_lb_api(rpcapi, topics.PLUGIN,
+        self._test_lb_api(rpcapi, None,
                           'get_devices_details_list', rpc_method='call',
                           devices=['fake_device1', 'fake_device2'],
                           agent_id='fake_agent_id', host='fake_host',
@@ -126,7 +128,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_update_device_down(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_lb_api(rpcapi, topics.PLUGIN,
+        self._test_lb_api(rpcapi, None,
                           'update_device_down', rpc_method='call',
                           device='fake_device',
                           agent_id='fake_agent_id',
@@ -134,7 +136,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_update_device_up(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_lb_api(rpcapi, topics.PLUGIN,
+        self._test_lb_api(rpcapi, None,
                           'update_device_up', rpc_method='call',
                           device='fake_device',
                           agent_id='fake_agent_id',

neutron/tests/unit/ml2/test_rpcapi.py

@@ -80,14 +80,14 @@ class RpcApiTestCase(base.BaseTestCase):
             retval = getattr(rpcapi, method)(ctxt, **kwargs)
 
         self.assertEqual(retval, expected_retval)
+        additional_args = {}
+        if topic:
+            additional_args['topic'] = topic
         if expected_version:
-            expected = [
+            additional_args['version'] = expected_version
-                mock.call(ctxt, expected_msg, topic=topic,
+        expected = [
-                          version=expected_version)]
+            mock.call(ctxt, expected_msg, **additional_args)
-        else:
+        ]
-            expected = [
-                mock.call(ctxt, expected_msg, topic=topic)
-            ]
         rpc_method_mock.assert_has_calls(expected)
 
     def test_delete_network(self):
@@ -122,7 +122,7 @@ class RpcApiTestCase(base.BaseTestCase):
 
     def test_device_details(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_rpc_api(rpcapi, topics.PLUGIN,
+        self._test_rpc_api(rpcapi, None,
                            'get_device_details', rpc_method='call',
                            device='fake_device',
                            agent_id='fake_agent_id',
@@ -130,7 +130,7 @@ class RpcApiTestCase(base.BaseTestCase):
 
     def test_devices_details_list(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_rpc_api(rpcapi, topics.PLUGIN,
+        self._test_rpc_api(rpcapi, None,
                            'get_devices_details_list', rpc_method='call',
                            devices=['fake_device1', 'fake_device2'],
                            agent_id='fake_agent_id', host='fake_host',
@@ -138,7 +138,7 @@ class RpcApiTestCase(base.BaseTestCase):
 
     def test_update_device_down(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_rpc_api(rpcapi, topics.PLUGIN,
+        self._test_rpc_api(rpcapi, None,
                            'update_device_down', rpc_method='call',
                            device='fake_device',
                            agent_id='fake_agent_id',
@@ -146,14 +146,14 @@ class RpcApiTestCase(base.BaseTestCase):
 
     def test_tunnel_sync(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_rpc_api(rpcapi, topics.PLUGIN,
+        self._test_rpc_api(rpcapi, None,
                            'tunnel_sync', rpc_method='call',
                            tunnel_ip='fake_tunnel_ip',
                            tunnel_type=None)
 
     def test_update_device_up(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_rpc_api(rpcapi, topics.PLUGIN,
+        self._test_rpc_api(rpcapi, None,
                            'update_device_up', rpc_method='call',
                            device='fake_device',
                            agent_id='fake_agent_id',

neutron/tests/unit/mlnx/test_rpcapi.py

@@ -33,7 +33,9 @@ class rpcApiTestCase(base.BaseTestCase):
                        expected_msg=None, **kwargs):
         ctxt = context.RequestContext('fake_user', 'fake_project')
         expected_retval = 'foo' if method == 'call' else None
-        expected_kwargs = {'topic': topic}
+        expected_kwargs = {}
+        if topic:
+            expected_kwargs['topic'] = topic
         if 'version' in kwargs:
             expected_kwargs['version'] = kwargs.pop('version')
         if not expected_msg:
@@ -133,7 +135,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_device_details(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_mlnx_api(rpcapi, topics.PLUGIN,
+        self._test_mlnx_api(rpcapi, None,
                             'get_device_details', rpc_method='call',
                             device='fake_device',
                             agent_id='fake_agent_id',
@@ -141,7 +143,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_devices_details_list(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_mlnx_api(rpcapi, topics.PLUGIN,
+        self._test_mlnx_api(rpcapi, None,
                             'get_devices_details_list', rpc_method='call',
                             devices=['fake_device1', 'fake_device1'],
                             agent_id='fake_agent_id', host='fake_host',
@@ -149,7 +151,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_update_device_down(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_mlnx_api(rpcapi, topics.PLUGIN,
+        self._test_mlnx_api(rpcapi, None,
                             'update_device_down', rpc_method='call',
                             device='fake_device',
                             agent_id='fake_agent_id',
@@ -157,7 +159,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_update_device_up(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_mlnx_api(rpcapi, topics.PLUGIN,
+        self._test_mlnx_api(rpcapi, None,
                             'update_device_up', rpc_method='call',
                             device='fake_device',
                             agent_id='fake_agent_id',

neutron/tests/unit/openvswitch/test_ovs_rpcapi.py

@@ -31,7 +31,9 @@ class rpcApiTestCase(base.BaseTestCase):
     def _test_ovs_api(self, rpcapi, topic, method, rpc_method, **kwargs):
         ctxt = context.RequestContext('fake_user', 'fake_project')
         expected_retval = 'foo' if method == 'call' else None
-        expected_kwargs = {'topic': topic}
+        expected_kwargs = {}
+        if topic:
+            expected_kwargs['topic'] = topic
         if 'version' in kwargs:
             expected_kwargs['version'] = kwargs.pop('version')
         expected_msg = rpcapi.make_msg(method, **kwargs)
@@ -94,7 +96,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_device_details(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_ovs_api(rpcapi, topics.PLUGIN,
+        self._test_ovs_api(rpcapi, None,
                            'get_device_details', rpc_method='call',
                            device='fake_device',
                            agent_id='fake_agent_id',
@@ -102,7 +104,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_devices_details_list(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_ovs_api(rpcapi, topics.PLUGIN,
+        self._test_ovs_api(rpcapi, None,
                            'get_devices_details_list', rpc_method='call',
                            devices=['fake_device1', 'fake_device2'],
                            agent_id='fake_agent_id', host='fake_host',
@@ -110,7 +112,7 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_update_device_down(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_ovs_api(rpcapi, topics.PLUGIN,
+        self._test_ovs_api(rpcapi, None,
                            'update_device_down', rpc_method='call',
                            device='fake_device',
                            agent_id='fake_agent_id',
@@ -118,14 +120,14 @@ class rpcApiTestCase(base.BaseTestCase):
 
     def test_tunnel_sync(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_ovs_api(rpcapi, topics.PLUGIN,
+        self._test_ovs_api(rpcapi, None,
                            'tunnel_sync', rpc_method='call',
                            tunnel_ip='fake_tunnel_ip',
                            tunnel_type=None)
 
     def test_update_device_up(self):
         rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
-        self._test_ovs_api(rpcapi, topics.PLUGIN,
+        self._test_ovs_api(rpcapi, None,
                            'update_device_up', rpc_method='call',
                            device='fake_device',
                            agent_id='fake_agent_id',

neutron/tests/unit/ryu/test_ryu_agent.py

@@ -240,7 +240,7 @@ class TestRyuPluginApi(RyuAgentTestCase):
             mock.call('get_ofp_rest_api')
         ])
         mock_call.assert_has_calls([
-            mock.call('context', 'msg', topic='topics')
+            mock.call('context', 'msg')
         ])
 
 

neutron/tests/unit/services/firewall/agents/test_firewall_agent_api.py

@@ -77,8 +77,7 @@ class TestFWaaSAgentApi(base.BaseTestCase):
 
             mock_call.assert_called_once_with(
                 mock.sentinel.context,
-                mock_make_msg.return_value,
+                mock_make_msg.return_value)
-                topic='topic')
 
     def test_firewall_deleted(self):
         with contextlib.nested(
@@ -99,5 +98,4 @@ class TestFWaaSAgentApi(base.BaseTestCase):
 
             mock_call.assert_called_once_with(
                 mock.sentinel.context,
-                mock_make_msg.return_value,
+                mock_make_msg.return_value)
-                topic='topic')

neutron/tests/unit/services/firewall/test_fwaas_plugin.py

@@ -184,8 +184,7 @@ class TestFirewallAgentApi(base.BaseTestCase):
         self.assertEqual(rv, self.mock_fanoutcast.return_value)
         self.mock_fanoutcast.assert_called_once_with(
             mock.sentinel.context,
-            self.mock_msg.return_value,
+            self.mock_msg.return_value
-            topic='topic'
         )
 
         self.mock_msg.assert_called_once_with(

neutron/tests/unit/services/loadbalancer/agent/test_api.py

@@ -41,8 +41,7 @@ class TestApiCache(base.BaseTestCase):
         self.make_msg.assert_called_once_with('get_ready_devices', host='host')
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
-            self.make_msg.return_value,
+            self.make_msg.return_value
-            topic='topic'
         )
 
     def test_get_logical_device(self):
@@ -57,8 +56,7 @@ class TestApiCache(base.BaseTestCase):
 
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
-            self.make_msg.return_value,
+            self.make_msg.return_value
-            topic='topic'
         )
 
     def test_pool_destroyed(self):
@@ -73,8 +71,7 @@ class TestApiCache(base.BaseTestCase):
 
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
-            self.make_msg.return_value,
+            self.make_msg.return_value
-            topic='topic'
         )
 
     def test_pool_deployed(self):
@@ -89,8 +86,7 @@ class TestApiCache(base.BaseTestCase):
 
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
-            self.make_msg.return_value,
+            self.make_msg.return_value
-            topic='topic'
         )
 
     def test_update_status(self):
@@ -108,7 +104,6 @@ class TestApiCache(base.BaseTestCase):
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
             self.make_msg.return_value,
-            topic='topic'
         )
 
     def test_plug_vip_port(self):
@@ -124,8 +119,7 @@ class TestApiCache(base.BaseTestCase):
 
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
-            self.make_msg.return_value,
+            self.make_msg.return_value
-            topic='topic'
         )
 
     def test_unplug_vip_port(self):
@@ -141,8 +135,7 @@ class TestApiCache(base.BaseTestCase):
 
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
-            self.make_msg.return_value,
+            self.make_msg.return_value
-            topic='topic'
         )
 
     def test_update_pool_stats(self):
@@ -159,6 +152,5 @@ class TestApiCache(base.BaseTestCase):
 
         self.mock_call.assert_called_once_with(
             mock.sentinel.context,
-            self.make_msg.return_value,
+            self.make_msg.return_value
-            topic='topic'
         )

neutron/tests/unit/services/metering/test_metering_plugin.py

@@ -118,8 +118,7 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
                              set_context=True):
                 with self.metering_label(tenant_id=self.tenant_id,
                                          set_context=True):
-                    self.mock_fanout.assert_called_with(self.ctx, expected,
+                    self.mock_fanout.assert_called_with(self.ctx, expected)
-                                                        topic=self.topic)
 
     def test_remove_metering_label_rpc_call(self):
         expected = {'args':
@@ -138,11 +137,9 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
         with self.router(tenant_id=self.tenant_id, set_context=True):
             with self.metering_label(tenant_id=self.tenant_id,
                                      set_context=True):
-                self.mock_fanout.assert_called_with(self.ctx, expected,
+                self.mock_fanout.assert_called_with(self.ctx, expected)
-                                                    topic=self.topic)
             expected['method'] = 'remove_metering_label'
-            self.mock_fanout.assert_called_with(self.ctx, expected,
+            self.mock_fanout.assert_called_with(self.ctx, expected)
-                                                topic=self.topic)
 
     def test_remove_one_metering_label_rpc_call(self):
         second_uuid = 'e27fe2df-376e-4ac7-ae13-92f050a21f84'
@@ -179,10 +176,8 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
                 self.mock_uuid.return_value = second_uuid
                 with self.metering_label(tenant_id=self.tenant_id,
                                          set_context=True):
-                    self.mock_fanout.assert_called_with(self.ctx, expected_add,
+                    self.mock_fanout.assert_called_with(self.ctx, expected_add)
-                                                        topic=self.topic)
+                self.mock_fanout.assert_called_with(self.ctx, expected_remove)
-                self.mock_fanout.assert_called_with(self.ctx, expected_remove,
-                                                    topic=self.topic)
 
     def test_update_metering_label_rules_rpc_call(self):
         second_uuid = 'e27fe2df-376e-4ac7-ae13-92f050a21f84'
@@ -237,11 +232,9 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
                     self.mock_uuid.return_value = second_uuid
                     with self.metering_label_rule(l['id'], direction='egress'):
                         self.mock_fanout.assert_called_with(self.ctx,
-                                                            expected_add,
+                                                            expected_add)
-                                                            topic=self.topic)
                     self.mock_fanout.assert_called_with(self.ctx,
-                                                        expected_del,
+                                                        expected_del)
-                                                        topic=self.topic)
 
     def test_delete_metering_label_does_not_clear_router_tenant_id(self):
         tenant_id = '654f6b9d-0f36-4ae5-bd1b-01616794ca60'

neutron/tests/unit/test_agent_rpc.py

@@ -76,7 +76,6 @@ class AgentPluginReportState(base.BaseTestCase):
                              {'agent_state': expected_agent_state})
             self.assertIsInstance(call.call_args[0][1]['args']['time'],
                                   str)
-            self.assertEqual(call.call_args[1]['topic'], topic)
 
     def test_plugin_report_state_cast(self):
         topic = 'test'
@@ -92,7 +91,6 @@ class AgentPluginReportState(base.BaseTestCase):
                              {'agent_state': expected_agent_state})
             self.assertIsInstance(cast.call_args[0][1]['args']['time'],
                                   str)
-            self.assertEqual(cast.call_args[1]['topic'], topic)
 
 
 class AgentRPCMethods(base.BaseTestCase):

neutron/tests/unit/test_security_groups_rpc.py

@@ -1184,8 +1184,7 @@ class SecurityGroupServerRpcApiTestCase(base.BaseTestCase):
                  {'devices': ['fake_device']},
               'method': 'security_group_rules_for_devices',
               'namespace': None},
-             version=sg_rpc.SG_RPC_VERSION,
+             version=sg_rpc.SG_RPC_VERSION)])
-             topic='fake_topic')])
 
 
 class FakeSGNotifierAPI(n_rpc.RpcProxy,