openstack
/
astara
Code Issues Proposed changes

Fix order of arguments in assertEqual 

Some tests used incorrect order assertEqual(observed, expected).
The correct order expected by testtools is
assertEqual(expected, observed).
This patch will fix the issue.

Change-Id: I44c7c6a6bf248b17c4f001586a79a91ff9d9dc3d
Partial-Bug: #1259292
This commit is contained in:
Ji-Wei 2016-06-23 16:12:40 +08:00 committed by JiWei
parent aaa6f0682e
commit 3fbbcd941c
29 changed files with 375 additions and 374 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
astara/test/functional/test_tenant_router.py
View File

@ -56,7 +56,7 @@ class TestAstaraHARouter(AstaraRouterTestBase):
def test_ha_router_servers(self):
service_instances = self.get_router_appliance_server(
self.router['id'], ha_router=self.HA_ROUTER)
self.assertEqual(len(service_instances), 2)
self.assertEqual(2, len(service_instances))
# kill the master and ensure it is backfilled with a new instance
master, backup = service_instances
@ -68,8 +68,8 @@ class TestAstaraHARouter(AstaraRouterTestBase):
service_instances = self.get_router_appliance_server(
self.router['id'], retries=600, ha_router=self.HA_ROUTER)
self.assertEqual(len(service_instances), 2)
self.assertEqual(service_instances[0], backup)
self.assertEqual(2, len(service_instances))
self.assertEqual(backup, service_instances[0])
class TestAstaraRouter(AstaraRouterTestBase):

22
astara/test/unit/api/config/test_common_config.py
View File

@ -87,7 +87,7 @@ class TestCommonConfig(unittest.TestCase):
}
]
}
self.assertEqual(result, expected)
self.assertEqual(expected, result)
def test_interface_config(self):
expected = {
@ -116,7 +116,7 @@ class TestCommonConfig(unittest.TestCase):
'host_routes': {},
'id': 's1',
}
self.assertEqual(common._subnet_config(fakes.fake_subnet), expected)
self.assertEqual(expected, common._subnet_config(fakes.fake_subnet))
def test_subnet_config_with_slaac_enabled(self):
expected = {
@ -128,7 +128,7 @@ class TestCommonConfig(unittest.TestCase):
'id': 'fake_id',
}
self.assertEqual(
common._subnet_config(fakes.fake_subnet_with_slaac), expected)
expected, common._subnet_config(fakes.fake_subnet_with_slaac))
def test_subnet_config_no_gateway(self):
expected = {
@ -147,7 +147,7 @@ class TestCommonConfig(unittest.TestCase):
dns_nameservers=['8.8.8.8'],
ipv6_ra_mode='',
host_routes={})
self.assertEqual(common._subnet_config(sn), expected)
self.assertEqual(expected, common._subnet_config(sn))
def test_subnet_config_gateway_none(self):
expected = {
@ -166,24 +166,24 @@ class TestCommonConfig(unittest.TestCase):
dns_nameservers=['8.8.8.8'],
ipv6_ra_mode='',
host_routes={})
self.assertEqual(common._subnet_config(sn), expected)
self.assertEqual(expected, common._subnet_config(sn))
def test_allocation_config_vrrp(self):
subnets_dict = {fakes.fake_subnet.id: fakes.fake_subnet}
self.assertEqual(
[],
common._allocation_config(
[fakes.fake_instance_vrrp_port],
subnets_dict),
[]
subnets_dict)
)
def test_allocation_config_lb(self):
subnets_dict = {fakes.fake_subnet.id: fakes.fake_subnet}
self.assertEqual(
[],
common._allocation_config(
[fakes.fake_instance_lb_port],
subnets_dict),
[]
subnets_dict)
)
def test_allocation_config_mgt(self):
@ -195,8 +195,8 @@ class TestCommonConfig(unittest.TestCase):
'device_id': 'v-v-v-v'}
]
self.assertEqual(
expected,
common._allocation_config([
fakes.fake_instance_mgt_port],
subnets_dict),
expected
subnets_dict)
)

2
astara/test/unit/api/config/test_loadbalancer_config.py
View File

@ -48,4 +48,4 @@ class TestLoadbalancerConfigAPI(base.RugTestBase):
'loadbalancer': fake_lb.to_dict(),
}
}
self.assertEqual(res, expected)
self.assertEqual(expected, res)

16
astara/test/unit/api/config/test_router_config.py
View File

@ -102,7 +102,7 @@ class TestAstaraClient(unittest.TestCase):
'vpn': {}
}
self.assertEqual(config, expected)
self.assertEqual(expected, config)
mocks['load_provider_rules'].assert_called_once_with('/the/path')
mocks['generate_network_config'].assert_called_once_with(
@ -117,13 +117,13 @@ class TestAstaraClient(unittest.TestCase):
mock_open.assert_called_once_with('/the/path')
load.assert_called_once_with(mock_open.return_value)
self.assertEqual(r, rules_dict)
self.assertEqual(rules_dict, r)
@mock.patch.object(__builtins__, 'open', autospec=True)
def test_load_provider_rules_not_found(self, mock_open):
mock_open.side_effect = IOError()
res = conf_mod.load_provider_rules('/tmp/path')
self.assertEqual(res, {})
self.assertEqual({}, res)
@mock.patch('astara.api.config.common.network_config')
def test_generate_network_config(self, mock_net_conf):
@ -146,7 +146,7 @@ class TestAstaraClient(unittest.TestCase):
'configured_network'
]
self.assertEqual(result, expected)
self.assertEqual(expected, result)
expected_calls = [
mock.call(
@ -174,7 +174,7 @@ class TestAstaraClient(unittest.TestCase):
result = conf_mod.generate_floating_config(rtr)
expected = [{'floating_ip': '9.9.9.9', 'fixed_ip': '192.168.1.1'}]
self.assertEqual(result, expected)
self.assertEqual(expected, result)
class TestAstaraClientGateway(unittest.TestCase):
@ -261,7 +261,7 @@ class TestAstaraClientGateway(unittest.TestCase):
fakes.fake_router,
self.networks,
)
self.assertEqual(result, '172.16.77.1')
self.assertEqual('172.16.77.1', result)
def test_without_ipv4_on_external_port(self):
# Only set a V6 address
@ -274,7 +274,7 @@ class TestAstaraClientGateway(unittest.TestCase):
fakes.fake_router,
self.networks,
)
self.assertEqual(result, '')
self.assertEqual('', result)
def test_extra_ipv4_on_external_port(self):
self.networks[0]['interface']['addresses'] = [
@ -289,4 +289,4 @@ class TestAstaraClientGateway(unittest.TestCase):
fakes.fake_router,
self.networks,
)
self.assertEqual(result, '172.16.77.1')
self.assertEqual('172.16.77.1', result)

10
astara/test/unit/api/test_astara_router.py
View File

@ -72,8 +72,8 @@ class TestAstaraClient(unittest.TestCase):
'interfaces': 'the_interfaces'
}
self.assertEqual(astara_client.get_interfaces('fe80::2', 5000),
'the_interfaces')
self.assertEqual('the_interfaces',
astara_client.get_interfaces('fe80::2', 5000))
self.mock_get.assert_called_once_with(
'http://[fe80::2]:5000/v1/system/interfaces',
timeout=30
@ -91,7 +91,7 @@ class TestAstaraClient(unittest.TestCase):
data=b'{"foo": "bar"}',
headers={'Content-type': 'application/json'},
timeout=90)
self.assertEqual(resp, config)
self.assertEqual(config, resp)
def test_update_config_with_custom_config(self):
config = {'foo': 'bar'}
@ -107,7 +107,7 @@ class TestAstaraClient(unittest.TestCase):
data=b'{"foo": "bar"}',
headers={'Content-type': 'application/json'},
timeout=5)
self.assertEqual(resp, config)
self.assertEqual(config, resp)
def test_update_config_failure(self):
config = {'foo': 'bar'}
@ -137,4 +137,4 @@ class TestAstaraClient(unittest.TestCase):
timeout=30
)
self.assertEqual(resp, ['label1', 'label2'])
self.assertEqual(['label1', 'label2'], resp)

2
astara/test/unit/api/test_keystone.py
View File

@ -35,4 +35,4 @@ class KeystoneTest(base.RugTestBase):
ks_session = keystone.KeystoneSession().session
mock_load_auth.assert_called_with(cfg.CONF, 'keystone_authtoken')
mock_session.assert_called_with(auth=fake_auth)
self.assertEqual(ks_session, fake_session)
self.assertEqual(fake_session, ks_session)

70
astara/test/unit/api/test_neutron_wrapper.py
View File

@ -28,14 +28,14 @@ class TestuNeutronModels(base.RugTestBase):
def test_router(self):
r = neutron.Router(
'1', 'tenant_id', 'name', True, 'ACTIVE', 'ext', ['int'], ['fip'])
self.assertEqual(r.id, '1')
self.assertEqual(r.tenant_id, 'tenant_id')
self.assertEqual(r.name, 'name')
self.assertEqual('1', r.id)
self.assertEqual('tenant_id', r.tenant_id)
self.assertEqual('name', r.name)
self.assertTrue(r.admin_state_up)
self.assertEqual(r.status, 'ACTIVE')
self.assertEqual(r.external_port, 'ext')
self.assertEqual(r.floating_ips, ['fip'])
self.assertEqual(r.internal_ports, ['int'])
self.assertEqual('ACTIVE', r.status)
self.assertEqual('ext', r.external_port)
self.assertEqual(['fip'], r.floating_ips)
self.assertEqual(['int'], r.internal_ports)
self.assertEqual(set(['ext', 'int']), set(r.ports))
def test_router_from_dict(self):
@ -88,14 +88,14 @@ class TestuNeutronModels(base.RugTestBase):
r = neutron.Router.from_dict(d)
self.assertEqual(r.id, '1')
self.assertEqual(r.tenant_id, 'tenant_id')
self.assertEqual(r.name, 'name')
self.assertEqual('1', r.id)
self.assertEqual('tenant_id', r.tenant_id)
self.assertEqual('name', r.name)
self.assertTrue(r.admin_state_up)
self.assertTrue(r.floating_ips) # just make sure this exists
self.assertEqual(
sorted([ip.id for ip in r.internal_ports]),
['ha_int', 'int'])
['ha_int', 'int'],
sorted([ip.id for ip in r.internal_ports]))
def test_router_eq(self):
r1 = neutron.Router(
@ -130,16 +130,16 @@ class TestuNeutronModels(base.RugTestBase):
s = neutron.Subnet.from_dict(d)
self.assertEqual(s.id, '1')
self.assertEqual(s.tenant_id, 'tenant_id')
self.assertEqual(s.name, 'name')
self.assertEqual(s.network_id, 'network_id')
self.assertEqual(s.ip_version, 6)
self.assertEqual(s.cidr, netaddr.IPNetwork('fe80::/64'))
self.assertEqual(s.gateway_ip, netaddr.IPAddress('fe80::1'))
self.assertEqual('1', s.id)
self.assertEqual('tenant_id', s.tenant_id)
self.assertEqual('name', s.name)
self.assertEqual('network_id', s.network_id)
self.assertEqual(6, s.ip_version)
self.assertEqual(netaddr.IPNetwork('fe80::/64'), s.cidr)
self.assertEqual(netaddr.IPAddress('fe80::1'), s.gateway_ip)
self.assertTrue(s.enable_dhcp, True)
self.assertEqual(s.dns_nameservers, ['8.8.8.8', '8.8.4.4'])
self.assertEqual(s.host_routes, [])
self.assertEqual(['8.8.8.8', '8.8.4.4'], s.dns_nameservers)
self.assertEqual([], s.host_routes)
def test_subnet_gateway_none(self):
d = {
@ -156,7 +156,7 @@ class TestuNeutronModels(base.RugTestBase):
'host_routes': []
}
s = neutron.Subnet.from_dict(d)
self.assertEqual(s.cidr, netaddr.IPNetwork('fe80::/64'))
self.assertEqual(netaddr.IPNetwork('fe80::/64'), s.cidr)
self.assertIs(None, s.gateway_ip)
def test_subnet_gateway_not_ip(self):
@ -174,7 +174,7 @@ class TestuNeutronModels(base.RugTestBase):
'host_routes': []
}
s = neutron.Subnet.from_dict(d)
self.assertEqual(s.cidr, netaddr.IPNetwork('fe80::/64'))
self.assertEqual(netaddr.IPNetwork('fe80::/64'), s.cidr)
self.assertIs(None, s.gateway_ip)
def test_subnet_cidr_none(self):
@ -228,11 +228,11 @@ class TestuNeutronModels(base.RugTestBase):
p = neutron.Port.from_dict(d)
self.assertEqual(p.id, '1')
self.assertEqual(p.device_id, 'device_id')
self.assertEqual(p.mac_address, 'aa:bb:cc:dd:ee:ff')
self.assertEqual(p.device_owner, 'test')
self.assertEqual(len(p.fixed_ips), 1)
self.assertEqual('1', p.id)
self.assertEqual('device_id', p.device_id)
self.assertEqual('aa:bb:cc:dd:ee:ff', p.mac_address)
self.assertEqual('test', p.device_owner)
self.assertEqual(1, len(p.fixed_ips))
def test_fixed_ip_model(self):
d = {
@ -242,8 +242,8 @@ class TestuNeutronModels(base.RugTestBase):
fip = neutron.FixedIp.from_dict(d)
self.assertEqual(fip.subnet_id, 'sub1')
self.assertEqual(fip.ip_address, netaddr.IPAddress('192.168.1.1'))
self.assertEqual('sub1', fip.subnet_id)
self.assertEqual(netaddr.IPAddress('192.168.1.1'), fip.ip_address)
def test_floating_ip_model(self):
d = {
@ -254,9 +254,9 @@ class TestuNeutronModels(base.RugTestBase):
fip = neutron.FloatingIP.from_dict(d)
self.assertEqual(fip.id, 'a-b-c-d')
self.assertEqual(fip.floating_ip, netaddr.IPAddress('9.9.9.9'))
self.assertEqual(fip.fixed_ip, netaddr.IPAddress('192.168.1.1'))
self.assertEqual('a-b-c-d', fip.id)
self.assertEqual(netaddr.IPAddress('9.9.9.9'), fip.floating_ip)
self.assertEqual(netaddr.IPAddress('192.168.1.1'), fip.fixed_ip)
class FakeConf:
@ -279,8 +279,8 @@ class TestNeutronWrapper(base.RugTestBase):
neutron_wrapper = neutron.Neutron(conf)
neutron_wrapper.purge_management_interface()
self.assertEqual(driver.get_device_name.call_count, 1)
self.assertEqual(driver.unplug.call_count, 1)
self.assertEqual(1, driver.get_device_name.call_count)
self.assertEqual(1, driver.unplug.call_count)
def test_clear_device_id(self):
neutron_wrapper = neutron.Neutron(mock.Mock())

26
astara/test/unit/api/test_nova_wrapper.py
View File

@ -164,7 +164,7 @@ class TestNovaWrapper(base.RugTestBase):
result = self.nova.get_instance_for_obj('foo_instance_name')
self.client.assert_has_calls(expected)
self.assertEqual(result, instance)
self.assertEqual(instance, result)
def test_get_instance_for_obj_not_found(self):
self.client.servers.list.return_value = []
@ -184,7 +184,7 @@ class TestNovaWrapper(base.RugTestBase):
]
result = self.nova.get_instance_by_id('instance_id')
self.client.servers.get.assert_has_calls(expected)
self.assertEqual(result, 'fake_instance')
self.assertEqual('fake_instance', result)
def test_get_instance_by_id_not_found(self):
not_found = novaclient_exceptions.NotFound('instance_id')
@ -200,7 +200,7 @@ class TestNovaWrapper(base.RugTestBase):
def test_format_userdata(self, fake_ssh_key):
fake_ssh_key.return_value = 'fake_key'
result = nova.format_userdata(fake_int_port)
self.assertEqual(result.strip(), EXPECTED_USERDATA.strip())
self.assertEqual(EXPECTED_USERDATA.strip(), result.strip())
@mock.patch.object(__builtins__, 'open', autospec=True)
def test_ssh_key(self, fake_open):
@ -209,7 +209,7 @@ class TestNovaWrapper(base.RugTestBase):
mock_key_file.__enter__.return_value = mock_key_file
fake_open.return_value = mock_key_file
result = nova._ssh_key()
self.assertEqual(result, 'fake-key')
self.assertEqual('fake-key', result)
@mock.patch.object(__builtins__, 'open', autospec=True)
def test_ssh_key_sanitize(self, fake_open):
@ -222,14 +222,14 @@ class TestNovaWrapper(base.RugTestBase):
mock_key_file.__enter__.return_value = mock_key_file
fake_open.return_value = mock_key_file
result = nova._ssh_key()
self.assertEqual(result, 'fake-key with some newlines')
self.assertEqual('fake-key with some newlines', result)
@mock.patch.object(nova, 'LOG', autospec=True)
@mock.patch.object(__builtins__, 'open', autospec=True)
def test_ssh_key_not_found(self, fake_open, fake_log):
fake_open.side_effect = IOError
result = nova._ssh_key()
self.assertEqual(result, '')
self.assertEqual('', result)
self.assertTrue(fake_log.warning.called)
@mock.patch.object(nova.Nova, 'get_instance_for_obj', return_value=None)
@ -252,7 +252,7 @@ class TestNovaWrapper(base.RugTestBase):
make_ports_callback='foo_callback',
)
fake_get.assert_called_with('foo_instance_name')
self.assertEqual(res, 'fake_new_instance_info')
self.assertEqual('fake_new_instance_info', res)
@mock.patch.object(nova.Nova, 'get_instance_for_obj')
def test_boot_instance_exists(self, fake_get):
@ -292,9 +292,9 @@ class TestNovaWrapper(base.RugTestBase):
)
fake_get.assert_called_with('foo_instance_name')
self.assertIsInstance(res, nova.InstanceInfo)
self.assertEqual(res.id_, 'existing_instance_id')
self.assertEqual(res.name, 'ak-appliance')
self.assertEqual(res.image_uuid, 'fake_image_uuid')
self.assertEqual('existing_instance_id', res.id_)
self.assertEqual('ak-appliance', res.name)
self.assertEqual('fake_image_uuid', res.image_uuid)
@mock.patch.object(nova.Nova, 'get_instance_by_id', return_value=None)
def test_boot_instance_prev_inst(self, fake_get):
@ -316,7 +316,7 @@ class TestNovaWrapper(base.RugTestBase):
flavor='foo_flavor',
make_ports_callback='foo_callback',
)
self.assertEqual(res, 'fake_new_instance_info')
self.assertEqual('fake_new_instance_info', res)
@mock.patch.object(nova.Nova, 'get_instance_by_id')
def test_boot_instance_exists_prev_inst(self, fake_get):
@ -355,8 +355,8 @@ class TestNovaWrapper(base.RugTestBase):
make_ports_callback='foo_callback',
)
# assert we get back the same instance_info but with updated status
self.assertEqual(res.nova_status, 'BUILD')
self.assertEqual(res.id_, fake_instance.id)
self.assertEqual('BUILD', res.nova_status)
self.assertEqual(fake_instance.id, res.id_)
self.assertIsInstance(res, nova.InstanceInfo)
def test_from_nova(self):

16
astara/test/unit/common/test_agent_linux_utils.py
View File

@ -49,35 +49,35 @@ class AgentUtilsExecuteTest(testtools.TestCase):
def test_without_helper(self):
result = utils.execute(["ls", self.test_file])
self.assertEqual(result, "%s\n" % self.test_file)
self.assertEqual("%s\n" % self.test_file, result)
def test_with_helper(self):
result = utils.execute(["ls", self.test_file],
self.root_helper)
self.assertEqual(result, "ls %s\n" % self.test_file)
self.assertEqual("ls %s\n" % self.test_file, result)
def test_stderr(self):
stdout, stderr = utils.execute(["ls", self.test_file],
return_stderr=True)
self.assertEqual(stdout, "%s\n" % self.test_file)
self.assertEqual(stderr, "")
self.assertEqual("%s\n" % self.test_file, stdout)
self.assertEqual("", stderr)
def test_check_exit_code(self):
stdout = utils.execute(["ls", self.test_file[:-1]],
check_exit_code=False)
self.assertEqual(stdout, "")
self.assertEqual("", stdout)
self.assertRaises(RuntimeError, utils.execute,
["ls", self.test_file[:-1]])
def test_process_input(self):
result = utils.execute(["cat"], process_input="%s\n" %
self.test_file[:-1])
self.assertEqual(result, "%s\n" % self.test_file[:-1])
self.assertEqual("%s\n" % self.test_file[:-1], result)
def test_with_addl_env(self):
result = utils.execute(["ls", self.test_file],
addl_env={'foo': 'bar'})
self.assertEqual(result, "%s\n" % self.test_file)
self.assertEqual("%s\n" % self.test_file, result)
class AgentUtilsGetInterfaceMAC(testtools.TestCase):
@ -88,7 +88,7 @@ class AgentUtilsGetInterfaceMAC(testtools.TestCase):
'\x01\x02\x03\x04\x05\x06',
'\x00' * 232])
actual_val = utils.get_interface_mac('eth0')
self.assertEqual(actual_val, expect_val)
self.assertEqual(expect_val, actual_val)
class AgentUtilsReplaceFile(testtools.TestCase):

4
astara/test/unit/common/test_config.py
View File

@ -26,8 +26,8 @@ class TestConfig(base.RugTestBase):
with mock.patch('os.path.isfile', side_effect=mock_isfile_f):
self.assertEqual(
config.get_best_config_path(original),
expected
expected,
config.get_best_config_path(original)
)
def test_get_best_config_path_preferred(self):

6
astara/test/unit/common/test_linux_interface.py
View File

@ -242,7 +242,7 @@ class TestBridgeInterfaceDriver(TestBase):
'tap0',
'aa:bb:cc:dd:ee:ff')
self.ip_dev.assert_has_calls([])
self.assertEqual(log.call_count, 1)
self.assertEqual(1, log.call_count)
def test_unplug_no_device(self):
self.device_exists.return_value = False
@ -253,14 +253,14 @@ class TestBridgeInterfaceDriver(TestBase):
[mock.call(),
mock.call('tap0', self.conf.root_helper),
mock.call().link.delete()]
self.assertEqual(log.exception.call_count, 1)
self.assertEqual(1, log.exception.call_count)
def test_unplug(self):
self.device_exists.return_value = True
with mock.patch('astara.common.linux.interface.LOG.debug') as log:
br = interface.BridgeInterfaceDriver(self.conf)
br.unplug('tap0')
self.assertEqual(log.call_count, 1)
self.assertEqual(1, log.call_count)
self.ip_dev.assert_has_calls([
mock.call('tap0', self.conf.root_helper, None),

70
astara/test/unit/common/test_linux_ip_lib.py
View File

@ -180,11 +180,11 @@ class TestIpWrapper(unittest.TestCase):
def test_get_devices(self):
self.execute.return_value = '\n'.join(LINK_SAMPLE)
retval = ip_lib.IPWrapper('sudo').get_devices()
self.assertEqual(retval,
[ip_lib.IPDevice('lo'),
self.assertEqual([ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
ip_lib.IPDevice('br-int'),
ip_lib.IPDevice('gw-ddc717df-49')])
ip_lib.IPDevice('gw-ddc717df-49')],
retval)
self.execute.assert_called_once_with('o', 'link', ('list',),
'sudo', None)
@ -192,11 +192,11 @@ class TestIpWrapper(unittest.TestCase):
def test_get_devices_malformed_line(self):
self.execute.return_value = '\n'.join(LINK_SAMPLE + ['gibberish'])
retval = ip_lib.IPWrapper('sudo').get_devices()
self.assertEqual(retval,
[ip_lib.IPDevice('lo'),
self.assertEqual([ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
ip_lib.IPDevice('br-int'),
ip_lib.IPDevice('gw-ddc717df-49')])
ip_lib.IPDevice('gw-ddc717df-49')],
retval)
self.execute.assert_called_once_with('o', 'link', ('list',),
'sudo', None)
@ -204,10 +204,10 @@ class TestIpWrapper(unittest.TestCase):
def test_get_namespaces(self):
self.execute.return_value = '\n'.join(NETNS_SAMPLE)
retval = ip_lib.IPWrapper.get_namespaces('sudo')
self.assertEqual(retval,
['12345678-1234-5678-abcd-1234567890ab',
self.assertEqual(['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
'cccccccc-cccc-cccc-cccc-cccccccccccc'],
retval)
self.execute.assert_called_once_with('', 'netns', ('list',),
root_helper='sudo')
@ -227,9 +227,9 @@ class TestIpWrapper(unittest.TestCase):
def test_get_device(self):
dev = ip_lib.IPWrapper('sudo', 'ns').device('eth0')
self.assertEqual(dev.root_helper, 'sudo')
self.assertEqual(dev.namespace, 'ns')
self.assertEqual(dev.name, 'eth0')
self.assertEqual('sudo', dev.root_helper)
self.assertEqual('ns', dev.namespace)
self.assertEqual('eth0', dev.name)
def test_ensure_namespace(self):
with mock.patch.object(ip_lib, 'IPDevice') as ip_dev:
@ -247,7 +247,7 @@ class TestIpWrapper(unittest.TestCase):
ip_ns_cmd.exists.return_value = True
ns = ip_lib.IPWrapper('sudo').ensure_namespace('ns')
self.assertFalse(self.execute.called)
self.assertEqual(ns.namespace, 'ns')
self.assertEqual('ns', ns.namespace)
def test_namespace_is_empty_no_devices(self):
ip = ip_lib.IPWrapper('sudo', 'ns')
@ -275,7 +275,7 @@ class TestIpWrapper(unittest.TestCase):
ip_ns_cmd_cls.assert_has_calls([mock.call().exists('ns')])
self.assertNotIn(mock.call().delete('ns'),
ip_ns_cmd_cls.return_value.mock_calls)
self.assertEqual(mock_is_empty.mock_calls, [])
self.assertEqual([], mock_is_empty.mock_calls)
def test_garbage_collect_namespace_existing_empty_ns(self):
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls:
@ -311,7 +311,7 @@ class TestIpWrapper(unittest.TestCase):
mock_is_empty.assert_called_once_with()
expected = [mock.call(ip),
mock.call().exists('ns')]
self.assertEqual(ip_ns_cmd_cls.mock_calls, expected)
self.assertEqual(expected, ip_ns_cmd_cls.mock_calls)
self.assertNotIn(mock.call().delete('ns'),
ip_ns_cmd_cls.mock_calls)
@ -323,7 +323,7 @@ class TestIpWrapper(unittest.TestCase):
def test_add_device_to_namespace_is_none(self):
dev = mock.Mock()
ip_lib.IPWrapper('sudo').add_device_to_namespace(dev)
self.assertEqual(dev.mock_calls, [])
self.assertEqual([], dev.mock_calls)
class TestIPDevice(unittest.TestCase):
@ -352,7 +352,7 @@ class TestIPDevice(unittest.TestCase):
self.assertNotEqual(dev1, None)
def test_str(self):
self.assertEqual(str(ip_lib.IPDevice('tap0')), 'tap0')
self.assertEqual('tap0', str(ip_lib.IPDevice('tap0')))
class TestIPCommandBase(unittest.TestCase):
@ -394,7 +394,7 @@ class TestIPDeviceCommandBase(unittest.TestCase):
self.ip_cmd.COMMAND = 'foo'
def test_name_property(self):
self.assertEqual(self.ip_cmd.name, 'eth0')
self.assertEqual('eth0', self.ip_cmd.name)
class TestIPCmdBase(unittest.TestCase):
@ -440,12 +440,12 @@ class TestIpLinkCommand(TestIPCmdBase):
def test_set_netns(self):
self.link_cmd.set_netns('foo')
self._assert_sudo([], ('set', 'eth0', 'netns', 'foo'))
self.assertEqual(self.parent.namespace, 'foo')
self.assertEqual('foo', self.parent.namespace)
def test_set_name(self):
self.link_cmd.set_name('tap1')
self._assert_sudo([], ('set', 'eth0', 'name', 'tap1'))
self.assertEqual(self.parent.name, 'tap1')
self.assertEqual('tap1', self.parent.name)
def test_set_alias(self):
self.link_cmd.set_alias('openvswitch')
@ -457,27 +457,27 @@ class TestIpLinkCommand(TestIPCmdBase):
def test_address_property(self):
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.address, 'cc:dd:ee:ff:ab:cd')
self.assertEqual('cc:dd:ee:ff:ab:cd', self.link_cmd.address)
def test_mtu_property(self):
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.mtu, 1500)
self.assertEqual(1500, self.link_cmd.mtu)
def test_qdisc_property(self):
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.qdisc, 'mq')
self.assertEqual('mq', self.link_cmd.qdisc)
def test_qlen_property(self):
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.qlen, 1000)
self.assertEqual(1000, self.link_cmd.qlen)
def test_alias_property(self):
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.alias, 'openvswitch')
self.assertEqual('openvswitch', self.link_cmd.alias)
def test_state_property(self):
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.state, 'UP')
self.assertEqual('UP', self.link_cmd.state)
def test_settings_property(self):
expected = {'mtu': 1500,
@ -488,7 +488,7 @@ class TestIpLinkCommand(TestIPCmdBase):
'link/ether': 'cc:dd:ee:ff:ab:cd',
'alias': 'openvswitch'}
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.attributes, expected)
self.assertEqual(expected, self.link_cmd.attributes)
self._assert_call('o', ('show', 'eth0'))
@ -546,7 +546,7 @@ class TestIpAddrCommand(TestIPCmdBase):
for test_case in test_cases:
self.parent._run = mock.Mock(return_value=test_case)
self.assertEqual(self.addr_cmd.list(), expected)
self.assertEqual(expected, self.addr_cmd.list())
self._assert_call([], ('show', 'tap0'))
def test_list_filtered(self):
@ -560,8 +560,9 @@ class TestIpAddrCommand(TestIPCmdBase):
for test_case in test_cases:
output = '\n'.join(test_case.split('\n')[0:4])
self.parent._run.return_value = output
self.assertEqual(self.addr_cmd.list('global',
filters=['permanent']), expected)
self.assertEqual(expected,
self.addr_cmd.list('global',
filters=['permanent']))
self._assert_call([], ('show', 'tap0', 'permanent', 'scope',
'global'))
@ -602,8 +603,9 @@ class TestIpRouteCommand(TestIPCmdBase):
'expected': {'gateway': '10.35.19.254'}}]
for test_case in test_cases:
self.parent._run = mock.Mock(return_value=test_case['sample'])
self.assertEqual(self.route_cmd.get_gateway(),
test_case['expected'])
self.assertEqual(test_case['expected'],
self.route_cmd.get_gateway()
)
def test_pullup_route(self):
# interface is not the first in the list - requires
@ -631,7 +633,7 @@ class TestIpRouteCommand(TestIPCmdBase):
self.parent._run = mock.Mock(side_effect=pullup_side_effect)
self.route_cmd.pullup_route('tap1d7888a7-10')
# Check two calls - device get and subnet get
self.assertEqual(len(self.parent._run.mock_calls), 2)
self.assertEqual(2, len(self.parent._run.mock_calls))
class TestIpNetnsCommand(TestIPCmdBase):
@ -643,7 +645,7 @@ class TestIpNetnsCommand(TestIPCmdBase):
def test_add_namespace(self):
ns = self.netns_cmd.add('ns')
self._assert_sudo([], ('add', 'ns'), force_root_namespace=True)
self.assertEqual(ns.namespace, 'ns')
self.assertEqual('ns', ns.namespace)
def test_delete_namespace(self):
with mock.patch('astara.common.linux.utils.execute'):

10
astara/test/unit/db/test_debug_mode.py
View File

@ -36,7 +36,7 @@ class TestDBDebugModes(base.DbTestCase):
self.dbapi.enable_global_debug(reason='foo')
enabled, reason = self.dbapi.global_debug()
self.assertTrue(enabled)
self.assertEqual(reason, 'foo')
self.assertEqual('foo', reason)
self.dbapi.disable_global_debug()
enabled, reason = self.dbapi.global_debug()
@ -60,7 +60,7 @@ class TestDBDebugModes(base.DbTestCase):
enabled, reason = self.dbapi.resource_in_debug(
resource_uuid=r_id)
self.assertTrue(enabled)
self.assertEqual(reason, 'foo')
self.assertEqual('foo', reason)
def test_resources_in_debug(self):
r_ids = [uuid.uuid4().hex for i in range(1, 3)]
@ -69,7 +69,7 @@ class TestDBDebugModes(base.DbTestCase):
resource_uuid=r_id, reason='resource %s is broken' % r_id)
for debug_r_id, reason in self.dbapi.resources_in_debug():
self.assertIn(debug_r_id, r_ids)
self.assertEqual(reason, 'resource %s is broken' % debug_r_id)
self.assertEqual('resource %s is broken' % debug_r_id, reason)
def test_tenant_debug(self):
t_id = uuid.uuid4().hex
@ -88,7 +88,7 @@ class TestDBDebugModes(base.DbTestCase):
enabled, reason = self.dbapi.tenant_in_debug(
tenant_uuid=t_id)
self.assertTrue(enabled)
self.assertEqual(reason, 'foo')
self.assertEqual('foo', reason)
def test_tenants_in_debug(self):
t_ids = [uuid.uuid4().hex for i in range(1, 3)]
@ -97,4 +97,4 @@ class TestDBDebugModes(base.DbTestCase):
tenant_uuid=t_id, reason='tenant %s is broken' % t_id)
for debug_t_id, reason in self.dbapi.tenants_in_debug():
self.assertIn(debug_t_id, t_ids)
self.assertEqual(reason, 'tenant %s is broken' % debug_t_id)
self.assertEqual('tenant %s is broken' % debug_t_id, reason)

10
astara/test/unit/drivers/test_factory.py
View File

@ -22,7 +22,7 @@ from astara import drivers
class DriverFactoryTest(base.RugTestBase):
def test_get_driver(self):
for k, v in drivers.AVAILABLE_DRIVERS.items():
self.assertEqual(drivers.get(k), v)
self.assertEqual(v, drivers.get(k))
def test_get_bad_driver(self):
self.assertRaises(
@ -35,14 +35,14 @@ class DriverFactoryTest(base.RugTestBase):
all_driver_obj = drivers.AVAILABLE_DRIVERS.values()
self.config(enabled_drivers=all_driver_cfg)
enabled_drivers = [d for d in drivers.enabled_drivers()]
self.assertEqual(set(all_driver_obj), set(enabled_drivers))
self.assertEqual(set(enabled_drivers), set(all_driver_obj))
def test_enabled_drivers_nonexistent_left_out(self):
all_driver_cfg = drivers.AVAILABLE_DRIVERS.keys() + ['foodriver']
all_driver_obj = drivers.AVAILABLE_DRIVERS.values()
self.config(enabled_drivers=all_driver_cfg)
enabled_drivers = [d for d in drivers.enabled_drivers()]
self.assertEqual(set(all_driver_obj), set(enabled_drivers))
self.assertEqual(set(enabled_drivers), set(all_driver_obj))
@mock.patch('astara.drivers.get')
def test_load_from_byonf(self, fake_get):
@ -58,6 +58,6 @@ class DriverFactoryTest(base.RugTestBase):
}
ctx = mock.Mock()
res = drivers.load_from_byonf(ctx, byonf, 'fake_resource_id')
self.assertEqual(res, fake_driver_obj)
self.assertEqual(res.image_uuid, 'custom_image_uuid')
self.assertEqual(fake_driver_obj, res)
self.assertEqual('custom_image_uuid', res.image_uuid)
fake_driver.assert_called_with(ctx, 'fake_resource_id')

50
astara/test/unit/drivers/test_loadbalancer.py
View File

@ -52,19 +52,19 @@ class LoadBalancerDriverTest(base.RugTestBase):
lb = self._init_driver()
lb.post_init = mock.Mock()
self.assertEqual(
lb.name,
'ak-%s-%s' % (lb.RESOURCE_NAME, self.loadbalancer_id))
'ak-%s-%s' % (lb.RESOURCE_NAME, self.loadbalancer_id),
lb.name)
mock_post_init.assert_called_with(self.ctx)
def test_ports_no_loadbalancer(self):
lb = self._init_driver()
self.assertEqual(lb.ports, [])
self.assertEqual([], lb.ports)
def test_ports_with_loadbalancer(self):
lb = self._init_driver()
fake_lb = fakes.fake_loadbalancer()
lb._loadbalancer = fake_lb
self.assertEqual(set(lb.ports), set(fake_lb.ports))
self.assertEqual(set(fake_lb.ports), set(lb.ports))
def test_pre_boot(self):
lb = self._init_driver()
@ -92,7 +92,7 @@ class LoadBalancerDriverTest(base.RugTestBase):
mock_build_config.return_value = 'fake_config'
mock_build_config.assert_called_with(
self.ctx.neutron, lb._loadbalancer, fake_mgt_port, fake_iface_map)
self.assertEqual(res, 'fake_config')
self.assertEqual('fake_config', res)
@mock.patch('astara.api.astara_client.update_config')
def test_update_config(self, mock_update_config):
@ -114,7 +114,7 @@ class LoadBalancerDriverTest(base.RugTestBase):
self.ctx.neutron.create_vrrp_port.return_value = fake_lb_port
callback = lb.make_ports(self.ctx)
res = callback()
self.assertEqual(res, ('fake_mgt_port', [fake_lb_port]))
self.assertEqual(('fake_mgt_port', [fake_lb_port]), res)
@mock.patch('astara.api.neutron.Neutron')
def test_pre_populate_retry_loop(self, mocked_neutron_api):
@ -127,15 +127,15 @@ class LoadBalancerDriverTest(base.RugTestBase):
with mock.patch('time.sleep'):
lb.pre_populate_hook()
self.assertEqual(
neutron_client.get_loadbalancers.call_args_list,
[
mock.call()
for value in six.moves.range(len(returned_value))
]
],
neutron_client.get_loadbalancers.call_args_list
)
self.assertEqual(
neutron_client.get_loadbalancers.call_count,
len(returned_value)
len(returned_value),
neutron_client.get_loadbalancers.call_count
)
def _exit_loop_bad_auth(self, mocked_neutron_api, log, exc):
@ -186,7 +186,7 @@ class LoadBalancerDriverTest(base.RugTestBase):
id='2',
tenant_id='1',
)
self.assertEqual(res, [expected_resource])
self.assertEqual([expected_resource], res)
def test_get_resource_id_loadbalancer_msg(self):
msg = mock.Mock(
@ -194,8 +194,8 @@ class LoadBalancerDriverTest(base.RugTestBase):
)
lb = self._init_driver()
self.assertEqual(
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg),
'lb_id'
'lb_id',
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg)
)
def test_get_resource_id_listener_msg(self):
@ -204,8 +204,8 @@ class LoadBalancerDriverTest(base.RugTestBase):
)
lb = self._init_driver()
self.assertEqual(
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg),
'lb_id'
'lb_id',
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg)
)
def test_get_resource_id_pool_msg(self):
@ -216,8 +216,8 @@ class LoadBalancerDriverTest(base.RugTestBase):
self.ctx.neutron.get_loadbalancer_by_listener.return_value = fake_lb
lb = self._init_driver()
self.assertEqual(
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg),
fake_lb.id
fake_lb.id,
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg)
)
self.ctx.neutron.get_loadbalancer_by_listener.assert_called_with(
'fake_listener_id', 'foo_tenant'
@ -231,8 +231,8 @@ class LoadBalancerDriverTest(base.RugTestBase):
self.ctx.neutron.get_loadbalancer_by_member.return_value = fake_lb
lb = self._init_driver()
self.assertEqual(
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg),
fake_lb.id
fake_lb.id,
lb.get_resource_id_for_tenant(self.ctx, 'foo_tenant', msg)
)
self.ctx.neutron.get_loadbalancer_by_member.assert_called_with(
'fake_member_id', 'foo_tenant'
@ -242,7 +242,7 @@ class LoadBalancerDriverTest(base.RugTestBase):
tenant_id = 'fake_tenant_id'
res = loadbalancer.LoadBalancer.process_notification(
tenant_id, event_type, payload)
self.assertEqual(res, expected)
self.assertEqual(expected, res)
def test_process_notification_loadbalancerstatus(self):
self._test_notification('loadbalancerstatus.update', {}, None)
@ -306,8 +306,8 @@ class LoadBalancerDriverTest(base.RugTestBase):
lb = self._init_driver()
lb._loadbalancer = None
self.assertEqual(
lb.get_state(self.ctx),
states.GONE,
lb.get_state(self.ctx)
)
mock_ensure_cache.assert_called_with(self.ctx)
@ -317,8 +317,8 @@ class LoadBalancerDriverTest(base.RugTestBase):
fake_lb = fakes.fake_loadbalancer()
lb._loadbalancer = fake_lb
self.assertEqual(
lb.get_state(self.ctx),
fake_lb.status,
lb.get_state(self.ctx)
)
mock_ensure_cache.assert_called_with(self.ctx)
@ -341,14 +341,14 @@ class LoadBalancerDriverTest(base.RugTestBase):
lb.id,
'ACTIVE',
)
self.assertEqual(lb._last_synced_status, 'ACTIVE')
self.assertEqual('ACTIVE', lb._last_synced_status)
@mock.patch('astara.api.astara_client.get_interfaces')
def test_get_interfaces(self, mock_get_interfaces):
mock_get_interfaces.return_value = ['fake_interface']
lb = self._init_driver()
self.assertEqual(
lb.get_interfaces('fake_mgt_addr'), ['fake_interface'])
['fake_interface'], lb.get_interfaces('fake_mgt_addr'))
mock_get_interfaces.assert_called_with(
'fake_mgt_addr', self.mgt_port)
@ -364,7 +364,7 @@ class LoadBalancerDriverTest(base.RugTestBase):
lb = self._init_driver()
self.ctx.neutron.get_loadbalancer_detail.return_value = 'fake_lb'
lb._ensure_cache(self.ctx)
self.assertEqual(lb._loadbalancer, 'fake_lb')
self.assertEqual('fake_lb', lb._loadbalancer)
self.ctx.neutron.get_loadbalancer_detail.assert_called_with(lb.id)
def test__ensure_cache_not_found(self):

44
astara/test/unit/drivers/test_router.py
View File

@ -52,17 +52,17 @@ class RouterDriverTest(base.RugTestBase):
rtr = self._init_driver()
rtr.post_init = mock.Mock()
self.assertEqual(
rtr.name,
'ak-%s-%s' % (rtr.RESOURCE_NAME, self.router_id))
'ak-%s-%s' % (rtr.RESOURCE_NAME, self.router_id),
rtr.name)
mock_post_init.assert_called_with(self.ctx)
@mock.patch('astara.drivers.router.Router._ensure_cache')
def test_post_init(self, mock_ensure_cache):
rtr = self._init_driver()
rtr.post_init(self.ctx)
self.assertEqual(rtr.image_uuid, self.image_uuid)
self.assertEqual(rtr.flavor, self.flavor)
self.assertEqual(rtr.mgt_port, self.mgt_port)
self.assertEqual(self.image_uuid, rtr.image_uuid)
self.assertEqual(self.flavor, rtr.flavor)
self.assertEqual(self.mgt_port, rtr.mgt_port)
mock_ensure_cache.assert_called_with(self.ctx)
def test__ensure_cache_no_router(self):
@ -74,17 +74,17 @@ class RouterDriverTest(base.RugTestBase):
rtr = self._init_driver()
self.ctx.neutron.get_router_detail.return_value = 'fake_router'
rtr._ensure_cache(self.ctx)
self.assertEqual(rtr._router, 'fake_router')
self.assertEqual('fake_router', rtr._router)
def test_ports_no_router(self):
rtr = self._init_driver()
self.assertEqual(rtr.ports, [])
self.assertEqual([], rtr.ports)
def test_ports_with_router(self):
rtr = self._init_driver()
fake_router_obj = fakes.fake_router()
rtr._router = fake_router_obj
self.assertEqual(set(rtr.ports), set(fake_router_obj.ports))
self.assertEqual(set(fake_router_obj.ports), set(rtr.ports))
@mock.patch('astara.drivers.router.Router.pre_plug')
def test_pre_boot(self, mock_pre_plug):
@ -106,7 +106,7 @@ class RouterDriverTest(base.RugTestBase):
mock_build_config.return_value = 'fake_config'
mock_build_config.assert_called_with(
self.ctx, rtr._router, fake_mgt_port, fake_iface_map)
self.assertEqual(res, 'fake_config')
self.assertEqual('fake_config', res)
@mock.patch('astara.api.astara_client.update_config')
def test_update_config(self, mock_update_config):
@ -131,7 +131,7 @@ class RouterDriverTest(base.RugTestBase):
expected_instance_ports = [
'fake_port_%s' % p.network_id for p in fake_router_obj.ports
]
self.assertEqual(res, ('fake_mgt_port', expected_instance_ports))
self.assertEqual(('fake_mgt_port', expected_instance_ports), res)
def test_delete_ports(self):
rtr = self._init_driver()
@ -153,15 +153,15 @@ class RouterDriverTest(base.RugTestBase):
with mock.patch('time.sleep'):
rtr.pre_populate_hook()
self.assertEqual(
neutron_client.get_routers.call_args_list,
[
mock.call(detailed=False)
for value in six.moves.range(len(returned_value))
]
],
neutron_client.get_routers.call_args_list
)
self.assertEqual(
neutron_client.get_routers.call_count,
len(returned_value)
len(returned_value),
neutron_client.get_routers.call_count
)
def _exit_loop_bad_auth(self, mocked_neutron_api, log, exc):
@ -212,14 +212,14 @@ class RouterDriverTest(base.RugTestBase):
id='2',
tenant_id='1',
)
self.assertEqual(res, [expected_resource])
self.assertEqual([expected_resource], res)
def test_get_resource_id_for_tenant(self):
fake_router = fakes.fake_router()
self.ctx.neutron.get_router_for_tenant.return_value = fake_router
res = router.Router.get_resource_id_for_tenant(
self.ctx, 'fake_tenant_id', 'fake_message')
self.assertEqual(res, fake_router.id)
self.assertEqual(fake_router.id, res)
self.ctx.neutron.get_router_for_tenant.assert_called_with(
'fake_tenant_id')
@ -235,7 +235,7 @@ class RouterDriverTest(base.RugTestBase):
tenant_id = 'fake_tenant_id'
res = router.Router.process_notification(
tenant_id, event_type, payload)
self.assertEqual(res, expected)
self.assertEqual(expected, res)
def test_process_notifications_floatingips(self):
payload = {'router': {'id': 'fake_router_id'}}
@ -322,8 +322,8 @@ class RouterDriverTest(base.RugTestBase):
rtr = self._init_driver()
rtr._router = None
self.assertEqual(
rtr.get_state(self.ctx),
states.GONE,
rtr.get_state(self.ctx)
)
mock_ensure_cache.assert_called_with(self.ctx)
@ -333,8 +333,8 @@ class RouterDriverTest(base.RugTestBase):
fake_router = fakes.fake_router()
rtr._router = fake_router
self.assertEqual(
rtr.get_state(self.ctx),
fake_router.status,
rtr.get_state(self.ctx)
)
mock_ensure_cache.assert_called_with(self.ctx)
@ -357,7 +357,7 @@ class RouterDriverTest(base.RugTestBase):
rtr.id,
'ACTIVE',
)
self.assertEqual(rtr._last_synced_status, 'ACTIVE')
self.assertEqual('ACTIVE', rtr._last_synced_status)
@mock.patch('astara.drivers.router.Router._ensure_cache')
def test_synchronize_state_no_change(self, mock_ensure_cache):
@ -374,7 +374,7 @@ class RouterDriverTest(base.RugTestBase):
mock_get_interfaces.return_value = ['fake_interface']
rtr = self._init_driver()
self.assertEqual(
rtr.get_interfaces('fake_mgt_addr'), ['fake_interface'])
['fake_interface'], rtr.get_interfaces('fake_mgt_addr'))
mock_get_interfaces.assert_called_with(
'fake_mgt_addr', self.mgt_port)
@ -393,7 +393,7 @@ class RouterDriverTest(base.RugTestBase):
rtr = self._init_driver()
self.ctx.neutron.get_router_detail.return_value = 'fake_router'
rtr._ensure_cache(self.ctx)
self.assertEqual(rtr._router, 'fake_router')
self.assertEqual('fake_router', rtr._router)
self.ctx.neutron.get_router_detail.assert_called_with(rtr.id)
def test__ensure_cache_not_found(self):

40
astara/test/unit/openvswitch/test_ovs_lib.py
View File

@ -72,11 +72,11 @@ class OVS_Lib_Test(unittest.TestCase):
# test __init__
port = ovs_lib.VifPort(pname, ofport, vif_id, mac, self.br)
self.assertEqual(port.port_name, pname)
self.assertEqual(port.ofport, ofport)
self.assertEqual(port.vif_id, vif_id)
self.assertEqual(port.vif_mac, mac)
self.assertEqual(port.switch.br_name, self.BR_NAME)
self.assertEqual(pname, port.port_name)
self.assertEqual(ofport, port.ofport)
self.assertEqual(vif_id, port.vif_id)
self.assertEqual(mac, port.vif_mac)
self.assertEqual(self.BR_NAME, port.switch.br_name)
# test __str__
foo = str(port)
@ -171,7 +171,7 @@ class OVS_Lib_Test(unittest.TestCase):
root_helper=self.root_helper).AndReturn(datapath_id)
self.mox.ReplayAll()
self.assertEqual(self.br.get_datapath_id(), datapath_id.strip('"'))
self.assertEqual(datapath_id.strip('"'), self.br.get_datapath_id())
self.mox.VerifyAll()
def test_count_flows(self):
@ -181,7 +181,7 @@ class OVS_Lib_Test(unittest.TestCase):
self.mox.ReplayAll()
# counts the number of flows as total lines of output - 2
self.assertEqual(self.br.count_flows(), 1)
self.assertEqual(1, self.br.count_flows())
self.mox.VerifyAll()
def test_delete_flow(self):
@ -224,7 +224,7 @@ class OVS_Lib_Test(unittest.TestCase):
root_helper=self.root_helper).AndReturn(ofport)
self.mox.ReplayAll()
self.assertEqual(self.br.add_tunnel_port(pname, ip), ofport)
self.assertEqual(ofport, self.br.add_tunnel_port(pname, ip))
self.mox.VerifyAll()
def test_add_patch_port(self):
@ -244,7 +244,7 @@ class OVS_Lib_Test(unittest.TestCase):
root_helper=self.root_helper).AndReturn(ofport)
self.mox.ReplayAll()
self.assertEqual(self.br.add_patch_port(pname, peer), ofport)
self.assertEqual(ofport, self.br.add_patch_port(pname, peer))
self.mox.VerifyAll()
def _test_get_vif_ports(self, is_xen=False):
@ -277,11 +277,11 @@ class OVS_Lib_Test(unittest.TestCase):
ports = self.br.get_vif_ports()
self.assertEqual(1, len(ports))
self.assertEqual(ports[0].port_name, pname)
self.assertEqual(ports[0].ofport, ofport)
self.assertEqual(ports[0].vif_id, vif_id)
self.assertEqual(ports[0].vif_mac, mac)
self.assertEqual(ports[0].switch.br_name, self.BR_NAME)
self.assertEqual(pname, ports[0].port_name)
self.assertEqual(ofport, ports[0].ofport)
self.assertEqual(vif_id, ports[0].vif_id)
self.assertEqual(mac, ports[0].vif_mac)
self.assertEqual(self.BR_NAME, ports[0].switch.br_name)
self.mox.VerifyAll()
def test_get_vif_ports_nonxen(self):
@ -308,10 +308,10 @@ class OVS_Lib_Test(unittest.TestCase):
vif_id = match.group('vif_id')
port_name = match.group('port_name')
ofport = int(match.group('ofport'))
self.assertEqual(vif_mac, 'fa:16:3e:23:5b:f2')
self.assertEqual(vif_id, '5c1321a7-c73f-4a77-95e6-9f86402e5c8f')
self.assertEqual(port_name, 'dhc5c1321a7-c7')
self.assertEqual(ofport, 2)
self.assertEqual('fa:16:3e:23:5b:f2', vif_mac)
self.assertEqual('5c1321a7-c73f-4a77-95e6-9f86402e5c8f', vif_id)
self.assertEqual('dhc5c1321a7-c7', port_name)
self.assertEqual(2, ofport)
def test_iface_to_br(self):
iface = 'tap0'
@ -321,7 +321,7 @@ class OVS_Lib_Test(unittest.TestCase):
root_helper=root_helper).AndReturn('br-int')
self.mox.ReplayAll()
self.assertEqual(ovs_lib.get_bridge_for_iface(root_helper, iface), br)
self.assertEqual(br, ovs_lib.get_bridge_for_iface(root_helper, iface))
self.mox.VerifyAll()
def test_iface_to_br_handles_ovs_vsctl_exception(self):
@ -365,5 +365,5 @@ class OVS_Lib_Test(unittest.TestCase):
root_helper=root_helper).AndReturn('br-int\nbr-ex\n')
self.mox.ReplayAll()
self.assertEqual(ovs_lib.get_bridges(root_helper), bridges)
self.assertEqual(bridges, ovs_lib.get_bridges(root_helper))
self.mox.VerifyAll()

2
astara/test/unit/pez/test_pool_manager.py
View File

@ -69,7 +69,7 @@ class PoolManagerTest(base.RugTestBase):
# deletion hasn't timed out yet
self.assertEqual(len(res), 0)
self.assertEqual(0, len(res))
# the deleting instance is added to the counter
self.assertIn(
pool[self.resource][0].id, self.pool_manager._delete_counters)

10
astara/test/unit/test_coordination.py
View File

@ -104,7 +104,7 @@ class TestRugCoordinator(base.RugTestBase):
)
self.fake_coord.get_members.return_value = fake_async_resp
self.coordinator = coordination.RugCoordinator(self.queue)
self.assertEqual(self.coordinator.members, ['foo', 'bar'])
self.assertEqual(['foo', 'bar'], self.coordinator.members)
self.fake_coord.get_members.assert_called_with(self.coordinator.group)
def test_members_filter_leader(self):
@ -113,7 +113,7 @@ class TestRugCoordinator(base.RugTestBase):
)
self.fake_coord.get_members.return_value = fake_async_resp
self.coordinator = coordination.RugCoordinator(self.queue)
self.assertEqual(self.coordinator.members, ['foo', 'bar'])
self.assertEqual(['foo', 'bar'], self.coordinator.members)
self.fake_coord.get_members.assert_called_with(self.coordinator.group)
def test_members_filter_no_leader(self):
@ -122,7 +122,7 @@ class TestRugCoordinator(base.RugTestBase):
)
self.fake_coord.get_members.return_value = fake_async_resp
self.coordinator = coordination.RugCoordinator(self.queue)
self.assertEqual(self.coordinator.members, ['foo', 'bar'])
self.assertEqual(['foo', 'bar'], self.coordinator.members)
self.fake_coord.get_members.assert_called_with(self.coordinator.group)
def test_is_leader(self):
@ -131,7 +131,7 @@ class TestRugCoordinator(base.RugTestBase):
)
self.fake_coord.get_leader.return_value = fake_async_resp
self.coordinator = coordination.RugCoordinator(self.queue)
self.assertEqual(self.coordinator.is_leader, True)
self.assertEqual(True, self.coordinator.is_leader)
self.fake_coord.get_leader.assert_called_with(self.coordinator.group)
@mock.patch('astara.coordination.RugCoordinator.start')
@ -147,4 +147,4 @@ class TestRugCoordinator(base.RugTestBase):
self.coordinator.cluster_changed(event=None)
expected = ('*', expected_rebalance_event)
res = self.queue.get()
self.assertEqual(res, expected)
self.assertEqual(expected, res)

2
astara/test/unit/test_debug.py
View File

@ -65,4 +65,4 @@ class TestDebug(base.RugTestBase):
automaton.return_value.send_message.assert_called_once_with(
CrudMatch('update')
)
self.assertEqual(automaton.return_value.update.call_count, 1)
self.assertEqual(1, automaton.return_value.update.call_count)

142
astara/test/unit/test_instance_manager.py
View File

@ -167,8 +167,8 @@ class TestInstanceManager(base.RugTestBase):
self.update_state_p.stop()
self.fake_driver.get_state.return_value = states.GONE
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.GONE
states.GONE,
self.instance_mgr.update_state(self.ctx)
)
def test_update_state_down_no_backing_instances(self):
@ -176,12 +176,12 @@ class TestInstanceManager(base.RugTestBase):
self.fake_driver.get_state.return_value = states.UP
self.instance_mgr.instances.__nonzero__.return_value = False
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.DOWN
states.DOWN,
self.instance_mgr.update_state(self.ctx)
)
self.assertEqual(
self.instance_mgr.state,
states.DOWN
states.DOWN,
self.instance_mgr.state
)
def test_update_state_degraded(self):
@ -189,12 +189,12 @@ class TestInstanceManager(base.RugTestBase):
self.fake_driver.get_state.return_value = states.UP
self.instance_mgr.instances.cluster_degraded = True
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.DEGRADED
states.DEGRADED,
self.instance_mgr.update_state(self.ctx)
)
self.assertEqual(
self.instance_mgr.state,
states.DEGRADED
states.DEGRADED,
self.instance_mgr.state
)
def test_update_state_booting(self):
@ -203,8 +203,8 @@ class TestInstanceManager(base.RugTestBase):
self.instance_mgr.instances.validate_ports.return_value = \
([], [mock.Mock()]) # (has_ports, no_ports)
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.BOOTING
states.BOOTING,
self.instance_mgr.update_state(self.ctx)
)
def test_update_state_down_all_instances_dead(self):
@ -216,8 +216,8 @@ class TestInstanceManager(base.RugTestBase):
([], [mock.Mock()]) # (alive, dead)
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.DOWN
states.DOWN,
self.instance_mgr.update_state(self.ctx)
)
def test_update_state_degraded_some_instances_dead(self):
@ -229,8 +229,8 @@ class TestInstanceManager(base.RugTestBase):
([mock.Mock()], [mock.Mock()]) # (alive, dead)
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.DEGRADED
states.DEGRADED,
self.instance_mgr.update_state(self.ctx)
)
def test_update_state_up(self):
@ -242,8 +242,8 @@ class TestInstanceManager(base.RugTestBase):
([mock.Mock()], []) # (alive, dead)
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.UP
states.UP,
self.instance_mgr.update_state(self.ctx)
)
def test_update_state_configured(self):
@ -258,21 +258,21 @@ class TestInstanceManager(base.RugTestBase):
([mock.Mock(booting=False)], []) # (alive, dead)
self.assertEqual(
self.instance_mgr.update_state(self.ctx),
states.CONFIGURED
states.CONFIGURED,
self.instance_mgr.update_state(self.ctx)
)
self.instance_mgr.update_state(self.ctx),
self.instance_mgr.update_state(self.ctx),
self.instance_mgr.update_state(self.ctx),
# ensure the boot was logged only once
self.assertEqual(len(self.instance_mgr.log.info.call_args_list), 1)
self.assertEqual(1, len(self.instance_mgr.log.info.call_args_list))
@mock.patch('time.sleep')
def test_boot_success(self, sleep):
self.next_state = states.UP
self.instance_mgr.boot(self.ctx)
self.assertEqual(self.instance_mgr.state, states.BOOTING)
self.assertEqual(states.BOOTING, self.instance_mgr.state)
self.instance_mgr.instances.create.assert_called_with(
self.ctx)
self.assertEqual(1, self.instance_mgr.attempts)
@ -283,13 +283,13 @@ class TestInstanceManager(base.RugTestBase):
self.instance_mgr.boot(self.ctx)
# a deleted VM should reset the vm mgr state and not as a failed
# attempt
self.assertEqual(self.instance_mgr.attempts, 0)
self.assertEqual(0, self.instance_mgr.attempts)
@mock.patch('time.sleep')
def test_boot_exception(self, sleep):
self.instance_mgr.instances.create.side_effect = RuntimeError
self.instance_mgr.boot(self.ctx)
self.assertEqual(self.instance_mgr.state, states.DOWN)
self.assertEqual(states.DOWN, self.instance_mgr.state)
self.instance_mgr.instances.create.assert_called_with(
self.ctx)
self.assertEqual(1, self.instance_mgr.attempts)
@ -307,7 +307,7 @@ class TestInstanceManager(base.RugTestBase):
self.instance_mgr.instances.destroy.assert_called_with(self.ctx)
self.instance_mgr.resource.delete_ports.assert_called_once_with(
self.ctx)
self.assertEqual(self.instance_mgr.state, states.DOWN)
self.assertEqual(states.DOWN, self.instance_mgr.state)
def test_stop_fail(self):
self.instance_mgr.state = states.UP
@ -318,7 +318,7 @@ class TestInstanceManager(base.RugTestBase):
('update_ports', mock.Mock())])
self.instance_mgr.instances.destroy.side_effect = Exception
self.instance_mgr.stop(self.ctx)
self.assertEqual(self.instance_mgr.state, states.UP)
self.assertEqual(states.UP, self.instance_mgr.state)
self.fake_driver.delete_ports.assert_called_with(self.ctx)
def test_stop_router_already_deleted_from_neutron(self):
@ -334,7 +334,7 @@ class TestInstanceManager(base.RugTestBase):
self.instance_mgr.instances.destroy.assert_called_with(self.ctx)
self.instance_mgr.resource.delete_ports.assert_called_once_with(
self.ctx)
self.assertEqual(self.instance_mgr.state, states.GONE)
self.assertEqual(states.GONE, self.instance_mgr.state)
def test_stop_no_inst_router_already_deleted_from_neutron(self):
self.instance_mgr.state = states.GONE
@ -345,7 +345,7 @@ class TestInstanceManager(base.RugTestBase):
('update_ports', mock.Mock())])
self.instance_mgr.stop(self.ctx)
self.fake_driver.delete_ports.assert_called_with(self.ctx)
self.assertEqual(self.instance_mgr.state, states.GONE)
self.assertEqual(states.GONE, self.instance_mgr.state)
def test_stop_instance_already_deleted_from_nova(self):
self.instance_mgr.state = states.RESTART
@ -357,26 +357,26 @@ class TestInstanceManager(base.RugTestBase):
self.instance_mgr.stop(self.ctx)
self.fake_driver.delete_ports.assert_called_with(self.ctx)
self.assertEqual(self.instance_mgr.state, states.DOWN)
self.assertEqual(states.DOWN, self.instance_mgr.state)
def test_configure_mismatched_interfaces(self):
self.instance_mgr.instances.verify_interfaces.return_value = False
self.assertEqual(
self.instance_mgr.configure(self.ctx),
states.REPLUG,
self.instance_mgr.configure(self.ctx)
)
def test_configure_gone(self):
self.fake_driver.get_state.return_value = states.GONE
self.assertEqual(
self.instance_mgr.configure(self.ctx), states.GONE)
states.GONE, self.instance_mgr.configure(self.ctx))
def test_configure(self):
self.instance_mgr.instances.verify_interfaces.return_value = True
self.instance_mgr.instances.configure.return_value = states.RESTART
self.assertEqual(
self.instance_mgr.configure(self.ctx),
states.RESTART,
self.instance_mgr.configure(self.ctx)
)
self.instance_mgr.instances.verify_interfaces.assert_called_with(
self.fake_driver.ports
@ -420,7 +420,7 @@ class TestInstanceManager(base.RugTestBase):
self.ctx.neutron.create_vrrp_port.assert_called_with(
self.fake_driver.id, 'additional-net'
)
self.assertEqual(self.instance_mgr.state, states.REPLUG)
self.assertEqual(states.REPLUG, self.instance_mgr.state)
fake_instance.interface_attach.assert_called_once_with(
fake_new_port.id, None, None
)
@ -458,7 +458,7 @@ class TestInstanceManager(base.RugTestBase):
self.fake_driver.ports.append(fake_new_port)
self.ctx.neutron.create_vrrp_port.return_value = fake_new_port
self.instance_mgr.replug(self.ctx)
self.assertEqual(self.instance_mgr.state, states.RESTART)
self.assertEqual(states.RESTART, self.instance_mgr.state)
fake_instance.interface_attach.assert_called_once_with(
fake_new_port.id, None, None)
@ -505,7 +505,7 @@ class TestInstanceManager(base.RugTestBase):
wait_for_hotplug.return_value = True
self.instance_mgr.replug(self.ctx)
self.assertEqual(self.instance_mgr.state, states.DEGRADED)
self.assertEqual(states.DEGRADED, self.instance_mgr.state)
for instance in instances:
instance.interface_attach.assert_called_with(
@ -553,7 +553,7 @@ class TestInstanceManager(base.RugTestBase):
wait_for_hotplug.side_effect = [True, False]
self.instance_mgr.replug(self.ctx)
self.assertEqual(self.instance_mgr.state, states.DEGRADED)
self.assertEqual(states.DEGRADED, self.instance_mgr.state)
for instance in instances:
instance.interface_attach.assert_called_with(
@ -591,7 +591,7 @@ class TestInstanceManager(base.RugTestBase):
wait_for_hotplug.return_value = True
self.instance_mgr.replug(self.ctx)
self.assertEqual(self.instance_mgr.state, states.REPLUG)
self.assertEqual(states.REPLUG, self.instance_mgr.state)
fake_instance.interface_detach.assert_called_once_with(
fake_add_port.id)
self.assertNotIn(fake_add_port, instance_1.ports)
@ -624,8 +624,8 @@ class TestInstanceManager(base.RugTestBase):
fake_instance.interface_detach.side_effect = Exception
self.instance_mgr.replug(self.ctx)
self.assertEqual(self.instance_mgr.state,
states.RESTART)
self.assertEqual(states.RESTART,
self.instance_mgr.state)
fake_instance.interface_detach.assert_called_once_with(
fake_add_port.id
)
@ -661,8 +661,8 @@ class TestInstanceManager(base.RugTestBase):
wait_for_hotplug.return_value = False
self.instance_mgr.replug(self.ctx)
self.assertEqual(self.instance_mgr.state,
states.RESTART)
self.assertEqual(states.RESTART,
self.instance_mgr.state)
fake_instance.interface_detach.assert_called_once_with(
fake_add_port.id
)
@ -685,9 +685,9 @@ class TestInstanceManager(base.RugTestBase):
],
]
self.assertEqual(
self.instance_mgr._wait_for_interface_hotplug(instance), True)
True, self.instance_mgr._wait_for_interface_hotplug(instance))
self.assertEqual(
len(self.fake_driver.get_interfaces.call_args_list), 3)
3, len(self.fake_driver.get_interfaces.call_args_list))
def test_wait_for_interface_hotplug_false(self):
self.conf.hotplug_timeout = 5
@ -699,9 +699,9 @@ class TestInstanceManager(base.RugTestBase):
]
for i in six.moves.range(5)]
self.assertEqual(
self.instance_mgr._wait_for_interface_hotplug(instance), False)
False, self.instance_mgr._wait_for_interface_hotplug(instance))
self.assertEqual(
len(self.fake_driver.get_interfaces.call_args_list), 4)
4, len(self.fake_driver.get_interfaces.call_args_list))
def test_set_error_when_booting(self):
self.instance_mgr.state = states.BOOTING
@ -728,7 +728,7 @@ class TestInstanceManager(base.RugTestBase):
rtr.ports.__iter__.return_value = []
self.instance_mgr.set_error(self.ctx)
self.instance_mgr.boot(self.ctx)
self.assertEqual(self.instance_mgr.state, states.BOOTING)
self.assertEqual(states.BOOTING, self.instance_mgr.state)
self.instance_mgr.instances.create.assert_called_with(self.ctx)
def test_error_cooldown(self):
@ -766,7 +766,7 @@ class TestInstanceManager(base.RugTestBase):
'fake_instance_id2': fake_inst_2,
}
self.assertEqual(
self.instance_mgr.instances, exp_updated_instances)
exp_updated_instances, self.instance_mgr.instances)
self.instance_mgr.instances.update_ports.assert_called_with(self.ctx)
@ -812,13 +812,13 @@ class TestInstanceGroupManager(base.RugTestBase):
self.fake_driver.is_alive.side_effect = [
False, False, True, False, True]
alive, dead = self.group_mgr.are_alive()
self.assertEqual(sorted(alive), sorted(self.instances))
self.assertEqual(sorted(self.instances), sorted(alive))
def test_are_alive_all_dead(self):
self.fake_driver.is_alive.return_value = False
alive, dead = self.group_mgr.are_alive()
self.assertEqual(sorted(dead), sorted(self.instances))
self.assertEqual(alive, [])
self.assertEqual(sorted(self.instances), sorted(dead))
self.assertEqual([], alive)
def test_are_alive_some_dead(self):
self.group_mgr = instance_manager.InstanceGroupManager(
@ -837,8 +837,8 @@ class TestInstanceGroupManager(base.RugTestBase):
[self.group_mgr.add_instance(i) for i in instances]
self.fake_driver.is_alive = fake_is_alive
alive, dead = self.group_mgr.are_alive()
self.assertEqual(dead, [self.instance_2])
self.assertEqual(alive, [self.instance_1])
self.assertEqual([self.instance_2], dead)
self.assertEqual([self.instance_1], alive)
def test_update_ports(self):
self.ctx.neutron.get_ports_for_instance.side_effect = [
@ -846,10 +846,10 @@ class TestInstanceGroupManager(base.RugTestBase):
('instance2_mgt_port', ['instance2_inst_port']),
]
self.group_mgr.update_ports(self.ctx)
self.assertEqual(self.instance_1.management_port, 'instance1_mgt_port')
self.assertEqual(self.instance_1.ports, ['instance1_inst_port'])
self.assertEqual(self.instance_2.management_port, 'instance2_mgt_port')
self.assertEqual(self.instance_2.ports, ['instance2_inst_port'])
self.assertEqual('instance1_mgt_port', self.instance_1.management_port)
self.assertEqual(['instance1_inst_port'], self.instance_1.ports)
self.assertEqual('instance2_mgt_port', self.instance_2.management_port)
self.assertEqual(['instance2_inst_port'], self.instance_2.ports)
def test_get_interfaces(self):
self.fake_driver.get_interfaces.side_effect = [
@ -948,17 +948,17 @@ class TestInstanceGroupManager(base.RugTestBase):
instance_1_ha_config = self.group_mgr._ha_config(self.instance_1)
instance_2_ha_config = self.group_mgr._ha_config(self.instance_2)
self.assertEqual(
instance_1_ha_config,
{
'priority': 100,
'peers': [self.instance_2.management_address],
})
},
instance_1_ha_config)
self.assertEqual(
instance_2_ha_config,
{
'priority': 50,
'peers': [self.instance_1.management_address],
})
},
instance_2_ha_config)
@mock.patch('astara.instance_manager.InstanceGroupManager._update_config')
@mock.patch('astara.instance_manager.InstanceGroupManager._ha_config')
@ -982,7 +982,7 @@ class TestInstanceGroupManager(base.RugTestBase):
])
fake_update_config.return_value = True
self.assertEqual(self.group_mgr.configure(self.ctx), states.CONFIGURED)
self.assertEqual(states.CONFIGURED, self.group_mgr.configure(self.ctx))
self.assertIn(
mock.call(
self.instance_1,
@ -1023,7 +1023,7 @@ class TestInstanceGroupManager(base.RugTestBase):
])
fake_update_config.return_value = False
self.assertEqual(self.group_mgr.configure(self.ctx), states.RESTART)
self.assertEqual(states.RESTART, self.group_mgr.configure(self.ctx))
@mock.patch('astara.instance_manager.InstanceGroupManager._update_config')
@mock.patch('astara.instance_manager.InstanceGroupManager._ha_config')
@ -1047,7 +1047,7 @@ class TestInstanceGroupManager(base.RugTestBase):
[self.instance_2.management_port]])])
fake_update_config.side_effect = [False, True]
self.assertEqual(self.group_mgr.configure(self.ctx), states.DEGRADED)
self.assertEqual(states.DEGRADED, self.group_mgr.configure(self.ctx))
@mock.patch('astara.instance_manager.InstanceGroupManager._update_config')
@mock.patch('astara.instance_manager.InstanceGroupManager._ha_config')
@ -1069,7 +1069,7 @@ class TestInstanceGroupManager(base.RugTestBase):
])
fake_update_config.return_value = True
self.assertEqual(self.group_mgr.configure(self.ctx), states.DEGRADED)
self.assertEqual(states.DEGRADED, self.group_mgr.configure(self.ctx))
def test_delete(self):
self.group_mgr.delete(self.instance_2)
@ -1106,13 +1106,13 @@ class TestInstanceGroupManager(base.RugTestBase):
def test_next_instance_index(self):
self.assertEqual(
self.group_mgr.next_instance_index, 2)
2, self.group_mgr.next_instance_index)
def test_next_instance_index_empty(self):
group_mgr = instance_manager.InstanceGroupManager(
log=mock.Mock(), resource=self.fake_driver)
self.assertEqual(
group_mgr.next_instance_index, 0)
0, group_mgr.next_instance_index)
def test_create_all(self):
[self.group_mgr.delete(i) for i in self.instances]
@ -1122,7 +1122,7 @@ class TestInstanceGroupManager(base.RugTestBase):
]
self.group_mgr.create(self.ctx)
self.assertEqual(
len(self.ctx.nova_client.boot_instance.call_args_list), 2)
2, len(self.ctx.nova_client.boot_instance.call_args_list))
def test_create_some(self):
self.group_mgr.delete(self.instance_1)
@ -1131,7 +1131,7 @@ class TestInstanceGroupManager(base.RugTestBase):
]
self.group_mgr.create(self.ctx)
self.assertEqual(
len(self.ctx.nova_client.boot_instance.call_args_list), 1)
1, len(self.ctx.nova_client.boot_instance.call_args_list))
self.ctx.nova_client.boot_instance.assert_called_with(
resource_type=self.fake_driver.RESOURCE_NAME,
prev_instance_info=None,
@ -1143,12 +1143,12 @@ class TestInstanceGroupManager(base.RugTestBase):
def test_required_instance_count(self):
self.fake_driver.is_ha = True
self.assertEqual(self.group_mgr.required_instance_count, 2)
self.assertEqual(2, self.group_mgr.required_instance_count)
self.fake_driver.is_ha = False
self.assertEqual(self.group_mgr.required_instance_count, 1)
self.assertEqual(1, self.group_mgr.required_instance_count)
def test_instance_count(self):
self.assertEqual(self.group_mgr.instance_count, 2)
self.assertEqual(2, self.group_mgr.instance_count)
def test_cluster_degraded_false(self):
self.assertFalse(self.group_mgr.cluster_degraded)

8
astara/test/unit/test_main.py
View File

@ -75,8 +75,8 @@ class TestMainPippo(base.RugTestBase):
notifications.NoopPublisher = mock.Mock(
spec=ak_notifications.NoopPublisher)
main.main(argv=self.argv)
self.assertEqual(len(notifications.Publisher.mock_calls), 0)
self.assertEqual(len(notifications.NoopPublisher.mock_calls), 2)
self.assertEqual(0, len(notifications.Publisher.mock_calls))
self.assertEqual(2, len(notifications.NoopPublisher.mock_calls))
@mock.patch('astara.main.shuffle_notifications')
def test_ceilometer_enabled(self, shuffle_notifications, health,
@ -87,5 +87,5 @@ class TestMainPippo(base.RugTestBase):
notifications.NoopPublisher = mock.Mock(
spec=ak_notifications.NoopPublisher)
main.main(argv=self.argv)
self.assertEqual(len(notifications.Publisher.mock_calls), 2)
self.assertEqual(len(notifications.NoopPublisher.mock_calls), 0)
self.assertEqual(2, len(notifications.Publisher.mock_calls))
self.assertEqual(0, len(notifications.NoopPublisher.mock_calls))

10
astara/test/unit/test_notifications.py
View File

@ -129,7 +129,7 @@ class TestGetCRUD(base.RugTestBase):
# message was discarded and not queued
return None
tenant, event = self.queue.get()
self.assertEqual(tenant, fake_tenant_id)
self.assertEqual(fake_tenant_id, tenant)
return event
def _get_event_l3_rpc(self, method, **kwargs):
@ -144,7 +144,7 @@ class TestGetCRUD(base.RugTestBase):
if not self.queue.qsize():
return None
tenant, event = self.queue.get()
self.assertEqual(tenant, fake_tenant_id)
self.assertEqual(fake_tenant_id, tenant)
return event
def test_rpc_router_deleted(self):
@ -152,7 +152,7 @@ class TestGetCRUD(base.RugTestBase):
method='router_deleted',
router_id='fake_router_id')
self.assertEqual(event.DELETE, e.crud)
self.assertEqual(e.resource.id, 'fake_router_id')
self.assertEqual('fake_router_id', e.resource.id)
def test_notification_port(self):
e = self._get_event_notification('port.create.start')
@ -209,8 +209,8 @@ class TestGetCRUD(base.RugTestBase):
}
}
e = self._get_event_notification('router.create.end', payload)
self.assertEqual(e.resource.id,
u'f95fb32d-0072-4675-b4bd-61d829a46aca')
self.assertEqual(u'f95fb32d-0072-4675-b4bd-61d829a46aca',
e.resource.id)
def test_interface_create_and_delete(self):
for action in ('create', 'delete'):

6
astara/test/unit/test_populate.py
View File

@ -74,8 +74,8 @@ class TestPrePopulateWorkers(base.RugTestBase):
name='PrePopulateWorkers'
)
self.assertEqual(
t.mock_calls,
[mock.call.setDaemon(True), mock.call.start()]
[mock.call.setDaemon(True), mock.call.start()],
t.mock_calls
)
@mock.patch('astara.drivers.enabled_drivers')
@ -91,4 +91,4 @@ class TestPrePopulateWorkers(base.RugTestBase):
enabled_drivers.return_value = drivers
res = populate.repopulate()
self.assertEqual(
set(res), set(['driver_0_resource', 'driver_1_resource']))
set(['driver_0_resource', 'driver_1_resource']), set(res))

14
astara/test/unit/test_rpc.py
View File

@ -36,8 +36,8 @@ class TestRPC(testtools.TestCase):
def test__deprecated_amqp_url(self):
self.config(amqp_url='amqp://stackrabbit:secretrabbit@127.0.0.1:/')
self.assertEqual(rpc._deprecated_amqp_url(),
'rabbit://stackrabbit:secretrabbit@127.0.0.1:5672/')
self.assertEqual('rabbit://stackrabbit:secretrabbit@127.0.0.1:5672/',
rpc._deprecated_amqp_url())
@mock.patch('oslo_messaging.get_transport')
@mock.patch.object(rpc, '_deprecated_amqp_url')
@ -45,7 +45,7 @@ class TestRPC(testtools.TestCase):
fake_amqp_url.return_value = 'fake_url'
fake_get_transport.return_value = 'fake_transport'
transport = rpc.get_transport()
self.assertEqual(transport, 'fake_transport')
self.assertEqual('fake_transport', transport)
fake_get_transport.assert_called_with(conf=cfg.CONF, url='fake_url')
@mock.patch.object(rpc, 'get_transport')
@ -55,7 +55,7 @@ class TestRPC(testtools.TestCase):
fake_get_server.return_value = 'fake_server'
fake_endpoints = [1, 2]
result = rpc.get_server(target='fake_target', endpoints=fake_endpoints)
self.assertEqual(result, 'fake_server')
self.assertEqual('fake_server', result)
fake_get_server.assert_called_with(
transport='fake_transport',
target='fake_target',
@ -73,7 +73,7 @@ class TestRPC(testtools.TestCase):
'server': 'fake_server',
}
result = rpc.get_target(**target_args)
self.assertEqual(result, 'fake_target')
self.assertEqual('fake_target', result)
fake_target.assert_called_with(**target_args)
@mock.patch.object(rpc, 'get_transport')
@ -90,7 +90,7 @@ class TestRPC(testtools.TestCase):
topic='foo_target', exchange='foo_exchange', version='2.5',
fanout=False,
)
self.assertEqual(res, 'fake_rpc_client')
self.assertEqual('fake_rpc_client', res)
fake_client.assert_called_with(
'fake_transport', 'fake_target'
)
@ -101,7 +101,7 @@ class TestRPC(testtools.TestCase):
fake_get_transport.return_value = 'fake_transport'
fake_notifier.return_value = 'fake_rpc_notifier'
res = rpc.get_rpc_notifier(topic='foo_topic')
self.assertEqual(res, 'fake_rpc_notifier')
self.assertEqual('fake_rpc_notifier', res)
fake_notifier.assert_called_with(
transport='fake_transport', driver='messaging', topic='foo_topic')

8
astara/test/unit/test_scheduler.py
View File

@ -47,11 +47,11 @@ class TestScheduler(unittest.TestCase):
s.stop()
for w in s.workers:
self.assertEqual(
w['queue'].put.call_args_list,
[mock.call(None), mock.call(None)] # one put for each worker
[mock.call(None), mock.call(None)],
w['queue'].put.call_args_list # one put for each worker
)
self.assertEqual(w['queue'].close.call_count, 2)
self.assertEqual(w['worker'].join.call_count, 2)
self.assertEqual(2, w['queue'].close.call_count)
self.assertEqual(2, w['worker'].join.call_count)
class TestDispatcher(unittest.TestCase):

74
astara/test/unit/test_state.py
View File

@ -60,14 +60,14 @@ class BaseTestStateCase(unittest.TestCase):
class TestBaseState(BaseTestStateCase):
def test_execute(self):
self.assertEqual(
self.state.execute('action', self.ctx),
'action'
'action',
self.state.execute('action', self.ctx)
)
def test_transition(self):
self.assertEqual(
self.state.transition('action', self.ctx),
self.state
self.state,
self.state.transition('action', self.ctx)
)
@ -78,10 +78,10 @@ class TestCalcActionState(BaseTestStateCase):
leftover=0, initial_action=event.POLL):
self.params.queue = deque(queue_states)
self.assertEqual(
self.state.execute(initial_action, self.ctx),
expected_action
expected_action,
self.state.execute(initial_action, self.ctx)
)
self.assertEqual(len(self.params.queue), leftover)
self.assertEqual(leftover, len(self.params.queue))
def test_execute_empty_queue(self):
self._test_hlpr('testaction', [], initial_action='testaction')
@ -227,8 +227,8 @@ class TestAliveState(BaseTestStateCase):
def test_execute(self):
self.assertEqual(
self.state.execute('passthrough', self.ctx),
'passthrough'
'passthrough',
self.state.execute('passthrough', self.ctx)
)
self.instance.update_state.assert_called_once_with(self.ctx)
@ -273,16 +273,16 @@ class TestCreateInstanceState(BaseTestStateCase):
def test_execute(self):
self.instance.attempts = 0
self.assertEqual(
self.state.execute('passthrough', self.ctx),
'passthrough'
'passthrough',
self.state.execute('passthrough', self.ctx)
)
self.instance.boot.assert_called_once_with(self.ctx)
def test_execute_too_many_attempts(self):
self.instance.attempts = self.params.reboot_error_threshold
self.assertEqual(
self.state.execute('passthrough', self.ctx),
'passthrough'
'passthrough',
self.state.execute('passthrough', self.ctx)
)
self.assertEqual([], self.instance.boot.mock_calls)
self.instance.set_error.assert_called_once_with(self.ctx)
@ -318,16 +318,16 @@ class TestRebuildInstanceState(BaseTestStateCase):
def test_execute(self):
self.assertEqual(
self.state.execute('ignored', self.ctx),
event.CREATE,
self.state.execute('ignored', self.ctx)
)
self.instance.stop.assert_called_once_with(self.ctx)
def test_execute_gone(self):
self.instance.state = states.GONE
self.assertEqual(
self.state.execute('ignored', self.ctx),
event.DELETE,
self.state.execute('ignored', self.ctx)
)
self.instance.stop.assert_called_once_with(self.ctx)
@ -337,16 +337,16 @@ class TestClearErrorState(BaseTestStateCase):
def test_execute(self):
self.assertEqual(
self.state.execute('passthrough', self.ctx),
'passthrough',
self.state.execute('passthrough', self.ctx)
)
self.instance.clear_error.assert_called_once_with(self.ctx)
def test_execute_after_error(self):
self.instance.state = states.ERROR
self.assertEqual(
self.state.execute('passthrough', self.ctx),
'passthrough',
self.state.execute('passthrough', self.ctx)
)
self.instance.clear_error.assert_called_once_with(self.ctx)
@ -370,8 +370,8 @@ class TestCheckBootState(BaseTestStateCase):
def test_execute(self):
self.assertEqual(
self.state.execute('passthrough', self.ctx),
'passthrough'
'passthrough',
self.state.execute('passthrough', self.ctx)
)
self.instance.update_state.assert_called_once_with(self.ctx)
assert list(self.params.queue) == ['passthrough']
@ -403,8 +403,8 @@ class TestStopInstanceState(BaseTestStateCase):
def test_execute(self):
self.assertEqual(
self.state.execute('passthrough', self.ctx),
'passthrough'
'passthrough',
self.state.execute('passthrough', self.ctx)
)
self.instance.stop.assert_called_once_with(self.ctx)
@ -427,8 +427,8 @@ class TestReplugState(BaseTestStateCase):
def test_execute(self):
self.assertEqual(
self.state.execute('update', self.ctx),
'update'
'update',
self.state.execute('update', self.ctx)
)
self.instance.replug.assert_called_once_with(self.ctx)
@ -456,20 +456,20 @@ class TestConfigureInstanceState(BaseTestStateCase):
def test_execute_read_configure_success(self):
self.instance.state = states.CONFIGURED
self.assertEqual(self.state.execute(event.READ, self.ctx),
event.READ)
self.assertEqual(event.READ,
self.state.execute(event.READ, self.ctx))
self.instance.configure.assert_called_once_with(self.ctx)
def test_execute_update_configure_success(self):
self.instance.state = states.CONFIGURED
self.assertEqual(self.state.execute(event.UPDATE, self.ctx),
event.POLL)
self.assertEqual(event.POLL,
self.state.execute(event.UPDATE, self.ctx))
self.instance.configure.assert_called_once_with(self.ctx)
def test_execute_configure_failure(self):
self.assertEqual(
self.state.execute(event.CREATE, self.ctx),
event.CREATE
event.CREATE,
self.state.execute(event.CREATE, self.ctx)
)
self.instance.configure.assert_called_once_with(self.ctx)
@ -510,8 +510,8 @@ class TestReadStatsState(BaseTestStateCase):
self.instance.read_stats.return_value = 'foo'
self.assertEqual(
self.state.execute(event.READ, self.ctx),
event.POLL
event.POLL,
self.state.execute(event.READ, self.ctx)
)
self.instance.read_stats.assert_called_once_with()
self.params.bandwidth_callback.assert_called_once_with('foo')
@ -549,7 +549,7 @@ class TestAutomaton(unittest.TestCase):
message.crud = 'update'
with mock.patch.object(self.sm.resource, 'log') as logger:
self.sm.send_message(message)
self.assertEqual(len(self.sm._queue), 1)
self.assertEqual(1, len(self.sm._queue))
logger.debug.assert_called_with(
'incoming message brings queue length to %s',
1,
@ -572,7 +572,7 @@ class TestAutomaton(unittest.TestCase):
message.crud = 'update'
self.sm.deleted = True
self.sm.send_message(message)
self.assertEqual(len(self.sm._queue), 0)
self.assertEqual(0, len(self.sm._queue))
self.assertFalse(self.sm.has_more_work())
def test_send_message_in_error(self):
@ -581,14 +581,14 @@ class TestAutomaton(unittest.TestCase):
message = mock.Mock()
message.crud = 'poll'
self.sm.send_message(message)
self.assertEqual(len(self.sm._queue), 0)
self.assertEqual(0, len(self.sm._queue))
self.assertFalse(self.sm.has_more_work())
# Non-POLL events should *not* be ignored for routers in ERROR state
message.crud = 'create'
with mock.patch.object(self.sm.resource, 'log') as logger:
self.sm.send_message(message)
self.assertEqual(len(self.sm._queue), 1)
self.assertEqual(1, len(self.sm._queue))
logger.debug.assert_called_with(
'incoming message brings queue length to %s',
1,
@ -603,14 +603,14 @@ class TestAutomaton(unittest.TestCase):
message.crud = 'rebuild'
message.body = {'image_uuid': 'ABC123'}
self.sm.send_message(message)
self.assertEqual(self.sm.image_uuid, 'ABC123')
self.assertEqual('ABC123', self.sm.image_uuid)
# rebuilds with image default.
message = mock.Mock()
message.crud = 'rebuild'
message.body = {}
self.sm.send_message(message)
self.assertEqual(self.sm.image_uuid, self.fake_driver.image_uuid)
self.assertEqual(self.fake_driver.image_uuid, self.sm.image_uuid)
def test_has_more_work(self):
with mock.patch.object(self.sm, '_queue'):

21
astara/test/unit/test_tenant.py
View File

@ -67,7 +67,7 @@ class TestTenantResourceManager(base.RugTestBase):
self.fake_load_resource.return_value = fakes.fake_driver(
resource_id='5678')
sm = self.trm.get_state_machines(msg, self.ctx)[0]
self.assertEqual(sm.resource_id, '5678')
self.assertEqual('5678', sm.resource_id)
self.assertIn('5678', self.trm.state_machines)
def test_get_state_machine_no_resoruce_id(self):
@ -240,7 +240,7 @@ class TestTenantResourceManager(base.RugTestBase):
self.trm._default_resource_id = 'abcd'
self.trm.state_machines['5678'] = mock.Mock()
self.trm._delete_resource(r)
self.assertEqual(self.trm.state_machines.values(), [])
self.assertEqual([], self.trm.state_machines.values())
r = event.Resource(
tenant_id='1234',
id='5678',
@ -252,7 +252,7 @@ class TestTenantResourceManager(base.RugTestBase):
body={'key': 'value'},
)
sms = self.trm.get_state_machines(msg, self.ctx)
self.assertEqual(sms, [])
self.assertEqual([], sms)
self.assertIn('5678', self.trm.state_machines.deleted)
def test_deleter_callback(self):
@ -297,9 +297,8 @@ class TestTenantResourceManager(base.RugTestBase):
fake_sm = mock.Mock()
self.trm.state_machines['fake_resource_id'] = fake_sm
self.assertEqual(
self.trm.get_state_machine_by_resource_id('fake_resource_id'),
fake_sm
)
fake_sm,
self.trm.get_state_machine_by_resource_id('fake_resource_id'))
def test_unmanage_resource(self):
fake_sm = mock.Mock()
@ -329,8 +328,8 @@ class TestTenantResourceManager(base.RugTestBase):
fake_get.return_value = fake_driver
self.assertEqual(
self.trm._load_resource_from_message(self.ctx, msg),
'fake_driver')
'fake_driver',
self.trm._load_resource_from_message(self.ctx, msg))
fake_get.assert_called_with(msg.resource.driver)
fake_driver.assert_called_with(self.ctx, msg.resource.id)
self.assertFalse(fake_byonf.called)
@ -355,7 +354,7 @@ class TestTenantResourceManager(base.RugTestBase):
self.ctx.neutron.tenant_has_byo_for_function.return_value = 'byonf_res'
self.assertEqual(
self.trm._load_resource_from_message(self.ctx, msg), fake_driver)
fake_driver, self.trm._load_resource_from_message(self.ctx, msg))
fake_byonf.assert_called_with(
self.ctx, 'byonf_res', msg.resource.id)
self.assertFalse(fake_get.called)
@ -382,6 +381,6 @@ class TestTenantResourceManager(base.RugTestBase):
self.ctx.neutron.tenant_has_byo_for_function.return_value = None
self.assertEqual(
self.trm._load_resource_from_message(self.ctx, msg),
'fake_fallback_driver')
'fake_fallback_driver',
self.trm._load_resource_from_message(self.ctx, msg))
fake_get.assert_called_with(msg.resource.driver)

48
astara/test/unit/test_worker.py
View File

@ -382,12 +382,12 @@ class TestWorker(WorkerTestBase):
self.config(host='foohost')
ctxt = worker.WorkerContext(fakes.FAKE_MGT_ADDR)
self.assertEqual(
ctxt.config,
{
'host': 'foohost',
'metadata_port': 1234,
'address': fakes.FAKE_MGT_ADDR,
})
},
ctxt.config)
@mock.patch('astara.worker.Worker._get_trms')
def test__get_all_state_machines(self, fake_get_trms):
@ -404,8 +404,8 @@ class TestWorker(WorkerTestBase):
fake_get_trms.return_value = trms
res = self.w._get_all_state_machines()
self.assertEqual(
res,
set(['sm1', 'sm2', 'sm3', 'sm4'])
set(['sm1', 'sm2', 'sm3', 'sm4']),
res
)
@mock.patch('astara.worker.hash_ring', autospec=True)
@ -418,10 +418,10 @@ class TestWorker(WorkerTestBase):
def test__defer_message(self):
self.assertEqual(
self.w._deferred_messages, [])
[], self.w._deferred_messages)
self.w._defer_message(self.target, self.msg)
self.assertEqual(
self.w._deferred_messages, [(self.target, self.msg)])
[(self.target, self.msg)], self.w._deferred_messages)
@mock.patch('astara.worker.Worker.handle_message')
def test__replay_deferred_messages_none(self, fakehandle):
@ -440,7 +440,7 @@ class TestWorker(WorkerTestBase):
self.w._replay_deferred_messages()
exp_calls = [mock.call(t, m) for t, m in msgs]
self.assertEqual(
fake_handle.call_args_list, exp_calls)
exp_calls, fake_handle.call_args_list)
class TestResourceCache(WorkerTestBase):
@ -463,7 +463,7 @@ class TestResourceCache(WorkerTestBase):
msg = event.Event(resource=r, crud=event.UPDATE, body={})
res = self.resource_cache.get_by_tenant(
resource=r, worker_context=self.worker_context, message=msg)
self.assertEqual(res, 'fake_cached_resource_id')
self.assertEqual('fake_cached_resource_id', res)
self.assertFalse(self.w._context.neutron.get_router_for_tenant.called)
def test_resource_cache_miss(self):
@ -481,7 +481,7 @@ class TestResourceCache(WorkerTestBase):
resource=r,
worker_context=self.worker_context,
message=msg)
self.assertEqual(res, 'fake_fetched_resource_id')
self.assertEqual('fake_fetched_resource_id', res)
self.w._context.neutron.get_router_for_tenant.assert_called_with(
'fake_tenant_id')
@ -501,8 +501,8 @@ class TestResourceCache(WorkerTestBase):
worker_context=self.worker_context,
message=msg)
self.assertEqual(
self.resource_cache._tenant_resources[r.driver][r.tenant_id],
r.id)
r.id,
self.resource_cache._tenant_resources[r.driver][r.tenant_id])
self.resource_cache.delete(r)
self.assertNotIn(
r.tenant_id,
@ -543,7 +543,7 @@ class TestCreatingResource(WorkerTestBase):
trm = self.w.tenant_managers[self.tenant_id]
sm = trm.get_state_machines(self.msg, worker.WorkerContext(
fakes.FAKE_MGT_ADDR))[0]
self.assertEqual(len(sm._queue), 1)
self.assertEqual(1, len(sm._queue))
class TestWildcardMessages(WorkerTestBase):
@ -579,12 +579,12 @@ class TestWildcardMessages(WorkerTestBase):
def test_wildcard_to_all(self):
trms = self.w._get_trms('*')
ids = sorted(trm.tenant_id for trm in trms)
self.assertEqual(ids, [self.tenant_id_1, self.tenant_id_2])
self.assertEqual([self.tenant_id_1, self.tenant_id_2], ids)
def test_wildcard_to_error(self):
trms = self.w._get_trms('error')
ids = sorted(trm.tenant_id for trm in trms)
self.assertEqual(ids, [self.tenant_id_1, self.tenant_id_2])
self.assertEqual([self.tenant_id_1, self.tenant_id_2], ids)
class TestShutdown(WorkerTestBase):
@ -686,7 +686,7 @@ class TestDebugRouters(WorkerTestBase):
self.w._should_process_command = mock.MagicMock(return_value=self.msg)
def testNoDebugs(self):
self.assertEqual(self.dbapi.resources_in_debug(), set())
self.assertEqual(set(), self.dbapi.resources_in_debug())
def testWithDebugs(self):
self.w.handle_message(
@ -718,7 +718,7 @@ class TestDebugRouters(WorkerTestBase):
'resource_id': 'this-resource-id'}),
)
self.assert_not_in_debug(resource_id='this-resource-id')
self.assertEqual(lock.release.call_count, 1)
self.assertEqual(1, lock.release.call_count)
def testManageNoLock(self):
self.enable_debug(resource_id='this-resource-id')
@ -771,7 +771,7 @@ class TestDebugTenants(WorkerTestBase):
self.w._should_process_command = mock.MagicMock(return_value=self.msg)
def testNoDebugs(self):
self.assertEqual(self.dbapi.tenants_in_debug(), set())
self.assertEqual(set(), self.dbapi.tenants_in_debug())
def testWithDebugs(self):
self.enable_debug(tenant_id='this-tenant-id')
@ -840,14 +840,14 @@ class TestNormalizeUUID(unittest.TestCase):
def test_upper(self):
self.assertEqual(
'ac194fc5-f317-412e-8611-fb290629f624',
worker._normalize_uuid(
'ac194fc5-f317-412e-8611-fb290629f624'.upper()),
'ac194fc5-f317-412e-8611-fb290629f624')
'ac194fc5-f317-412e-8611-fb290629f624'.upper()))
def test_no_dashes(self):
self.assertEqual(
worker._normalize_uuid('ac194fc5f317412e8611fb290629f624'),
'ac194fc5-f317-412e-8611-fb290629f624')
'ac194fc5-f317-412e-8611-fb290629f624',
worker._normalize_uuid('ac194fc5f317412e8611fb290629f624'))
class TestGlobalDebug(WorkerTestBase):
@ -991,9 +991,9 @@ class TestRebalance(WorkerTestBase):
self.w.scheduler.dispatcher.pick_workers = mock.Mock(return_value=tgt)
self.w._repopulate()
post_rebalance_sms = self.w._get_all_state_machines()
self.assertEqual(len(post_rebalance_sms), 1)
self.assertEqual(1, len(post_rebalance_sms))
sm = post_rebalance_sms.pop()
self.assertEqual(sm.resource_id, rsc2.id)
self.assertEqual(rsc2.id, sm.resource_id)
@mock.patch('astara.populate.repopulate')
def test__repopulate_sm_added(self, fake_repopulate):
@ -1040,7 +1040,7 @@ class TestRebalance(WorkerTestBase):
self.w.scheduler.dispatcher.pick_workers = mock.Mock(return_value=tgt)
self.w._repopulate()
post_rebalance_sms = self.w._get_all_state_machines()
self.assertEqual(len(post_rebalance_sms), 3)
self.assertEqual(3, len(post_rebalance_sms))
rids = [r.id for r in resources]
for sm in post_rebalance_sms:
self.assertIn(sm.resource_id, rids)