def callback1(resource, event, trigger, **kwargs):
- print 'Callback1 called by trigger: ', trigger
- print 'kwargs: ', kwargs
+ print('Callback1 called by trigger: ', trigger)
+ print('kwargs: ', kwargs)
def callback2(resource, event, trigger, **kwargs):
- print 'Callback2 called by trigger: ', trigger
- print 'kwargs: ', kwargs
+ print('Callback2 called by trigger: ', trigger)
+ print('kwargs: ', kwargs)
# B and C express interest with I
registry.subscribe(callback1, resources.ROUTER, events.BEFORE_CREATE)
registry.subscribe(callback2, resources.ROUTER, events.BEFORE_CREATE)
- print 'Subscribed'
+ print('Subscribed')
# A notifies
registry.notify(resources.ROUTER, events.BEFORE_CREATE, do_notify, **kwargs)
- print 'Notifying...'
+ print('Notifying...')
do_notify()
raise Exception('I am failing!')
def callback2(resource, event, trigger, **kwargs):
- print 'Callback2 called by %s on event %s' % (trigger, event)
+ print('Callback2 called by %s on event %s' % (trigger, event))
registry.subscribe(callback1, resources.ROUTER, events.BEFORE_CREATE)
registry.subscribe(callback2, resources.ROUTER, events.BEFORE_CREATE)
registry.subscribe(callback2, resources.ROUTER, events.ABORT_CREATE)
- print 'Subscribed'
+ print('Subscribed')
def do_notify():
registry.notify(resources.ROUTER, events.BEFORE_CREATE, do_notify, **kwargs)
- print 'Notifying...'
+ print('Notifying...')
try:
do_notify()
except exceptions.CallbackFailure as e:
- print 'Error: ', e
+ print('Error: ', e)
The output is:
def callback1(resource, event, trigger, **kwargs):
- print 'Callback1 called by %s on event %s for resource %s' % (trigger, event, resource)
+ print('Callback1 called by %s on event %s for resource %s' % (trigger, event, resource))
def callback2(resource, event, trigger, **kwargs):
- print 'Callback2 called by %s on event %s for resource %s' % (trigger, event, resource)
+ print('Callback2 called by %s on event %s for resource %s' % (trigger, event, resource))
registry.subscribe(callback1, resources.ROUTER, events.BEFORE_READ)
registry.subscribe(callback1, resources.ROUTER, events.AFTER_DELETE)
registry.subscribe(callback1, resources.PORT, events.BEFORE_UPDATE)
registry.subscribe(callback2, resources.ROUTER_GATEWAY, events.BEFORE_UPDATE)
- print 'Subscribed'
+ print('Subscribed')
def do_notify():
- print 'Notifying...'
+ print('Notifying...')
kwargs = {'foo': 'bar'}
registry.notify(resources.ROUTER, events.BEFORE_READ, do_notify, **kwargs)
registry.notify(resources.ROUTER, events.BEFORE_CREATE, do_notify, **kwargs)
def callback1(resource, event, trigger, **kwargs):
- print 'module callback'
+ print('module callback')
class MyCallback(object):
def callback2(self, resource, event, trigger, **kwargs):
- print 'object callback'
+ print('object callback')
@classmethod
def callback3(cls, resource, event, trigger, **kwargs):
- print 'class callback'
+ print('class callback')
c = MyCallback()
def do_notify():
def nested_subscribe(resource, event, trigger, **kwargs):
- print 'nested callback'
+ print('nested callback')
registry.subscribe(nested_subscribe, resources.ROUTER, events.BEFORE_CREATE)
registry.notify(resources.ROUTER, events.BEFORE_CREATE, do_notify, **kwargs)
- print 'Notifying...'
+ print('Notifying...')
do_notify()
And the output is going to be:
>>> time.sleep(5)
>>> proc.stop()
>>> for line in proc.iter_stdout():
- ... print line
+ ... print(line)
"""
def __init__(self, cmd, run_as_root=False, respawn_interval=None,
m.start()
for iterable in m:
event = IPMonitorEvent.from_text(iterable)
- print event, event.added, event.interface, event.cidr
+ print(event, event.added, event.interface, event.cidr)
"""
def __init__(self,
def __unicode__(self):
return unicode(self.msg)
+ def __str__(self):
+ return self.msg
+
def use_fatal_exceptions(self):
return False
raise TypeError(_("TLD '%s' must not be all numeric") % names[-1])
except TypeError as e:
msg = _("'%(data)s' not a valid PQDN or FQDN. Reason: %(reason)s") % {
- 'data': data, 'reason': e.message}
+ 'data': data, 'reason': str(e)}
return msg
# Leave part of the bridge name on for easier identification
hashlen = 6
namelen = n_const.DEVICE_NAME_MAX_LEN - len(prefix) - hashlen
+ if isinstance(name, six.text_type):
+ hashed_name = hashlib.sha1(name.encode('utf-8'))
+ else:
+ hashed_name = hashlib.sha1(name)
new_name = ('%(prefix)s%(truncated)s%(hash)s' %
{'prefix': prefix, 'truncated': name[0:namelen],
- 'hash': hashlib.sha1(name).hexdigest()[0:hashlen]})
+ 'hash': hashed_name.hexdigest()[0:hashlen]})
LOG.warning(_LW("Creating an interface named %(name)s exceeds the "
"%(limit)d character limitation. It was shortened to "
"%(new_name)s to fit."),
self.pd_update.append(prefix_update)
for intf in intfs:
for subnet in intf['subnets']:
- if subnet['id'] == prefix_update.keys()[0]:
+ if subnet['id'] in prefix_update:
# Update the prefix
- subnet['cidr'] = prefix_update.values()[0]
+ subnet['cidr'] = prefix_update[subnet['id']]
# Process the router for removed interfaces
agent.pd.notifier = pd_notifier
external_process call is followed with either an enable() or disable()
'''
- num_ext_calls = len(expected) / 2
+ num_ext_calls = len(expected) // 2
expected_ext_calls = []
actual_ext_calls = []
expected_action_calls = []
db_pools.append(db_pool)
return db_pools
- def _validate_ips(self, pool, db_pool):
- self.assertEqual(pool[0], db_pool.first_ip)
- self.assertEqual(pool[1], db_pool.last_ip)
+ def _validate_ips(self, pools, db_pool):
+ self.assertTrue(
+ any(pool == (db_pool.first_ip, db_pool.last_ip) for pool in pools))
def test_create_pool(self):
db_pools = self._create_pools([self.single_pool])
ipam_pool = self.ctx.session.query(db_models.IpamAllocationPool).\
filter_by(ipam_subnet_id=self.ipam_subnet_id).first()
- self._validate_ips(self.single_pool, ipam_pool)
+ self._validate_ips([self.single_pool], ipam_pool)
range = self.ctx.session.query(db_models.IpamAvailabilityRange).\
filter_by(allocation_pool_id=db_pools[0].id).first()
- self._validate_ips(self.single_pool, range)
+ self._validate_ips([self.single_pool], range)
def _test_get_first_range(self, locking):
self._create_pools(self.multi_pool)
range = self.subnet_manager.get_first_range(self.ctx.session,
locking=locking)
- self._validate_ips(self.multi_pool[0], range)
+ self._validate_ips(self.multi_pool, range)
def test_get_first_range(self):
self._test_get_first_range(False)
db_pools[0].id).all()
self.assertEqual(1, len(db_ranges))
self.assertEqual(db_models.IpamAvailabilityRange, type(db_ranges[0]))
- self._validate_ips(self.single_pool, db_ranges[0])
+ self._validate_ips([self.single_pool], db_ranges[0])
def test_create_range(self):
self._create_pools([self.single_pool])
pool = self.ctx.session.query(db_models.IpamAllocationPool).\
filter_by(ipam_subnet_id=self.ipam_subnet_id).first()
- self._validate_ips(self.single_pool, pool)
+ self._validate_ips([self.single_pool], pool)
allocation_pool_id = pool.id
# delete the range
db_range = self.subnet_manager.list_ranges_by_allocation_pool(
self.ctx.session,
pool.id).first()
- self._validate_ips(self.single_pool, db_range)
+ self._validate_ips([self.single_pool], db_range)
self.ctx.session.delete(db_range)
# create a new range
from oslo_versionedobjects import fields as obj_fields
from neutron.common import exceptions as n_exc
+from neutron.common import utils as common_utils
from neutron import context
from neutron.db import api as db_api
from neutron.objects import base
def _validate_objects(self, expected, observed):
self.assertTrue(all(self._is_test_class(obj) for obj in observed))
self.assertEqual(
- sorted(expected),
- sorted(get_obj_db_fields(obj) for obj in observed))
+ sorted(expected,
+ key=common_utils.safe_sort_key),
+ sorted([get_obj_db_fields(obj) for obj in observed],
+ key=common_utils.safe_sort_key))
def _check_equal(self, obj, db_obj):
self.assertEqual(
import mock
from oslo_config import cfg
+import six
import six.moves.urllib.request as urlrequest
import testtools
import webob
response = open_no_proxy('http://127.0.0.1:%d/' % server.port)
- self.assertEqual(greetings, response.read())
+ self.assertEqual(greetings.encode('utf-8'), response.read())
server.stop()
self.assertEqual(expected_json, result)
+ # TODO(cbrandily): support this test in py3K
+ @testtools.skipIf(six.PY3, "bug/1491824")
def test_json_with_utf8(self):
input_dict = dict(servers=dict(a=(2, '\xe7\xbd\x91\xe7\xbb\x9c')))
expected_json = b'{"servers":{"a":[2,"\\u7f51\\u7edc"]}}'
exception.MalformedRequestBody, deserializer.default, data_string)
def test_json_with_utf8(self):
- data = '{"a": "\xe7\xbd\x91\xe7\xbb\x9c"}'
+ data = b'{"a": "\xe7\xbd\x91\xe7\xbb\x9c"}'
as_dict = {'body': {'a': u'\u7f51\u7edc'}}
deserializer = wsgi.JSONDeserializer()
self.assertEqual(as_dict,
deserializer.deserialize(data))
def test_json_with_unicode(self):
- data = '{"a": "\u7f51\u7edc"}'
+ data = b'{"a": "\u7f51\u7edc"}'
as_dict = {'body': {'a': u'\u7f51\u7edc'}}
deserializer = wsgi.JSONDeserializer()
self.assertEqual(as_dict,
class TestWSGIServerWithSSL(base.BaseTestCase):
"""WSGI server tests."""
+ def setUp(self):
+ super(TestWSGIServerWithSSL, self).setUp()
+ if six.PY3:
+ self.skip("bug/1482633")
+
@mock.patch("exceptions.RuntimeError")
@mock.patch("os.path.exists")
def test__check_ssl_settings(self, exists_mock, runtime_error_mock):
[testenv:docs]
commands = sphinx-build -W -b html doc/source doc/build/html
-[testenv:py34]
-commands = python setup.py test --testr-args='{posargs: \
- neutron.tests.unit.test_context \
- neutron.tests.unit.services.metering.drivers.test_iptables \
- neutron.tests.unit.services.metering.agents.test_metering_agent \
- neutron.tests.unit.services.test_provider_configuration \
- neutron.tests.unit.plugins.ml2.drivers.mech_sriov.agent.test_sriov_nic_agent \
- neutron.tests.unit.plugins.ml2.drivers.mech_sriov.agent.test_eswitch_manager \
- neutron.tests.unit.plugins.ml2.drivers.mech_sriov.agent.common.test_config \
- neutron.tests.unit.plugins.ml2.drivers.mech_sriov.agent.test_pci_lib \
- neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent.ovs_test_base \
- neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl.test_br_phys \
- neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl.test_br_int \
- neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl.test_br_tun \
- neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent.test_ovs_tunnel \
- neutron.tests.unit.plugins.brocade.test_brocade_db \
- neutron.tests.unit.plugins.brocade.test_brocade_plugin \
- neutron.tests.unit.plugins.brocade.test_brocade_vlan \
- neutron.tests.unit.plugins.embrane.test_embrane_neutron_plugin \
- neutron.tests.unit.plugins.oneconvergence.test_nvsd_agent \
- neutron.tests.unit.plugins.oneconvergence.test_nvsd_plugin \
- neutron.tests.unit.plugins.oneconvergence.test_plugin_helper \
- neutron.tests.unit.plugins.oneconvergence.test_nvsdlib \
- neutron.tests.unit.plugins.ibm.test_sdnve_agent \
- neutron.tests.unit.plugins.ibm.test_sdnve_api \
- neutron.tests.unit.plugins.ml2.test_db \
- neutron.tests.unit.plugins.ml2.test_driver_context \
- neutron.tests.unit.plugins.ml2.test_port_binding \
- neutron.tests.unit.plugins.ml2.test_extension_driver_api \
- neutron.tests.unit.plugins.ml2.test_rpc \
- neutron.tests.unit.plugins.ml2.drivers.mlnx.test_mech_mlnx \
- neutron.tests.unit.plugins.ml2.drivers.openvswitch.mech_driver.test_mech_openvswitch \
- neutron.tests.unit.plugins.ml2.drivers.linuxbridge.mech_driver.test_mech_linuxbridge \
- neutron.tests.unit.plugins.ml2.drivers.linuxbridge.agent.test_linuxbridge_neutron_agent \
- neutron.tests.unit.plugins.ml2.drivers.base_type_tunnel \
- neutron.tests.unit.plugins.ml2.drivers.opendaylight.test_driver \
- neutron.tests.unit.plugins.ml2.drivers.ext_test \
- neutron.tests.unit.plugins.ml2.drivers.mech_sriov.mech_driver.test_mech_sriov_nic_switch \
- neutron.tests.unit.plugins.ml2.drivers.mech_fake_agent \
- neutron.tests.unit.plugins.ml2.drivers.test_type_vxlan \
- neutron.tests.unit.plugins.ml2.drivers.test_type_gre \
- neutron.tests.unit.plugins.ml2.drivers.test_helpers \
- neutron.tests.unit.plugins.ml2.drivers.test_type_local \
- neutron.tests.unit.plugins.ml2.drivers.mechanism_logger \
- neutron.tests.unit.plugins.ml2.drivers.test_type_flat \
- neutron.tests.unit.plugins.ml2.drivers.test_type_vlan \
- neutron.tests.unit.plugins.ml2.drivers.mechanism_test \
- neutron.tests.unit.plugins.ml2.drivers.l2pop.rpc_manager.l2population_rpc_base \
- neutron.tests.unit.plugins.ml2.extensions.fake_extension \
- neutron.tests.unit.plugins.ml2.drivers.l2pop.rpc_manager.test_l2population_rpc \
- neutron.tests.unit.plugins.ml2.drivers.l2pop.test_mech_driver \
- neutron.tests.unit.plugins.cisco.n1kv.test_n1kv_db \
- neutron.tests.unit.plugins.cisco.n1kv.fake_client \
- neutron.tests.unit.plugins.cisco.test_network_db \
- neutron.tests.unit.quota.test_resource \
- neutron.tests.unit.quota.test_resource_registry \
- neutron.tests.unit.scheduler.test_l3_agent_scheduler \
- neutron.tests.unit.scheduler.test_dhcp_agent_scheduler \
- neutron.tests.unit.db.test_agentschedulers_db \
- neutron.tests.unit.db.test_allowedaddresspairs_db \
- neutron.tests.unit.db.test_db_base_plugin_v2 \
- neutron.tests.unit.db.test_ipam_backend_mixin \
- neutron.tests.unit.db.test_l3_dvr_db \
- neutron.tests.unit.db.test_l3_hamode_db \
- neutron.tests.unit.db.test_ipam_pluggable_backend \
- neutron.tests.unit.db.test_migration \
- neutron.tests.unit.db.test_agents_db \
- neutron.tests.unit.db.quota.test_api \
- neutron.tests.unit.db.quota.test_driver \
- neutron.tests.unit.db.test_dvr_mac_db \
- neutron.tests.unit.db.test_securitygroups_db \
- neutron.tests.unit.debug.test_commands \
- neutron.tests.unit.tests.test_post_mortem_debug \
- neutron.tests.unit.tests.test_base \
- neutron.tests.unit.database_stubs \
- neutron.tests.unit.dummy_plugin \
- neutron.tests.unit.extension_stubs \
- neutron.tests.unit.testlib_api \
- neutron.tests.unit.api.test_api_common \
- neutron.tests.unit.api.rpc.handlers.test_dhcp_rpc \
- neutron.tests.unit.api.rpc.handlers.test_securitygroups_rpc \
- neutron.tests.unit.api.rpc.handlers.test_dvr_rpc \
- neutron.tests.unit.api.rpc.agentnotifiers.test_dhcp_rpc_agent_api \
- neutron.tests.unit.api.v2.test_attributes \
- neutron.tests.unit.agent.metadata.test_agent \
- neutron.tests.unit.agent.metadata.test_driver \
- neutron.tests.unit.agent.metadata.test_namespace_proxy \
- neutron.tests.unit.agent.test_rpc \
- neutron.tests.unit.agent.test_securitygroups_rpc \
- neutron.tests.unit.agent.l3.test_link_local_allocator \
- neutron.tests.unit.agent.l3.test_dvr_local_router \
- neutron.tests.unit.agent.l3.test_ha_router \
- neutron.tests.unit.agent.l3.test_legacy_router \
- neutron.tests.unit.agent.l3.test_router_info \
- neutron.tests.unit.agent.l3.test_router_processing_queue \
- neutron.tests.unit.agent.l3.test_namespace_manager \
- neutron.tests.unit.agent.l3.test_dvr_fip_ns \
- neutron.tests.unit.agent.ovsdb.native.test_helpers \
- neutron.tests.unit.agent.common.test_config \
- neutron.tests.unit.agent.common.test_ovs_lib \
- neutron.tests.unit.agent.common.test_polling \
- neutron.tests.unit.agent.common.test_utils \
- neutron.tests.unit.agent.linux.test_ip_lib \
- neutron.tests.unit.agent.linux.test_keepalived \
- neutron.tests.unit.agent.linux.test_daemon \
- neutron.tests.unit.agent.linux.test_ipset_manager \
- neutron.tests.unit.agent.linux.test_iptables_firewall \
- neutron.tests.unit.agent.linux.test_ebtables_manager \
- neutron.tests.unit.agent.linux.test_iptables_firewall \
- neutron.tests.unit.agent.linux.test_ebtables_driver \
- neutron.tests.unit.agent.linux.test_polling \
- neutron.tests.unit.agent.linux.test_ip_lib \
- neutron.tests.unit.agent.linux.test_ip_monitor \
- neutron.tests.unit.agent.linux.test_iptables_manager \
- neutron.tests.unit.agent.linux.test_external_process \
- neutron.tests.unit.agent.linux.test_dhcp \
- neutron.tests.unit.agent.linux.test_async_process \
- neutron.tests.unit.agent.linux.test_ovsdb_monitor \
- neutron.tests.unit.agent.linux.test_bridge_lib \
- neutron.tests.unit.agent.linux.test_ip_link_support \
- neutron.tests.unit.agent.linux.test_interface \
- neutron.tests.unit.agent.linux.test_utils \
- neutron.tests.unit.agent.dhcp.test_agent \
- neutron.tests.unit.test_manager \
- neutron.tests.unit.test_service \
- neutron.tests.unit.test_auth \
- neutron.tests.unit.test_policy \
- neutron.tests.unit.extensions.v2attributes \
- neutron.tests.unit.extensions.test_address_scope \
- neutron.tests.unit.extensions.test_agent \
- neutron.tests.unit.extensions.test_external_net \
- neutron.tests.unit.extensions.test_flavors \
- neutron.tests.unit.extensions.test_l3_ext_gw_mode \
- neutron.tests.unit.extensions.test_extra_dhcp_opt \
- neutron.tests.unit.extensions.test_extraroute \
- neutron.tests.unit.extensions.test_netmtu \
- neutron.tests.unit.extensions.test_vlantransparent \
- neutron.tests.unit.extensions.extendedattribute \
- neutron.tests.unit.extensions.base \
- neutron.tests.unit.extensions.foxinsocks \
- neutron.tests.unit.extensions.extensionattribute \
- neutron.tests.unit.extensions.test_servicetype \
- neutron.tests.unit.extensions.test_portsecurity \
- neutron.tests.unit.extensions.test_providernet \
- neutron.tests.unit.callbacks.test_manager \
- neutron.tests.unit.hacking.test_checks \
- neutron.tests.unit.common.test_utils \
- neutron.tests.unit.common.test_config \
- neutron.tests.unit.common.test_rpc \
- neutron.tests.unit.common.test_ipv6_utils \
- neutron.tests.unit.cmd.test_ovs_cleanup \
- neutron.tests.unit.cmd.test_netns_cleanup \
- neutron.tests.unit.ipam.drivers.neutrondb_ipam.test_db_api \
- neutron.tests.unit.ipam.drivers.neutrondb_ipam.test_driver \
- neutron.tests.unit.ipam.test_subnet_alloc \
- neutron.tests.unit.ipam.test_utils \
- neutron.tests.unit.ipam.test_requests \
- neutron.tests.unit.notifiers.test_nova \
- neutron.tests.unit.notifiers.test_batch_notifier \
- neutron.tests.unit.api.test_extensions \
- neutron.tests.unit.db.test_db_base_plugin_common}'
-
[flake8]
# E125 continuation line does not distinguish itself from next logical line
# E126 continuation line over-indented for hanging indent