Lines Matching refs:self
51 def __init__(self, iface_alias: str, iface_name: str): argument
52 self.name = iface_name
53 self.alias = iface_alias
54 self.vnet_name = ""
55 self.jailed = False
56 self.addr_map: Dict[str, Dict] = {"inet6": {}, "inet": {}}
57 self.prefixes4: List[List[str]] = []
58 self.prefixes6: List[List[str]] = []
60 self.iftype = self.IFT_LOOP
62 self.iftype = self.IFT_ETHER
65 def ifindex(self): argument
66 return socket.if_nametoindex(self.name)
69 def first_ipv6(self): argument
70 d = self.addr_map["inet6"]
74 def first_ipv4(self): argument
75 d = self.addr_map["inet"]
78 def set_vnet(self, vnet_name: str): argument
79 self.vnet_name = vnet_name
81 def set_jailed(self, jailed: bool): argument
82 self.jailed = jailed
85 self, argument
89 if self.vnet_name and not self.jailed:
90 cmd = "jexec {} {}".format(self.vnet_name, cmd)
110 def setup_addr(self, _addr: str): argument
114 cmd = "/sbin/ifconfig {} {} {}".format(self.name, family, addr)
117 if self.addr_map[family]:
118 cmd = "/sbin/ifconfig {} alias {}".format(self.name, addr)
120 cmd = "/sbin/ifconfig {} {} {}".format(self.name, family, addr)
121 self.run_cmd(cmd)
122 self.addr_map[family][str(addr.ip)] = addr
124 def delete_addr(self, _addr: str): argument
128 cmd = "/sbin/ifconfig {} inet6 {} delete".format(self.name, addr)
131 cmd = "/sbin/ifconfig {} -alias {}".format(self.name, addr)
132 self.run_cmd(cmd)
133 del self.addr_map[family][str(addr)]
135 def turn_up(self): argument
136 cmd = "/sbin/ifconfig {} up".format(self.name)
137 self.run_cmd(cmd)
139 def enable_ipv6(self): argument
140 cmd = "/usr/sbin/ndp -i {} -disabled".format(self.name)
141 self.run_cmd(cmd)
143 def has_tentative(self) -> bool: argument
145 cmd = "/sbin/ifconfig {} inet6".format(self.name)
146 out = self.run_cmd(cmd, verbose=False)
157 def __init__(self): argument
158 self.file_name = self.INTERFACES_FNAME
160 def _register_iface(self, iface_name: str): argument
161 with open(self.file_name, "a") as f:
164 def _list_ifaces(self) -> List[str]: argument
167 with open(self.file_name, "r") as f:
174 def create_iface(self, alias_name: str, iface_name: str) -> List[VnetInterface]: argument
177 if not self.is_autodeleted(iface.name):
178 self._register_iface(iface.name)
186 def cleanup_vnet_interfaces(self, vnet_name: str) -> List[str]: argument
192 if not self.is_autodeleted(iface_name):
193 if iface_name not in self._list_ifaces():
200 def cleanup(self): argument
202 os.unlink(self.INTERFACES_FNAME)
209 self, vnet_alias: str, vnet_name: str, jid: int, ifaces: List[VnetInterface] argument
211 self.name = vnet_name
212 self.alias = vnet_alias # reference in the test topology
213 self.jid = jid
214 self.ifaces = ifaces
215 self.iface_alias_map = {} # iface.alias: iface
216 self.iface_map = {} # iface.name: iface
220 self.iface_alias_map[iface.alias] = iface
221 self.iface_map[iface.name] = iface
223 setattr(self, iface.alias, iface)
224 self.need_dad = False # Disable duplicate address detection by default
225 self.attached = False
226 self.pipe = None
227 self.subprocess = None
229 def run_vnet_cmd(self, cmd): argument
230 if not self.attached:
231 cmd = "jexec {} {}".format(self.name, cmd)
234 def disable_dad(self): argument
235 self.run_vnet_cmd("/sbin/sysctl net.inet6.ip6.dad_count=0")
237 def set_pipe(self, pipe): argument
238 self.pipe = pipe
240 def set_subprocess(self, p): argument
241 self.subprocess = p
249 def attach(self): argument
250 self.attach_jid(self.jid)
251 self.attached = True
257 def __init__(self, topology_id: str): argument
258 self.topology_id = topology_id
259 self.file_name = self.JAILS_FNAME
260 self._vnets: List[str] = []
262 def _register_vnet(self, vnet_name: str): argument
263 self._vnets.append(vnet_name)
264 with open(self.file_name, "a") as f:
282 def create_vnet(self, vnet_alias: str, ifaces: List[VnetInterface]): argument
283 vnet_name = "pytest:{}".format(convert_test_name(self.topology_id))
284 if self._vnets:
286 vnet_name = "{}_{}".format(vnet_name, len(self._vnets) + 1)
298 self._register_vnet(vnet_name)
303 not_found = self._wait_interfaces(vnet_name, [i.name for i in ifaces])
310 def cleanup(self): argument
313 with open(self.file_name) as f:
318 os.unlink(self.JAILS_FNAME)
338 def _require_default_modules(self): argument
340 self.require_module("if_epair")
342 def _get_vnet_handler(self, vnet_alias: str): argument
344 return getattr(self, handler_name, None)
346 def _setup_vnet(self, vnet: VnetInstance, obj_map: Dict, pipe): argument
381 handler = self._get_vnet_handler(vnet.alias)
386 self.drop_privileges()
389 def _get_topo_ifmap(self, topo: Dict): argument
406 def setup_topology(self, topo: Dict, topology_id: str): argument
410 iface_map = self._get_topo_ifmap(topo)
423 setattr(self, obj_name, vnet)
428 handler = self._get_vnet_handler(vnet.alias)
456 def setup_method(self, _method): argument
459 self._require_default_modules()
462 topology_id = get_topology_id(self.test_id)
463 topology = self.TOPOLOGY
465 obj_map = self.setup_topology(topology, topology_id)
468 if self._get_vnet_handler(vnet_alias):
472 target=self._setup_vnet,
488 self.vnet = main_vnet
489 self._setup_vnet(main_vnet, obj_map, None)
491 self.iface_map = obj_map.iface_map
492 self.vnet_map = obj_map.vnet_map
493 self.drop_privileges()
495 def cleanup(self, test_id: str): argument
497 topology_id = get_topology_id(self.test_id)
504 def wait_object(self, pipe, timeout=5): argument
509 def wait_objects_any(self, pipe_list, timeout=5): argument
515 def send_object(self, pipe, obj): argument
518 def wait(self): argument
523 def curvnet(self): argument
532 def _setup_default_topology(self): argument
536 "if1": {"type": self.IFTYPE, "prefixes4": [], "prefixes6": []},
539 for prefix in self.IPV6_PREFIXES:
541 for prefix in self.IPV4_PREFIXES:
545 def setup_method(self, method): argument
546 if not getattr(self, "TOPOLOGY", None):
547 self.TOPOLOGY = self._setup_default_topology()
549 names = self.TOPOLOGY.keys()