Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def interfaces(self, *args, **kwargs):
"""Public function to view interfaces"""
def filter_func(interface):
"""Filter based on args and kwargs"""
return all(interface.get(key) for key in args) \
and all(interface.get(key) == val for (key,
val) in kwargs.items())
all_interfaces = iter(NmPort(self.anm,
self.overlay_id, self.node_id,
interface_id) for interface_id in
self._interface_ids())
retval = [i for i in all_interfaces if filter_func(i)]
return retval
def dst_int(self):
"""Interface bound to destination node of edge
>>> anm = autonetkit.topos.house()
>>> edge = anm['phy'].edge("r1", "r2")
>>> edge.dst_int
eth0.r2
"""
dst_int_id = self._ports[self.dst_id]
return NmPort(self.anm, self.overlay_id,
self.dst_id, dst_int_id)
return NmEdge(self._anm, self._overlay_id, src, dst)
if isinstance(src, NmNode) and isinstance(dst, NmNode):
src_id = src.node_id
dst_id = dst.node_id
if self.is_multigraph():
if self._graph.has_edge(src_id, dst_id, key):
return NmEdge(self._anm, self._overlay_id, src, dst, key)
else:
if self._graph.has_edge(src_id, dst_id):
return NmEdge(self._anm, self._overlay_id, src, dst)
if isinstance(src, NmPort) and isinstance(dst, NmPort):
# further filter result by ports
src_id = src.node_id
dst_id = dst.node_id
src_int = src.interface_id
dst_int = dst.interface_id
# TODO: combine duplicated logic from above
#TODO: test with directed graph
if self.is_multigraph():
# search edges from src to dst
for src, iter_dst, iter_key in self._graph.edges(src_id, keys=True):
if iter_dst != dst_id:
continue # to a different node
ports = {k: v for k, v in edge.raw_interfaces.items()
if k in self._graph} # only if exists in this overlay
# TODO: debug log if skipping a binding?
data['_ports'] = ports
# this is the only case where copy across data
# but want to copy attributes for all cases
elif len(in_edge) == 2:
in_a, in_b = in_edge[0], in_edge[1]
if isinstance(in_a, NmNode) and isinstance(in_b, NmNode):
src = in_a.node_id
dst = in_b.node_id
elif isinstance(in_a, NmPort) and isinstance(in_b, NmPort):
src = in_a.node.node_id
dst = in_b.node.node_id
ports = {}
if src in self:
ports[src] = in_a.interface_id
if dst in self:
ports[dst] = in_b.interface_id
data['_ports'] = ports
elif isinstance(in_a, NmNode) and isinstance(in_b, NmPort):
src = in_a.node_id
dst = in_b.node.node_id
ports = {}
if dst in self:
ports[dst] = in_b.interface_id
data['_ports'] = ports
def interfaces(self):
"""
>>> anm = autonetkit.topos.house()
>>> edge = anm['phy'].edge("r1", "r2")
>>> list(edge.interfaces())
[eth0.r1, eth0.r2]
"""
# TODO: warn if interface doesn't exist on node
return [NmPort(self.anm, self.overlay_id,
node_id, interface_id) for (node_id,
interface_id) in self._ports.items()]
def add_loopback(self, *args, **kwargs):
'''Public function to add a loopback interface'''
interface_id = self._add_interface(category='loopback', *args,
**kwargs)
# TODO: want to add loopbacks to all overlays the node is in
self._sync_loopbacks(interface_id)
return NmPort(self.anm, self.overlay_id,
self.node_id, interface_id)