test/vpp_dhcp.py

   1 from vpp_object import VppObject
   2
   3
   4 class VppDHCPProxy(VppObject):
   5     def __init__(
   6         self,
   7         test,
   8         dhcp_server,
   9         dhcp_src_address,
  10         rx_vrf_id=0,
  11         server_vrf_id=0,
  12     ):
  13         self._test = test
  14         self._rx_vrf_id = rx_vrf_id
  15         self._server_vrf_id = server_vrf_id
  16         self._dhcp_server = dhcp_server
  17         self._dhcp_src_address = dhcp_src_address
  18
  19     def set_proxy(self, dhcp_server, dhcp_src_address, rx_vrf_id=0, server_vrf_id=0):
  20         if self.query_vpp_config():
  21             raise Exception("Vpp config present")
  22         self._rx_vrf_id = rx_vrf_id
  23         self._server_vrf_id = server_vrf_id
  24         self._dhcp_server = dhcp_server
  25         self._dhcp_src_address = dhcp_src_address
  26
  27     def add_vpp_config(self):
  28         self._test.vapi.dhcp_proxy_config(
  29             is_add=1,
  30             rx_vrf_id=self._rx_vrf_id,
  31             server_vrf_id=self._server_vrf_id,
  32             dhcp_server=self._dhcp_server,
  33             dhcp_src_address=self._dhcp_src_address,
  34         )
  35         self._test.registry.register(self, self._test.logger)
  36
  37     def remove_vpp_config(self):
  38         self._test.vapi.dhcp_proxy_config(
  39             rx_vrf_id=self._rx_vrf_id,
  40             server_vrf_id=self._server_vrf_id,
  41             dhcp_server=self._dhcp_server,
  42             dhcp_src_address=self._dhcp_src_address,
  43             is_add=0,
  44         )
  45
  46     def get_vpp_dump(self):
  47         dump = self._test.vapi.dhcp_proxy_dump()
  48         for entry in dump:
  49             if entry.rx_vrf_id == self._rx_vrf_id:
  50                 return entry
  51
  52     def query_vpp_config(self):
  53         dump = self.get_vpp_dump()
  54         return True if dump else False
  55
  56     def object_id(self):
  57         return "dhcp-proxy-%d" % self._rx_vrf_id
  58
  59
  60 class VppDHCPClient(VppObject):
  61     def __init__(
  62         self,
  63         test,
  64         sw_if_index,
  65         hostname,
  66         id=None,
  67         want_dhcp_event=False,
  68         set_broadcast_flag=True,
  69         dscp=None,
  70         pid=None,
  71     ):
  72         self._test = test
  73         self._sw_if_index = sw_if_index
  74         self._hostname = hostname
  75         self._id = id
  76         self._want_dhcp_event = want_dhcp_event
  77         self._set_broadcast_flag = set_broadcast_flag
  78         self._dscp = dscp
  79         self._pid = pid
  80
  81     def set_client(
  82         self,
  83         sw_if_index,
  84         hostname,
  85         id=None,
  86         want_dhcp_event=False,
  87         set_broadcast_flag=True,
  88         dscp=None,
  89         pid=None,
  90     ):
  91         if self.query_vpp_config():
  92             raise Exception("Vpp config present")
  93         self._sw_if_index = sw_if_index
  94         self._hostname = hostname
  95         self._id = id
  96         self._want_dhcp_event = want_dhcp_event
  97         self._set_broadcast_flag = set_broadcast_flag
  98         self._dscp = dscp
  99         self._pid = pid
 100
 101     def add_vpp_config(self):
 102         id = self._id.encode("ascii") if self._id else None
 103         client = {
 104             "sw_if_index": self._sw_if_index,
 105             "hostname": self._hostname,
 106             "id": id,
 107             "want_dhcp_event": self._want_dhcp_event,
 108             "set_broadcast_flag": self._set_broadcast_flag,
 109             "dscp": self._dscp,
 110             "pid": self._pid,
 111         }
 112         self._test.vapi.dhcp_client_config(is_add=1, client=client)
 113         self._test.registry.register(self, self._test.logger)
 114
 115     def remove_vpp_config(self):
 116         client = client = {"sw_if_index": self._sw_if_index, "hostname": self._hostname}
 117         self._test.vapi.dhcp_client_config(client=client, is_add=0)
 118
 119     def get_vpp_dump(self):
 120         dump = self._test.vapi.dhcp_client_dump()
 121         for entry in dump:
 122             if entry.client.sw_if_index == self._sw_if_index:
 123                 return entry
 124
 125     def query_vpp_config(self):
 126         dump = self.get_vpp_dump()
 127         return True if dump else False
 128
 129     def object_id(self):
 130         return "dhcp-client-%s/%d" % (self._hostname, self._sw_if_index)