"""NAT utilities library."""
+from math import log2, modf
from pprint import pformat
from enum import IntEnum
: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):
]
PapiSocketExecutor.dump_and_log(node, cmds)
+ @staticmethod
+ def compute_max_translations_per_thread(sessions, threads):
+ """Compute value of max_translations_per_thread NAT44 parameter based on
+ total number of worker threads.
+
+ :param sessions: Required number of NAT44 sessions.
+ :param threads: Number of worker threads.
+ :type sessions: int
+ :type threads: int
+ :returns: Value of max_translations_per_thread NAT44 parameter.
+ :rtype: int
+ """
+ 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 NAT
+ # DET44 means deterministic mode of NAT44
@staticmethod
def enable_det44_plugin(node, inside_vrf=0, outside_vrf=0):
"""Enable DET44 plugin.
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args_in).get_reply(err_msg)
+ @staticmethod
+ def get_det44_mapping(node):
+ """Get DET44 mapping data.
+
+ :param node: DUT node.
+ :type node: dict
+ :returns: Dictionary of DET44 mapping data.
+ :rtype: dict
+ """
+ cmd = u"det44_map_dump"
+ err_msg = f"Failed to get DET44 mapping data on the host " \
+ f"{node[u'host']}!"
+ args_in = dict()
+ with PapiSocketExecutor(node) as papi_exec:
+ details = papi_exec.add(cmd, **args_in).get_reply(err_msg)
+
+ return details
+
+ @staticmethod
+ 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
def show_det44(node):
"""Show DET44 data.