"""NAT utilities library."""
+from math import log2, modf
from pprint import pformat
from enum import IntEnum
flag=u"NAT_IS_NONE"):
"""Set NAT44 address range.
+ The return value is a callable (zero argument Python function)
+ which can be used to reset NAT state, so repeated trial measurements
+ hit the same slow path.
+
:param node: DUT node.
:param start_ip: IP range start.
:param end_ip: IP range end.
:type end_ip: str
:type vrf_id: int
:type flag: str
+ :returns: Resetter of the NAT state.
+ :rtype: Callable[[], None]
"""
cmd = u"nat44_add_del_address_range"
err_msg = f"Failed to set NAT44 address range on host {node[u'host']}"
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args_in).get_reply(err_msg)
+ # A closure, accessing the variables above.
+ def resetter():
+ """Delete and re-add the NAT range setting."""
+ with PapiSocketExecutor(node) as papi_exec:
+ args_in[u"is_add"] = False
+ papi_exec.add(cmd, **args_in)
+ args_in[u"is_add"] = True
+ papi_exec.add(cmd, **args_in)
+ papi_exec.get_replies(err_msg)
+
+ return resetter
+
@staticmethod
def show_nat_config(node):
"""Show the NAT configuration.
:param node: Topology node.
:type node: dict
+ :returns: NAT44 summary data.
+ :rtype: str
"""
- PapiSocketExecutor.run_cli_cmd(node, u"show nat44 summary")
+ return PapiSocketExecutor.run_cli_cmd(node, u"show nat44 summary")
@staticmethod
def show_nat_base_data(node):
:returns: Value of max_translations_per_thread NAT44 parameter.
:rtype: int
"""
- from math import log2, modf
rest, mult = modf(log2(sessions/(10*threads)))
return 2 ** (int(mult) + (1 if rest else 0)) * 10
+ @staticmethod
+ def get_nat44_sessions_number(node, proto):
+ """Get number of established NAT44 sessions from actual NAT44 mapping
+ data.
+
+ :param node: DUT node.
+ :param proto: Required protocol - TCP/UDP/ICMP.
+ :type node: dict
+ :type proto: str
+ :returns: Number of established NAT44 sessions.
+ :rtype: int
+ :raises ValueError: If not supported protocol.
+ """
+ nat44_data = dict()
+ if proto in [u"UDP", u"TCP", u"ICMP"]:
+ for line in NATUtil.show_nat44_summary(node).splitlines():
+ sum_k, sum_v = line.split(u":") if u":" in line \
+ else (line, None)
+ nat44_data[sum_k] = sum_v.strip() if isinstance(sum_v, str) \
+ else sum_v
+ else:
+ raise ValueError(f"Unsupported protocol: {proto}!")
+ return nat44_data.get(f"total {proto.lower()} sessions", 0)
+
# DET44 PAPI calls
# DET44 means deterministic mode of NAT44
@staticmethod
def set_det44_mapping(node, ip_in, subnet_in, ip_out, subnet_out):
"""Set DET44 mapping.
+ The return value is a callable (zero argument Python function)
+ which can be used to reset NAT state, so repeated trial measurements
+ hit the same slow path.
+
:param node: DUT node.
:param ip_in: Inside IP.
:param subnet_in: Inside IP subnet.
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args_in).get_reply(err_msg)
+ # A closure, accessing the variables above.
+ def resetter():
+ """Delete and re-add the deterministic NAT mapping."""
+ with PapiSocketExecutor(node) as papi_exec:
+ args_in[u"is_add"] = False
+ papi_exec.add(cmd, **args_in)
+ args_in[u"is_add"] = True
+ papi_exec.add(cmd, **args_in)
+ papi_exec.get_replies(err_msg)
+
+ return resetter
+
@staticmethod
def get_det44_mapping(node):
"""Get DET44 mapping data.
def get_det44_sessions_number(node):
"""Get number of established DET44 sessions from actual DET44 mapping
data.
-
:param node: DUT node.
:type node: dict
:returns: Number of established DET44 sessions.
:rtype: int
"""
det44_data = NATUtil.get_det44_mapping(node)
-
return det44_data.get(u"ses_num", 0)
@staticmethod