+ if worker is None:
+ total = mk_counter()
+ for t in c:
+ total["packets"] += t[self.stat_index]["packets"]
+ return total
+ else:
+ # +1 to skip main thread
+ return c[worker + 1][self.stat_index]
+
+ def get_lost(self, worker=None):
+ c = self.test.statistics.get_counter("/net/ipsec/sa/lost")
+ if worker is None:
+ total = 0
+ for t in c:
+ total += t[self.stat_index]
+ return total
+ else:
+ # +1 to skip main thread
+ return c[worker + 1][self.stat_index]
+
+
+class VppIpsecTunProtect(VppObject):
+ """
+ VPP IPSEC tunnel protection
+ """
+
+ def __init__(self, test, itf, sa_out, sas_in, nh=None):
+ self.test = test
+ self.itf = itf
+ self.sas_in = []
+ for sa in sas_in:
+ self.sas_in.append(sa.id)
+ self.sa_out = sa_out.id
+ self.nh = nh
+ if not self.nh:
+ self.nh = "0.0.0.0"
+
+ def update_vpp_config(self, sa_out, sas_in):
+ self.sas_in = []
+ for sa in sas_in:
+ self.sas_in.append(sa.id)
+ self.sa_out = sa_out.id
+ self.test.vapi.ipsec_tunnel_protect_update(
+ tunnel={
+ "sw_if_index": self.itf._sw_if_index,
+ "n_sa_in": len(self.sas_in),
+ "sa_out": self.sa_out,
+ "sa_in": self.sas_in,
+ "nh": self.nh,
+ }
+ )
+
+ def object_id(self):
+ return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
+
+ def add_vpp_config(self):
+ self.test.vapi.ipsec_tunnel_protect_update(
+ tunnel={
+ "sw_if_index": self.itf._sw_if_index,
+ "n_sa_in": len(self.sas_in),
+ "sa_out": self.sa_out,
+ "sa_in": self.sas_in,
+ "nh": self.nh,
+ }
+ )
+ self.test.registry.register(self, self.test.logger)
+
+ def remove_vpp_config(self):
+ self.test.vapi.ipsec_tunnel_protect_del(
+ sw_if_index=self.itf.sw_if_index, nh=self.nh
+ )
+
+ def query_vpp_config(self):
+ bs = self.test.vapi.ipsec_tunnel_protect_dump(sw_if_index=self.itf.sw_if_index)
+ for b in bs:
+ if b.tun.sw_if_index == self.itf.sw_if_index and self.nh == str(b.tun.nh):
+ return True
+ return False
+
+
+class VppIpsecInterface(VppInterface):
+ """
+ VPP IPSec interface
+ """
+
+ def __init__(self, test, mode=None, instance=0xFFFFFFFF):
+ super(VppIpsecInterface, self).__init__(test)
+
+ self.mode = mode
+ if not self.mode:
+ self.mode = VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_P2P
+ self.instance = instance
+
+ def add_vpp_config(self):
+ r = self.test.vapi.ipsec_itf_create(
+ itf={
+ "user_instance": self.instance,
+ "mode": self.mode,
+ }
+ )
+ self.set_sw_if_index(r.sw_if_index)
+ self.test.registry.register(self, self.test.logger)
+ ts = self.test.vapi.ipsec_itf_dump(sw_if_index=self._sw_if_index)
+ self.instance = ts[0].itf.user_instance
+ return self
+
+ def remove_vpp_config(self):
+ self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
+
+ def query_vpp_config(self):
+ ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xFFFFFFFF)
+ for t in ts:
+ if t.itf.sw_if_index == self._sw_if_index:
+ return True
+ return False
+
+ def __str__(self):
+ return self.object_id()
+
+ def object_id(self):
+ return "ipsec%d" % self.instance