self.conf = cfg.CONF
self.setup_eswitch_mgr(physical_devices_mappings,
exclude_devices)
- configurations = {'device_mappings': physical_devices_mappings}
- self.agent_state = {
- 'binary': 'neutron-sriov-nic-agent',
- 'host': self.conf.host,
- 'topic': n_constants.L2_AGENT_TOPIC,
- 'configurations': configurations,
- 'agent_type': n_constants.AGENT_TYPE_NIC_SWITCH,
- 'start_flag': True}
# Stores port update notifications for processing in the main loop
self.updated_devices = set()
self._setup_rpc()
self.ext_manager = self._create_agent_extension_manager(
self.connection)
+
+ configurations = {'device_mappings': physical_devices_mappings,
+ 'extensions': self.ext_manager.names()}
+ self.agent_state = {
+ 'binary': 'neutron-sriov-nic-agent',
+ 'host': self.conf.host,
+ 'topic': n_constants.L2_AGENT_TOPIC,
+ 'configurations': configurations,
+ 'agent_type': n_constants.AGENT_TYPE_NIC_SWITCH,
+ 'start_flag': True}
+
# The initialization is complete; we can start receiving messages
self.connection.consume_in_threads()
# Initialize iteration counter
from oslo_config import cfg
from oslo_utils import uuidutils
+from neutron.agent.l2.extensions import manager as l2_ext_manager
+from neutron.agent import rpc as agent_rpc
from neutron.extensions import portbindings
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import config # noqa
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import exceptions
kwargs = {'context': self.context, 'port': port}
self.sriov_rpc_callback.port_update(**kwargs)
self.assertEqual(set(), self.agent.updated_devices)
+
+
+class TestSRIOVAgentExtensionConfig(base.BaseTestCase):
+ def setUp(self):
+ super(TestSRIOVAgentExtensionConfig, self).setUp()
+ l2_ext_manager.register_opts(cfg.CONF)
+ # disable setting up periodic state reporting
+ cfg.CONF.set_override('report_interval', 0, group='AGENT')
+ cfg.CONF.set_override('extensions', ['qos'], group='agent')
+
+ @mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.eswitch_manager"
+ ".ESwitchManager.get_assigned_devices_info", return_value=[])
+ def test_report_loaded_extension(self, *args):
+ with mock.patch.object(agent_rpc.PluginReportStateAPI,
+ 'report_state') as mock_report_state:
+ agent = sriov_nic_agent.SriovNicSwitchAgent({}, {}, 0)
+ agent._report_state()
+ mock_report_state.assert_called_with(
+ agent.context, agent.agent_state)
+ self.assertEqual(
+ ['qos'], agent.agent_state['configurations']['extensions'])