Fix superfluous-parens pylint warnings

I had noticed a few of these had snuck-in over the
past cycle, let's fix them and enable the pylint
check.

TrivialFix

Assisted-by: Claude Sonnet 4.6
Signed-off-by: Brian Haley <haleyb.dev@gmail.com>
Change-Id: I8fcd95fd8be0328277c794280900f8fb3d65ccbb
This commit is contained in:
Brian Haley committed 2026-06-15 21:36:19 -04:00
1 parent d8de4a7e54
commit b877e9b4bd
48 files changed
+92 -92

No files matched your search

-1
View File
@@ -83,7 +83,6 @@ disable=
invalid-name,
missing-docstring,
singleton-comparison,
superfluous-parens,
ungrouped-imports,
wrong-import-order,
consider-using-f-string,
+1 -1
View File
@@ -231,7 +231,7 @@ class DHCPIPv6Responder(dhcp_base.DHCPResponderBase):
# Client FQDN: host-<ip-v6-address>
fqdn_bin = struct.pack('!%ds' % len(fqdn), bytes(str(fqdn).encode()))
fqdn_str_len = struct.pack('!b', len(fqdn_bin))
dns_data = (dns_tag + fqdn_str_len + fqdn_bin)
dns_data = dns_tag + fqdn_str_len + fqdn_bin
option_list.append(
dhcp6.option(code=DHCPV6_OPTION_FQDN,
data=dns_data, length=len(dns_data)))
@@ -27,7 +27,7 @@ class FipPriority:
def __eq__(self, other):
if isinstance(other, FipPriority):
return (self.index == other.index)
return self.index == other.index
return False
def __int__(self):
+1 -1
View File
@@ -329,7 +329,7 @@ class IPWrapper(SubProcessBase):
if dstport:
kwargs['vxlan_port'] = dstport
privileged.create_interface(name, self.namespace, "vxlan", **kwargs)
return (IPDevice(name, namespace=self.namespace))
return IPDevice(name, namespace=self.namespace)
class IPDevice(SubProcessBase):
+1 -1
View File
@@ -100,7 +100,7 @@ class IpsetManager:
else:
add_ips = self._get_new_set_ips(set_name, member_ips)
del_ips = self._get_deleted_set_ips(set_name, member_ips)
if (len(add_ips) + len(del_ips) < IPSET_ADD_BULK_THRESHOLD):
if len(add_ips) + len(del_ips) < IPSET_ADD_BULK_THRESHOLD:
self._add_members_to_set(set_name, add_ips)
self._del_members_from_set(set_name, del_ips)
else:
+1 -1
View File
@@ -1015,7 +1015,7 @@ class IptablesFirewallDriver(firewall.FirewallDriver):
self.pre_sg_members, sg_id, ethertype)
cur_ips = self._get_sg_members(
self.sg_members, sg_id, ethertype)
ips = (pre_ips - cur_ips)
ips = pre_ips - cur_ips
if devices and ips:
self.ipconntrack.delete_conntrack_state_by_remote_ips(
devices, ethertype, ips)
+1 -1
View File
@@ -225,7 +225,7 @@ class IptablesTable:
def _wrap_target_chain(self, s, wrap):
if s.startswith('$'):
s = (f'{self.wrap_name}-{get_chain_name(s[1:], wrap)}')
s = f'{self.wrap_name}-{get_chain_name(s[1:], wrap)}'
return s
@@ -29,4 +29,4 @@ class OVSFWTagNotFound(exceptions.NeutronException):
class OVSFWPortNotHandled(exceptions.NeutronException):
message = ("Port %(port_id)s is not handled by the firewall.")
message = "Port %(port_id)s is not handled by the firewall."
+1 -1
View File
@@ -132,6 +132,6 @@ def apply_metadata_nat_rules(router, proxy):
if netutils.is_ipv6_enabled():
for c, r in metadata_nat_rules(
proxy.metadata_port,
metadata_address=(constants.METADATA_V6_CIDR)):
metadata_address=constants.METADATA_V6_CIDR):
router.iptables_manager.ipv6['nat'].add_rule(c, r)
router.iptables_manager.apply()
+1 -1
View File
@@ -210,7 +210,7 @@ class Controller:
if not fields_to_strip:
return data
return dict(item for item in data.items()
if (item[0] not in fields_to_strip))
if item[0] not in fields_to_strip)
def _do_field_list(self, original_fields):
fields_to_add = None
+1 -1
View File
@@ -36,7 +36,7 @@ def is_eui64_address(ip_address):
ip = netaddr.IPAddress(ip_address)
# '0xfffe' addition is used to build EUI-64 from MAC (RFC4291)
# Look for it in the middle of the EUI-64 part of address
return ip.version == 6 and not ((ip & 0xffff000000) ^ 0xfffe000000)
return ip.version == 6 and not (ip & 0xffff000000) ^ 0xfffe000000
def is_ipv6_pd_enabled(subnet):
+1 -1
View File
@@ -963,7 +963,7 @@ def timecost(f):
call_id = uuidutils.generate_uuid()
message_base = ("Time-cost: call %(call_id)s function %(fname)s ") % {
"call_id": call_id, "fname": f.__name__}
end_message = (message_base + "took %(seconds).3fs seconds to run")
end_message = message_base + "took %(seconds).3fs seconds to run"
@timeutils.time_it(LOG, message=end_message, min_duration=None)
def wrapper(*args, **kwargs):
+1 -1
View File
@@ -41,7 +41,7 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
Using is_accessible expresses these conditions.
"""
if not (policy_object.QosPolicy.is_accessible(context, old_policy)):
if not policy_object.QosPolicy.is_accessible(context, old_policy):
raise qos_exc.PolicyRemoveAuthorizationError(
policy_id=old_policy.id)
+1 -1
View File
@@ -240,7 +240,7 @@ class AgentDbMixin(ext_agent.AgentPluginBase, AgentAvailabilityZoneMixin):
configs = agent.get('configurations', {})
load_type = None
load = 0
if (agent['agent_type'] == constants.AGENT_TYPE_DHCP):
if agent['agent_type'] == constants.AGENT_TYPE_DHCP:
load_type = cfg.CONF.dhcp_load_type
if load_type:
load = int(configs.get(load_type, 0))
+2 -2
View File
@@ -401,7 +401,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
'OpenStack uses the EUI-64 address format, '
'which requires the prefix to be /64')
raise exc.InvalidInput(
error_message=(msg % subnet['cidr']))
error_message=msg % subnet['cidr'])
def _validate_ipv6_combination(self, ra_mode, address_mode):
if ra_mode != address_mode:
@@ -811,7 +811,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
def _validate_subnet_for_pd(self, subnet):
"""Validates that subnet parameters are correct for IPv6 PD"""
if (subnet.get('ip_version') != constants.IP_VERSION_6):
if subnet.get('ip_version') != constants.IP_VERSION_6:
reason = _("Prefix Delegation can only be used with IPv6 "
"subnets.")
raise exc.BadRequest(resource='subnets', msg=reason)
+1 -1
View File
@@ -76,7 +76,7 @@ def get_node(context, group_name, node_uuid):
@db_api.retry_if_session_inactive()
def remove_nodes_from_host(context, group_name, created_at=None):
with (db_api.CONTEXT_WRITER.using(context)):
with db_api.CONTEXT_WRITER.using(context):
query = context.session.query(ovn_models.OVNHashRing).filter(
ovn_models.OVNHashRing.hostname == CONF.host,
ovn_models.OVNHashRing.group_name == group_name)
+1 -1
View File
@@ -66,7 +66,7 @@ class DbQuotaNoLockDriver(quota_driver.DbQuotaDriver):
unlimited_resources = {
resource for (resource, limit) in limits.items()
if limit <= quota_api.UNLIMITED_QUOTA}
requested_resources = (set(deltas.keys()) - unlimited_resources)
requested_resources = set(deltas.keys()) - unlimited_resources
# Count the number of (1) used and (2) reserved resources for this
# project_id. If any resource limit is exceeded, raise exception.
+1 -1
View File
@@ -783,7 +783,7 @@ class SecurityGroupDbMixin(
constants.PROTO_NUM_IPV6_ICMP]:
for attr, field in [('port_range_min', 'type'),
('port_range_max', 'code')]:
if rule[attr] is not None and not (0 <= rule[attr] <= 255):
if rule[attr] is not None and not 0 <= rule[attr] <= 255:
raise ext_sg.SecurityGroupInvalidIcmpValue(
field=field, attr=attr, value=rule[attr])
if (rule['port_range_min'] is None and
+1 -1
View File
@@ -187,7 +187,7 @@ def check_builtins_gettext(logical_line, tokens, filename, lines, noqa):
def check_no_imports_from_tests(logical_line, filename, noqa):
"""N343 - Production code must not import from neutron.tests.*
"""
msg = ("N343: Production code must not import from neutron.tests.*")
msg = "N343: Production code must not import from neutron.tests.*"
if noqa:
return
+3 -3
View File
@@ -73,7 +73,7 @@ class Reservation(base.NeutronDbObject):
def delete_expired(cls, context, expiring_time, project_id):
resv_query = context.session.query(models.Reservation)
if project_id:
project_expr = (models.Reservation.project_id == project_id)
project_expr = models.Reservation.project_id == project_id
else:
project_expr = sql.true()
# TODO(manjeets) Fetch and delete objects using
@@ -95,9 +95,9 @@ class Reservation(base.NeutronDbObject):
sql.func.sum(models.ResourceDelta.amount),
sqltypes.Integer)).join(models.Reservation)
if expired:
exp_expr = (models.Reservation.expiration < now)
exp_expr = models.Reservation.expiration < now
else:
exp_expr = (models.Reservation.expiration >= now)
exp_expr = models.Reservation.expiration >= now
resv_query = resv_query.filter(sa.and_(
models.Reservation.project_id == project_id,
models.ResourceDelta.resource.in_(resources),
@@ -223,7 +223,7 @@ class CommonAgentLoop(service.Service):
if device_info.get('removed'):
resync_b = self.treat_devices_removed(device_info['removed'])
# If one of the above operations fails => resync with plugin
return (resync_a | resync_b)
return resync_a | resync_b
def treat_devices_added_updated(self, devices):
try:
@@ -478,8 +478,8 @@ class CommonAgentLoop(service.Service):
sync = True
# sleep till end of polling interval
elapsed = (time.time() - start)
if (elapsed < self.polling_interval):
elapsed = time.time() - start
if elapsed < self.polling_interval:
time.sleep(self.polling_interval - elapsed)
else:
LOG.debug("Loop iteration exceeded interval "
@@ -108,7 +108,7 @@ class L2populationMechanismDriver(api.MechanismDriver):
def _fixed_ips_changed(self, context, orig, port, diff_ips):
orig_ips, port_ips = diff_ips
if (port['device_owner'] == const.DEVICE_OWNER_DVR_INTERFACE):
if port['device_owner'] == const.DEVICE_OWNER_DVR_INTERFACE:
agent_host = context.host
else:
agent_host = context.original_host
@@ -300,7 +300,7 @@ class SriovNicSwitchAgent:
if device_info.get('removed'):
resync_b = self.treat_devices_removed(device_info['removed'])
# If one of the above operations fails => resync with plugin
return (resync_a | resync_b)
return resync_a | resync_b
def treat_device(self, device_info, admin_state_up, spoofcheck=True,
propagate_uplink_state=False):
@@ -521,8 +521,8 @@ class SriovNicSwitchAgent:
self.activated_bindings |= activated_bindings_copy
# sleep till end of polling interval
elapsed = (time.time() - start)
if (elapsed < self.polling_interval):
elapsed = time.time() - start
if elapsed < self.polling_interval:
self.daemon_loop_event.wait(self.polling_interval - elapsed)
else:
LOG.debug("Loop iteration exceeded interval "
@@ -2306,7 +2306,7 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
# Update the list of current ports storing only those which
# have been actually processed.
skipped_devices = set(skipped_devices)
port_info['current'] = (port_info['current'] - skipped_devices)
port_info['current'] = port_info['current'] - skipped_devices
# TODO(salv-orlando): Optimize avoiding applying filters
# unnecessarily, (eg: when there are no IP address changes)
@@ -2635,7 +2635,7 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
# AlwaysPoll used by windows implementations
# REVISIT (rossella_s) This needs to be reworked to hide implementation
# details regarding polling in BasePollingManager subclasses
if sync or not (hasattr(polling_manager, 'get_events')):
if sync or not hasattr(polling_manager, 'get_events'):
if sync:
LOG.info("Agent out of sync with plugin!")
consecutive_resyncs = consecutive_resyncs + 1
@@ -155,7 +155,7 @@ class OpenvswitchMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
# NOTE(sean-k-mooney): this function converts the ovs vhost user
# driver mode into the qemu vhost user mode. If OVS is the server,
# qemu is the client and vice-versa.
if (ovs_const.OVS_DPDK_VHOST_USER_CLIENT in iface_types):
if ovs_const.OVS_DPDK_VHOST_USER_CLIENT in iface_types:
return portbindings.VHOST_USER_MODE_SERVER
return portbindings.VHOST_USER_MODE_CLIENT
@@ -1245,7 +1245,7 @@ class OVNMechanismDriver(api.MechanismDriver):
vif_details = copy.deepcopy(self.vif_details[vif_type])
vif_details[portbindings.VHOST_USER_SOCKET] = (
vhost_user_socket)
elif (vnic_type == portbindings.VNIC_VIRTIO_FORWARDER):
elif vnic_type == portbindings.VNIC_VIRTIO_FORWARDER:
vhost_user_socket = ovn_utils.ovn_vhu_sockpath(
ovn_conf.get_ovn_vhost_sock_dir(), port['id'])
vif_type = portbindings.VIF_TYPE_AGILIO_OVS
@@ -2701,7 +2701,7 @@ class OVNClient:
if 'options' in new_options and 'options' in original_options:
orig_dns_server = original_options['options'].get('dns_server')
new_dns_server = new_options['options'].get('dns_server')
dns_server_changed = (orig_dns_server != new_dns_server)
dns_server_changed = orig_dns_server != new_dns_server
else:
dns_server_changed = False
+1 -1
View File
@@ -242,7 +242,7 @@ class TypeManager(stevedore.named.NamedExtensionManager):
segmentation_id = net_data.get(provider.SEGMENTATION_ID)
network_type = segment[api.NETWORK_TYPE]
if network_type != constants.TYPE_VLAN:
msg = (_('Only VLAN type networks can be updated.'))
msg = _('Only VLAN type networks can be updated.')
raise exc.InvalidInput(error_message=msg)
if not segmentation_id:
msg = (_('Only %s field can be updated in VLAN type networks') %
+2 -2
View File
@@ -2679,7 +2679,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
network_db = self._get_network(context, network_id)
network_db.mtu = self._get_network_mtu(
network_db,
validate=(event != events.PRECOMMIT_DELETE))
validate=event != events.PRECOMMIT_DELETE)
network_db.save(session=context.session)
try:
@@ -2886,7 +2886,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
port_db.port_bindings, host)
if not original_binding:
raise exc.PortBindingNotFound(port_id=port_id, host=host)
is_active_binding = (original_binding.status == const.ACTIVE)
is_active_binding = original_binding.status == const.ACTIVE
network = self.get_network(context, port_db['network_id'])
port_dict = self._make_port_dict(port_db)
mech_context = driver_context.PortContext(self, context, port_dict,
+1 -1
View File
@@ -147,7 +147,7 @@ class NDPProxyPlugin(l3_ndp_proxy.NDPProxyBase):
return False
v6_fixed_ips = [
fixed_ip for fixed_ip in port_dict['fixed_ips']
if (netaddr.IPNetwork(fixed_ip['ip_address']).version == V6)]
if netaddr.IPNetwork(fixed_ip['ip_address']).version == V6]
# If the router's external gateway port user LLA address, The
# external network needn't IPv6 subnet.
if v6_fixed_ips:
+1 -1
View File
@@ -172,7 +172,7 @@ class PlacementReportPlugin(service_base.ServicePluginBase):
errors = True
placement_error_str = \
're-parenting a provider is not currently allowed'
if (placement_error_str in str(e)):
if placement_error_str in str(e):
msg = (
'placement client call failed'
' (this may be due to bug'
+1 -1
View File
@@ -134,7 +134,7 @@ class SegmentDbMixin:
# NOTE(xiaohhui): The new index is the last index + 1, this
# may cause discontinuous segment_index. But segment_index
# can functionally work as the order index for segments.
segment_index = (segments[-1].get('segment_index') + 1)
segment_index = segments[-1].get('segment_index') + 1
args['segment_index'] = segment_index
new_segment = network.NetworkSegment(context, **args)
@@ -5510,7 +5510,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
gateway_ip='10.0.1.1',
cidr='10.0.1.11/24') as v2:
subnets = (v1, v2)
query_params = ('cidr=10.0.0.11/24&cidr=10.0.1.11/24')
query_params = 'cidr=10.0.0.11/24&cidr=10.0.1.11/24'
self._test_list_resources('subnet', subnets,
query_params=query_params)
@@ -7414,7 +7414,7 @@ class TestNetworks(testlib_api.SqlTestCase):
network['network']['shared'] = False
if (expected_exception):
if expected_exception:
with testlib_api.ExpectedException(expected_exception):
plugin.update_network(ctx, net_id, network)
else:
@@ -346,7 +346,7 @@ class L3AgentTestFramework(base.BaseSudoTestCase):
router_info=None):
router_info = router_info or self.generate_router_info(
enable_ha, ip_version, dual_stack=dual_stack,
v6_ext_gw_with_sub=(v6_ext_gw_with_sub))
v6_ext_gw_with_sub=v6_ext_gw_with_sub)
return_copy = copy.deepcopy(router_info)
router = self.manage_router(self.agent, router_info)
@@ -562,7 +562,7 @@ class L3AgentTestFramework(base.BaseSudoTestCase):
def _assert_metadata_chains(self, router):
def metadata_port_filter(rule):
return (str(self.agent.conf.metadata_port) in rule.rule)
return str(self.agent.conf.metadata_port) in rule.rule
self.assertTrue(self._get_rule(router.iptables_manager,
'nat',
@@ -39,7 +39,7 @@ class BaseMonitorTest(linux_base.BaseOVSLinuxTestCase):
def setUp(self):
super().setUp()
rootwrap_not_configured = (cfg.CONF.AGENT.root_helper == base.SUDO_CMD)
rootwrap_not_configured = cfg.CONF.AGENT.root_helper == base.SUDO_CMD
if rootwrap_not_configured:
# The monitor tests require a nested invocation that has
# to be emulated by double sudo if rootwrap is not
@@ -581,7 +581,7 @@ class RouteTestCase(functional_base.BaseSudoTestCase):
scope = 0
routes = priv_ip_lib.list_ip_routes(self.namespace, ip_version)
for route in routes:
if not (linux_utils.get_attr(route, 'RTA_GATEWAY') == gateway):
if not linux_utils.get_attr(route, 'RTA_GATEWAY') == gateway:
continue
self.assertEqual(table, route['table'])
self.assertEqual(
@@ -80,7 +80,7 @@ class L3SchedulerBaseTest(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
def _create_routers(self, scheduled_router_count,
expected_scheduled_router_count):
routers = []
if (scheduled_router_count + expected_scheduled_router_count):
if scheduled_router_count + expected_scheduled_router_count:
for i in range(scheduled_router_count +
expected_scheduled_router_count):
router = self._create_router('schd_rtr' + str(i))
+1 -1
View File
@@ -2112,7 +2112,7 @@ class TestDeviceManager(base.BaseTestCase):
self.ensure_device_is_ready_p = mock.patch(
'neutron.agent.linux.ip_lib.ensure_device_is_ready')
self.ensure_device_is_ready = (self.ensure_device_is_ready_p.start())
self.ensure_device_is_ready = self.ensure_device_is_ready_p.start()
self.dvr_cls_p = mock.patch('neutron.agent.linux.interface.NullDriver')
self.iproute_cls_p = mock.patch('neutron.agent.linux.'
+25 -25
View File
@@ -43,57 +43,57 @@ NETNS_SAMPLE = [
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc']
GATEWAY_SAMPLE1 = ("""
GATEWAY_SAMPLE1 = """
default via 10.35.19.254 metric 100
10.35.16.0/22 proto kernel scope link src 10.35.17.97
""")
"""
GATEWAY_SAMPLE2 = ("""
GATEWAY_SAMPLE2 = """
default via 10.35.19.254 metric 100
""")
"""
GATEWAY_SAMPLE3 = ("""
GATEWAY_SAMPLE3 = """
10.35.16.0/22 proto kernel scope link src 10.35.17.97
""")
"""
GATEWAY_SAMPLE4 = ("""
GATEWAY_SAMPLE4 = """
default via 10.35.19.254
""")
"""
GATEWAY_SAMPLE5 = ("""
GATEWAY_SAMPLE5 = """
default via 192.168.99.1 proto static
""")
"""
GATEWAY_SAMPLE6 = ("""
GATEWAY_SAMPLE6 = """
default via 192.168.99.1 proto static metric 100
""")
"""
GATEWAY_SAMPLE7 = ("""
GATEWAY_SAMPLE7 = """
default dev qg-31cd36 metric 1
""")
"""
IPv6_GATEWAY_SAMPLE1 = ("""
IPv6_GATEWAY_SAMPLE1 = """
default via 2001:470:9:1224:4508:b885:5fb:740b metric 100
2001:db8::/64 proto kernel scope link src 2001:470:9:1224:dfcc:aaff:feb9:76ce
""")
"""
IPv6_GATEWAY_SAMPLE2 = ("""
IPv6_GATEWAY_SAMPLE2 = """
default via 2001:470:9:1224:4508:b885:5fb:740b metric 100
""")
"""
IPv6_GATEWAY_SAMPLE3 = ("""
IPv6_GATEWAY_SAMPLE3 = """
2001:db8::/64 proto kernel scope link src 2001:470:9:1224:dfcc:aaff:feb9:76ce
""")
"""
IPv6_GATEWAY_SAMPLE4 = ("""
IPv6_GATEWAY_SAMPLE4 = """
default via fe80::dfcc:aaff:feb9:76ce
""")
"""
IPv6_GATEWAY_SAMPLE5 = ("""
IPv6_GATEWAY_SAMPLE5 = """
default via 2001:470:9:1224:4508:b885:5fb:740b metric 1024
""")
"""
DEVICE_ROUTE_SAMPLE = ("10.0.0.0/24 scope link src 10.0.0.2")
DEVICE_ROUTE_SAMPLE = "10.0.0.0/24 scope link src 10.0.0.2"
SUBNET_SAMPLE1 = ("10.0.0.0/24 dev qr-23380d11-d2 scope link src 10.0.0.1\n"
"10.0.0.0/24 dev tap1d7888a7-10 scope link src 10.0.0.2")
@@ -1136,7 +1136,7 @@ class IptablesManagerStateFulTestCase(IptablesManagerBaseTestCase):
def test_add_blank_rule(self):
iptables_args = {}
iptables_args.update(IPTABLES_ARG)
filter_rules = ('-A %(bn)s-test-filter\n' % iptables_args)
filter_rules = '-A %(bn)s-test-filter\n' % iptables_args
iptables_args['filter_rules'] = filter_rules
filter_dump_mod = FILTER_RESTORE_DUMP % iptables_args
@@ -1217,7 +1217,7 @@ class IptablesManagerStateFulTestCaseIPv6(IptablesManagerStateFulTestCase):
class IptablesManagerStateFulTestCaseCustomBinaryName(
IptablesManagerBaseTestCase):
use_ipv6 = False
bn = ("xbcdef" * 5)
bn = "xbcdef" * 5
def setUp(self):
super().setUp()
@@ -1354,7 +1354,7 @@ class IptablesManagerStateLessTestCase(base.BaseTestCase):
def setUp(self):
super().setUp()
cfg.CONF.set_override('comment_iptables_rules', False, 'AGENT')
self.iptables = (iptables_manager.IptablesManager(state_less=True))
self.iptables = iptables_manager.IptablesManager(state_less=True)
def test_nat_found(self):
self.assertIn('nat', self.iptables.ipv4)
@@ -1380,7 +1380,7 @@ class IptablesManagerNoNatTestCase(base.BaseTestCase):
def setUp(self):
super().setUp()
cfg.CONF.set_override('comment_iptables_rules', False, 'AGENT')
self.iptables = (iptables_manager.IptablesManager(nat=False))
self.iptables = iptables_manager.IptablesManager(nat=False)
def test_nat_found(self):
self.assertIn('nat', self.iptables.ipv4)
+1 -1
View File
@@ -227,7 +227,7 @@ class RequestTest(base.BaseTestCase):
self.assertEqual("application/json", result)
request = wsgi.Request.blank('/tests/123')
request.headers["Accept"] = ("application/json; q=0.3")
request.headers["Accept"] = "application/json; q=0.3"
result = request.best_match_content_type()
self.assertEqual("application/json", result)
+1 -1
View File
@@ -89,7 +89,7 @@ class TestAgentsDbBase(testlib_api.SqlTestCase):
for agent in agents[down_agents_count:(
down_but_version_considered + down_agents_count)]:
agent['heartbeat_timestamp'] -= datetime.timedelta(
seconds=(cfg.CONF.agent_down_time + 1))
seconds=cfg.CONF.agent_down_time + 1)
for agent in agents:
agent.create()
+3 -3
View File
@@ -410,7 +410,7 @@ class TestCli(base.BaseTestCase):
def _test_validate_head_files_helper(self, heads, contract_head='',
expand_head=''):
fake_config = self.configs[0]
head_files_not_exist = (contract_head == expand_head == '')
head_files_not_exist = contract_head == expand_head == ''
with mock.patch('alembic.script.ScriptDirectory.from_config') as fc,\
mock.patch('os.path.exists') as os_mock:
if head_files_not_exist:
@@ -706,11 +706,11 @@ class TestCli(base.BaseTestCase):
alembic_ag_api.render_python_code(expand.upgrade_ops),
matchers.MatchesRegex(_get_regex(ALEMBIC_EXPECTED_REGEX)))
expected_regex = ("""\
expected_regex = """\
### commands auto generated by Alembic - please adjust! ###
op.drop_constraint('user', 'uq_user_org'.*)
op.drop_column('user', 'organization_name')
### end Alembic commands ###""")
### end Alembic commands ###"""
self.assertThat(
alembic_ag_api.render_python_code(contract.upgrade_ops),
matchers.MatchesRegex(_get_regex(expected_regex)))
@@ -57,7 +57,7 @@ class L3NConntrackHelperTestCase(test_l3.L3BaseForIntTests,
'neutron.tests.unit.extensions.'
'test_l3_conntrack_helper.'
'TestL3ConntrackHelperServicePlugin')
plugin = ('neutron.tests.unit.extensions.test_l3.TestL3NatIntPlugin')
plugin = 'neutron.tests.unit.extensions.test_l3.TestL3NatIntPlugin'
ext_mgr = ExtendL3ConntrackHelperExtensionManager()
super().setUp(
ext_mgr=ext_mgr, service_plugins=svc_plugins, plugin=plugin)
@@ -54,6 +54,7 @@ class PortHardwareOffloadTypeExtensionTestCase(
if hardware_offload_type in constants.VALID_HWOL_TYPES:
port_args['hardware_offload_type'] = hardware_offload_type
with self.port(**port_args) as port:
print('port:', port)
for k, v in keys:
self.assertEqual(v, port['port'][k])
@@ -119,7 +119,7 @@ class TestL2PopulationRpcTestCase(test_plugin.Ml2PluginV2TestCase):
cast_patch = mock.patch(cast)
self.mock_cast = cast_patch.start()
uptime = ('neutron.plugins.ml2.drivers.l2pop.db.get_agent_uptime')
uptime = 'neutron.plugins.ml2.drivers.l2pop.db.get_agent_uptime'
uptime_patch = mock.patch(uptime, return_value=190)
uptime_patch.start()
# Extend network HA extension.
@@ -20,7 +20,7 @@ from neutron.privileged.agent.linux import utils as priv_utils
from neutron.tests import base
NETSTAT_NETNS_OUTPUT = ("""
NETSTAT_NETNS_OUTPUT = """
Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address Foreign Address State\
PID/Program name
@@ -37,21 +37,21 @@ Proto RefCnt Flags Type State I-Node PID/Program name\
Path
unix 2 [ ACC ] STREAM LISTENING 82039530 1353/python\
/tmp/rootwrap-VKSm8a/rootwrap.sock
""")
"""
NETSTAT_NO_NAMESPACE = ("""
NETSTAT_NO_NAMESPACE = """
Cannot open network namespace "qrouter-e6f206b2-4e8d-4597-a7e1-c3a20337e9c6":\
No such file or directory
""")
"""
NETSTAT_NO_LISTEN_PROCS = ("""
NETSTAT_NO_LISTEN_PROCS = """
Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address Foreign Address State\
PID/Program name
Active UNIX domain sockets (only servers)
Proto RefCnt Flags Type State I-Node PID/Program name\
Path
""")
"""
class FindListenPidsNamespaceTestCase(base.BaseTestCase):
@@ -2373,7 +2373,7 @@ class OVNL3ExtrarouteTests(test_l3_gw.ExtGwModeIntTestCase,
def setUp(self):
plugin = 'neutron.tests.unit.extensions.test_l3.TestOVNL3Plugin'
l3_plugin = ('neutron.services.ovn_l3.plugin.OVNL3RouterPlugin')
l3_plugin = 'neutron.services.ovn_l3.plugin.OVNL3RouterPlugin'
service_plugins = {'l3_plugin_name': l3_plugin}
config.register_opts()
cfg.CONF.set_default('max_routes', 3)