test_vif_pool.py 文件源码

def test__return_ports_to_pool_no_update(self, max_pool, m_sleep):
        cls = vif_pool.NeutronVIFPool
        m_driver = mock.MagicMock(spec=cls)
        neutron = self.useFixture(k_fix.MockNeutronClient()).client

        pool_key = ('node_ip', 'project_id', tuple(['security_group']))
        port_id = mock.sentinel.port_id
        pool_length = 5

        m_driver._recyclable_ports = {port_id: pool_key}
        m_driver._available_ports_pools = {}
        oslo_cfg.CONF.set_override('ports_pool_max',
                                   max_pool,
                                   group='vif_pool')
        oslo_cfg.CONF.set_override('port_debug',
                                   False,
                                   group='kubernetes')
        m_driver._get_ports_by_attrs.return_value = [
            {'id': port_id, 'security_groups': ['security_group']}]
        m_driver._get_pool_size.return_value = pool_length

        self.assertRaises(SystemExit, cls._return_ports_to_pool, m_driver)

        neutron.update_port.assert_not_called()
        neutron.delete_port.assert_not_called()