openstack
/
kuryr-kubernetes
Code Issues Proposed changes

Fix tests in test_public_ip.py 

Looks like all the tests in test_public_ip.py were fundametaly broken
because of mocking *all* the methods of FipPubIpDriver causing the
tested method to not be exectued. This commit fixes that.

Closes-Bug: 1798782

Change-Id: I6286db8d70a111efc88292b47c631d0d81108cd2
This commit is contained in:
Michał Dulko 2018-10-18 13:56:55 +02:00
parent 0be427f349
commit e176dd35cf
1 changed files with 40 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

129
kuryr_kubernetes/tests/unit/controller/drivers/test_public_ip.py
View File

@ -22,170 +22,121 @@ from kuryr_kubernetes.tests.unit import kuryr_fixtures as k_fix
class TestFipPubIpDriver(test_base.TestCase):
def setUp(self):
super(TestFipPubIpDriver, self).setUp()
self.driver = d_public_ip.FipPubIpDriver()
self.neutron = self.useFixture(k_fix.MockNeutronClient()).client
def test_is_ip_available_none_param(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
fip_ip_addr = None
fip_id = cls.is_ip_available(m_driver, fip_ip_addr)
self.assertIsNone(fip_id)
def test_is_ip_available_empty_param(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
fip_ip_addr = None
fip_id = cls.is_ip_available(m_driver, fip_ip_addr)
fip_id = self.driver.is_ip_available(None)
self.assertIsNone(fip_id)
def test_is_ip_available_ip_not_exist(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
floating_ip = {'floating_ip_address': '1.2.3.4', 'port_id': None,
'id': 'a2a62ea7-e3bf-40df-8c09-aa0c29876a6b'}
neutron.list_floatingips.return_value = {'floatingips': [floating_ip]}
self.neutron.list_floatingips.return_value = {
'floatingips': [floating_ip]}
fip_ip_addr = '1.1.1.1'
fip_id = cls.is_ip_available(m_driver, fip_ip_addr)
fip_id = self.driver.is_ip_available(fip_ip_addr)
self.assertIsNone(fip_id)
def test_is_ip_available_empty_fip_list(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
floating_ip = None
neutron.list_floatingips.return_value = {'floatingips': [floating_ip]}
self.neutron.list_floatingips.return_value = {
'floatingips': [floating_ip]}
fip_ip_addr = '1.1.1.1'
fip_id = cls.is_ip_available(m_driver, fip_ip_addr)
fip_id = self.driver.is_ip_available(fip_ip_addr)
self.assertIsNone(fip_id)
def test_is_ip_available_occupied_fip(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
floating_ip = {'floating_ip_address': '1.2.3.4',
'port_id': 'ec29d641-fec4-4f67-928a-124a76b3a8e6'}
neutron.list_floatingips.return_value = {'floatingips': [floating_ip]}
self.neutron.list_floatingips.return_value = {
'floatingips': [floating_ip]}
fip_ip_addr = '1.2.3.4'
fip_id = cls.is_ip_available(m_driver, fip_ip_addr)
fip_id = self.driver.is_ip_available(fip_ip_addr)
self.assertIsNone(fip_id)
def test_is_ip_available_ip_exist_and_available(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
floating_ip = {'floating_ip_address': '1.2.3.4', 'port_id': None,
'id': 'a2a62ea7-e3bf-40df-8c09-aa0c29876a6b'}
neutron.list_floatingips.return_value = {'floatingips': [floating_ip]}
self.neutron.list_floatingips.return_value = {
'floatingips': [floating_ip]}
fip_ip_addr = '1.2.3.4'
fip_id = cls.is_ip_available(m_driver, fip_ip_addr)
fip_id = self.driver.is_ip_available(fip_ip_addr)
self.assertEqual(fip_id, 'a2a62ea7-e3bf-40df-8c09-aa0c29876a6b')
def test_allocate_ip_all_green(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
pub_net_id = mock.sentinel.pub_net_id
pub_subnet_id = mock.sentinel.pub_subnet_id
project_id = mock.sentinel.project_id
description = mock.sentinel.description
neutron = self.useFixture(k_fix.MockNeutronClient()).client
floating_ip = {'floating_ip_address': '1.2.3.5',
'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'}
neutron.create_floatingip.return_value = {'floatingip': floating_ip}
self.neutron.create_floatingip.return_value = {
'floatingip': floating_ip}
fip_id, fip_addr = cls.allocate_ip(
m_driver, pub_net_id, project_id, pub_subnet_id, description)
fip_id, fip_addr = self.driver.allocate_ip(pub_net_id, project_id,
pub_subnet_id, description)
self.assertEqual(fip_id, floating_ip['id'])
self.assertEqual(fip_addr, floating_ip['floating_ip_address'])
def test_allocate_ip_neutron_exception(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
pub_net_id = mock.sentinel.pub_net_id
pub_subnet_id = mock.sentinel.pub_subnet_id
project_id = mock.sentinel.project_id
description = mock.sentinel.description
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.create_floatingip.side_effect = n_exc.NeutronClientException
cf = self.neutron.create_floatingip
cf.side_effect = n_exc.NeutronClientException
self.assertRaises(
n_exc.NeutronClientException, cls.allocate_ip,
m_driver, pub_net_id, project_id, pub_subnet_id, description)
n_exc.NeutronClientException, self.driver.allocate_ip,
pub_net_id, project_id, pub_subnet_id, description)
def test_free_ip_neutron_exception(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
res_id = mock.sentinel.res_id
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.delete_floatingip.side_effect = n_exc.NeutronClientException
rc = cls.free_ip(m_driver, res_id)
df = self.neutron.delete_floatingip
df.side_effect = n_exc.NeutronClientException
rc = self.driver.free_ip(res_id)
self.assertEqual(rc, False)
def test_free_ip_succeeded(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
res_id = mock.sentinel.res_id
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.delete_floatingip.return_value = None
rc = cls.free_ip(m_driver, res_id)
rc = self.driver.free_ip(res_id)
self.assertEqual(rc, True)
# try:
# cls.free_ip(m_driver, res_id)
# except Exception:
# self.fail("Encountered an unexpected exception.")
def test_associate_neutron_exception(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
res_id = mock.sentinel.res_id
vip_port_id = mock.sentinel.vip_port_id
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.update_floatingip.side_effect = n_exc.NeutronClientException
retcode = cls.associate(m_driver, res_id, vip_port_id)
self.assertIsNone(retcode)
uf = self.neutron.update_floatingip
uf.side_effect = n_exc.NeutronClientException
self.assertRaises(n_exc.NeutronClientException, self.driver.associate,
res_id, vip_port_id)
def test_associate_succeeded(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
res_id = mock.sentinel.res_id
vip_port_id = mock.sentinel.vip_port_id
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.update_floatingip.return_value = None
retcode = cls.associate(m_driver, res_id, vip_port_id)
retcode = self.driver.associate(res_id, vip_port_id)
self.assertIsNone(retcode)
def test_disassociate_neutron_exception(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
res_id = mock.sentinel.res_id
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.update_floatingip.side_effect = n_exc.NeutronClientException
self.assertIsNone(cls.disassociate
(m_driver, res_id))
uf = self.neutron.update_floatingip
uf.side_effect = n_exc.NeutronClientException
self.assertRaises(n_exc.NeutronClientException,
self.driver.disassociate, res_id)
def test_disassociate_succeeded(self):
cls = d_public_ip.FipPubIpDriver
m_driver = mock.Mock(spec=cls)
res_id = mock.sentinel.res_id
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.update_floatingip.return_value = None
self.assertIsNone(cls.disassociate
(m_driver, res_id))
self.assertIsNone(self.driver.disassociate(res_id))