obj, attr = self._resolve(attr)
return hasattr(obj, attr)
- def apply(self, attr, val):
- try:
- d = attrs[self._name + '.' + attr]
- except KeyError:
- raise KeyError('Unknown ' + self._name +
- ' attribute: ' + attr)
-
- if 'immutable' in d:
- raise ImmutableError(attr)
-
- if not self._hasattr(attr):
- raise KeyError('Invalid ' + self._name +
- ' attribute: ' + attr)
- self._setattr(attr, val)
-
class ObjIterator(object):
def __init__(self, cache, obj):
self._cache = cache
capi.nl_addr_set_family(self._nl_addr, int(value))
-# global dictionay for all object attributes
-#
-# attrs[type][keyword] : value
-#
# keyword:
# type = { int | str }
# immutable = { True | False }
# fmt = func (formatting function)
-#
-attrs = {}
-
-def add_attr(name, **kwds):
- attrs[name] = {}
- for k in kwds:
- attrs[name][k] = kwds[k]
+# title = string
-def nlattr(name, **kwds):
+def nlattr(**kwds):
"""netlink object attribute decorator
decorator used to mark mutable and immutable properties
of netlink objects. All properties marked as such are
regarded to be accessable.
- @netlink.nlattr('my_type.my_attr', type=int)
@property
+ @netlink.nlattr(type=int)
def my_attr(self):
return self._my_attr
"""
- attrs[name] = {}
- for k in kwds:
- attrs[name][k] = kwds[k]
-
def wrap_fn(func):
+ func.formatinfo = kwds
return func
-
return wrap_fn
-
def _new_instance(obj):
return Address(obj)
- @netlink.nlattr('address.ifindex', type=int, immutable=True,
- fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def ifindex(self):
"""interface index"""
return capi.rtnl_addr_get_ifindex(self._rtnl_addr)
self.link = link
- @netlink.nlattr('address.link', type=str, fmt=util.string)
@property
+ @netlink.nlattr(type=str, fmt=util.string)
def link(self):
link = capi.rtnl_addr_get_link(self._rtnl_addr)
if not link:
if capi.rtnl_addr_get_ifindex(self._orig) == 0:
capi.rtnl_addr_set_ifindex(self._orig, value.ifindex)
- @netlink.nlattr('address.label', type=str, fmt=util.string)
@property
+ @netlink.nlattr(type=str, fmt=util.string)
def label(self):
"""address label"""
return capi.rtnl_addr_get_label(self._rtnl_addr)
def label(self, value):
capi.rtnl_addr_set_label(self._rtnl_addr, value)
- @netlink.nlattr('address.flags', type=str, fmt=util.string)
@property
+ @netlink.nlattr(type=str, fmt=util.string)
def flags(self):
"""Flags
else:
self._set_flag(value)
- @netlink.nlattr('address.family', type=int, immutable=True,
- fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def family(self):
"""Address family"""
fam = capi.rtnl_addr_get_family(self._rtnl_addr)
capi.rtnl_addr_set_family(self._rtnl_addr, int(value))
- @netlink.nlattr('address.scope', type=int, fmt=util.num)
@property
+ @netlink.nlattr(type=int, fmt=util.num)
def scope(self):
"""Address scope"""
scope = capi.rtnl_addr_get_scope(self._rtnl_addr)
value = capi.rtnl_str2scope(value)
capi.rtnl_addr_set_scope(self._rtnl_addr, value)
- @netlink.nlattr('address.local', type=str, immutable=True,
- fmt=util.addr)
@property
+ @netlink.nlattr(type=str, immutable=True, fmt=util.addr)
def local(self):
"""Local address"""
a = capi.rtnl_addr_get_local(self._rtnl_addr)
if capi.rtnl_addr_get_local(self._orig) is None:
capi.rtnl_addr_set_local(self._orig, a._nl_addr)
- @netlink.nlattr('address.peer', type=str, fmt=util.addr)
@property
+ @netlink.nlattr(type=str, fmt=util.addr)
def peer(self):
"""Peer address"""
a = capi.rtnl_addr_get_peer(self._rtnl_addr)
a = netlink.AbstractAddress(value)
capi.rtnl_addr_set_peer(self._rtnl_addr, a._nl_addr)
- @netlink.nlattr('address.broadcast', type=str, fmt=util.addr)
@property
+ @netlink.nlattr(type=str, fmt=util.addr)
def broadcast(self):
"""Broadcast address"""
a = capi.rtnl_addr_get_broadcast(self._rtnl_addr)
a = netlink.AbstractAddress(value)
capi.rtnl_addr_set_broadcast(self._rtnl_addr, a._nl_addr)
- @netlink.nlattr('address.multicast', type=str, fmt=util.addr)
@property
+ @netlink.nlattr(type=str, fmt=util.addr)
def multicast(self):
"""multicast address"""
a = capi.rtnl_addr_get_multicast(self._rtnl_addr)
capi.rtnl_addr_set_multicast(self._rtnl_addr, a._nl_addr)
- @netlink.nlattr('address.anycast', type=str, fmt=util.addr)
@property
+ @netlink.nlattr(type=str, fmt=util.addr)
def anycast(self):
"""anycast address"""
a = capi.rtnl_addr_get_anycast(self._rtnl_addr)
a = netlink.AbstractAddress(value)
capi.rtnl_addr_set_anycast(self._rtnl_addr, a._nl_addr)
- @netlink.nlattr('address.valid_lifetime', type=int, immutable=True,
- fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def valid_lifetime(self):
"""Valid lifetime"""
msecs = capi.rtnl_addr_get_valid_lifetime(self._rtnl_addr)
def valid_lifetime(self, value):
capi.rtnl_addr_set_valid_lifetime(self._rtnl_addr, int(value))
- @netlink.nlattr('address.preferred_lifetime', type=int,
- immutable=True, fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def preferred_lifetime(self):
"""Preferred lifetime"""
msecs = capi.rtnl_addr_get_preferred_lifetime(self._rtnl_addr)
def preferred_lifetime(self, value):
capi.rtnl_addr_set_preferred_lifetime(self._rtnl_addr, int(value))
- @netlink.nlattr('address.create_time', type=int, immutable=True,
- fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def create_time(self):
"""Creation time"""
hsec = capi.rtnl_addr_get_create_time(self._rtnl_addr)
return datetime.timedelta(milliseconds=10*hsec)
- @netlink.nlattr('address.last_update', type=int, immutable=True,
- fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def last_update(self):
"""Last update"""
hsec = capi.rtnl_addr_get_last_update_time(self._rtnl_addr)
return Link(obj)
- @netlink.nlattr('link.ifindex', type=int, immutable=True, fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def ifindex(self):
"""interface index"""
return capi.rtnl_link_get_ifindex(self._rtnl_link)
if capi.rtnl_link_get_ifindex(self._orig) == 0:
capi.rtnl_link_set_ifindex(self._orig, int(value))
- @netlink.nlattr('link.name', type=str, fmt=util.bold)
@property
+ @netlink.nlattr(type=str, fmt=util.bold)
def name(self):
"""Name of link"""
return capi.rtnl_link_get_name(self._rtnl_link)
if capi.rtnl_link_get_name(self._orig) is None:
capi.rtnl_link_set_name(self._orig, value)
- @netlink.nlattr('link.flags', type=str, fmt=util.string)
@property
+ @netlink.nlattr(type=str, fmt=util.string)
def flags(self):
"""Flags
Setting this property will *Not* reset flags to value you supply in
else:
self._set_flag(value)
- @netlink.nlattr('link.mtu', type=int, fmt=util.num)
@property
+ @netlink.nlattr(type=int, fmt=util.num)
def mtu(self):
"""Maximum Transmission Unit"""
return capi.rtnl_link_get_mtu(self._rtnl_link)
def mtu(self, value):
capi.rtnl_link_set_mtu(self._rtnl_link, int(value))
- @netlink.nlattr('link.family', type=int, immutable=True, fmt=util.num)
@property
+ @netlink.nlattr(type=int, immutable=True, fmt=util.num)
def family(self):
"""Address family"""
return capi.rtnl_link_get_family(self._rtnl_link)
def family(self, value):
capi.rtnl_link_set_family(self._rtnl_link, value)
- @netlink.nlattr('link.address', type=str, fmt=util.addr)
@property
+ @netlink.nlattr(type=str, fmt=util.addr)
def address(self):
"""Hardware address (MAC address)"""
a = capi.rtnl_link_get_addr(self._rtnl_link)
def address(self, value):
capi.rtnl_link_set_addr(self._rtnl_link, value._addr)
- @netlink.nlattr('link.broadcast', type=str, fmt=util.addr)
@property
+ @netlink.nlattr(type=str, fmt=util.addr)
def broadcast(self):
"""Hardware broadcast address"""
a = capi.rtnl_link_get_broadcast(self._rtnl_link)
def broadcast(self, value):
capi.rtnl_link_set_broadcast(self._rtnl_link, value._addr)
- @netlink.nlattr('link.qdisc', type=str, immutable=True, fmt=util.string)
@property
+ @netlink.nlattr(type=str, immutable=True, fmt=util.string)
def qdisc(self):
"""Name of qdisc (cannot be changed)"""
return capi.rtnl_link_get_qdisc(self._rtnl_link)
def qdisc(self, value):
capi.rtnl_link_set_qdisc(self._rtnl_link, value)
- @netlink.nlattr('link.txqlen', type=int, fmt=util.num)
@property
+ @netlink.nlattr(type=int, fmt=util.num)
def txqlen(self):
"""Length of transmit queue"""
return capi.rtnl_link_get_txqlen(self._rtnl_link)
def txqlen(self, value):
capi.rtnl_link_set_txqlen(self._rtnl_link, int(value))
- @netlink.nlattr('link.weight', type=str, fmt=util.string)
@property
+ @netlink.nlattr(type=str, fmt=util.string)
def weight(self):
"""Weight"""
v = capi.rtnl_link_get_weight(self._rtnl_link)
v = int(value)
capi.rtnl_link_set_weight(self._rtnl_link, v)
- @netlink.nlattr('link.arptype', type=str, immutable=True, fmt=util.string)
@property
+ @netlink.nlattr(type=str, immutable=True, fmt=util.string)
def arptype(self):
"""Type of link (cannot be changed)"""
type_ = capi.rtnl_link_get_arptype(self._rtnl_link)
i = core_capi.nl_str2llproto(value)
capi.rtnl_link_set_arptype(self._rtnl_link, i)
- @netlink.nlattr('link.operstate', type=str, immutable=True,
- fmt=util.string, title='state')
@property
+ @netlink.nlattr(type=str, immutable=True, fmt=util.string, title='state')
def operstate(self):
"""Operational status"""
operstate = capi.rtnl_link_get_operstate(self._rtnl_link)
i = capi.rtnl_link_str2operstate(value)
capi.rtnl_link_set_operstate(self._rtnl_link, i)
- @netlink.nlattr('link.mode', type=str, immutable=True, fmt=util.string)
@property
+ @netlink.nlattr(type=str, immutable=True, fmt=util.string)
def mode(self):
"""Link mode"""
mode = capi.rtnl_link_get_linkmode(self._rtnl_link)
i = capi.rtnl_link_str2mode(value)
capi.rtnl_link_set_linkmode(self._rtnl_link, i)
- @netlink.nlattr('link.alias', type=str, fmt=util.string)
@property
+ @netlink.nlattr(type=str, fmt=util.string)
def alias(self):
"""Interface alias (SNMP)"""
return capi.rtnl_link_get_ifalias(self._rtnl_link)
def alias(self, value):
capi.rtnl_link_set_ifalias(self._rtnl_link, value)
- @netlink.nlattr('link.type', type=str, fmt=util.string)
@property
+ @netlink.nlattr(type=str, fmt=util.string)
def type(self):
"""Link type"""
return capi.rtnl_link_get_type(self._rtnl_link)
return capi.rtnl_link_inet_set_conf(self._link._rtnl_link,
_resolve(id), int(value))
- @netlink.nlattr('link.inet.forwarding', type=bool, fmt=util.boolean)
@property
+ @netlink.nlattr(type=bool, fmt=util.boolean)
def forwarding(self):
return bool(self.get_conf(DEVCONF_FORWARDING))
def forwarding(self, value):
self.set_conf(DEVCONF_FORWARDING, int(value))
- @netlink.nlattr('link.inet.mc_forwarding', type=bool, fmt=util.boolean)
+
@property
+ @netlink.nlattr(type=bool, fmt=util.boolean)
def mc_forwarding(self):
return bool(self.get_conf(DEVCONF_MC_FORWARDING))
def mc_forwarding(self, value):
self.set_conf(DEVCONF_MC_FORWARDING, int(value))
- @netlink.nlattr('link.inet.proxy_arp', type=bool, fmt=util.boolean)
+
@property
+ @netlink.nlattr(type=bool, fmt=util.boolean)
def proxy_arp(self):
return bool(self.get_conf(DEVCONF_PROXY_ARP))
def proxy_arp(self, value):
self.set_conf(DEVCONF_PROXY_ARP, int(value))
- @netlink.nlattr('link.inet.accept_redirects', type=bool, fmt=util.boolean)
@property
+ @netlink.nlattr(type=bool, fmt=util.boolean)
def accept_redirects(self):
return bool(self.get_conf(DEVCONF_ACCEPT_REDIRECTS))
def accept_redirects(self, value):
self.set_conf(DEVCONF_ACCEPT_REDIRECTS, int(value))
- @netlink.nlattr('link.inet.secure_redirects', type=bool, fmt=util.boolean)
@property
+ @netlink.nlattr(type=bool, fmt=util.boolean)
def secure_redirects(self):
return bool(self.get_conf(DEVCONF_SECURE_REDIRECTS))
def secure_redirects(self, value):
self.set_conf(DEVCONF_SECURE_REDIRECTS, int(value))
- @netlink.nlattr('link.inet.send_redirects', type=bool, fmt=util.boolean)
@property
+ @netlink.nlattr(type=bool, fmt=util.boolean)
def send_redirects(self):
return bool(self.get_conf(DEVCONF_SEND_REDIRECTS))
def send_redirects(self, value):
self.set_conf(DEVCONF_SEND_REDIRECTS, int(value))
- @netlink.nlattr('link.inet.shared_media', type=bool, fmt=util.boolean)
@property
+ @netlink.nlattr(type=bool, fmt=util.boolean)
def shared_media(self):
return bool(self.get_conf(DEVCONF_SHARED_MEDIA))
def __init__(self, link):
self._link = link
- @netlink.nlattr('link.vlan.id', type=int)
@property
+ @netlink.nlattr(type=int)
def id(self):
"""vlan identifier"""
return capi.rtnl_link_vlan_get_id(self._link)
def id(self, value):
capi.rtnl_link_vlan_set_id(self._link, int(value))
- @netlink.nlattr('link.vlan.flags', type=str)
@property
+ @netlink.nlattr(type=str)
def flags(self):
""" VLAN flags
Setting this property will *Not* reset flags to value you supply in
def __init__(self, qdisc):
self._qdisc = qdisc
- @netlink.nlattr('qdisc.htb.default_class', type=int)
@property
+ @netlink.nlattr(type=int)
def default_class(self):
return tc.Handle(capi.rtnl_htb_get_defcls(self._qdisc._rtnl_qdisc))
def default_class(self, value):
capi.rtnl_htb_set_defcls(self._qdisc._rtnl_qdisc, int(value))
- @netlink.nlattr('qdisc.htb.r2q', type=int)
@property
+ @netlink.nlattr('r2q', type=int)
def r2q(self):
return capi.rtnl_htb_get_rate2quantum(self._qdisc._rtnl_qdisc)
def __init__(self, cl):
self._class = cl
- @netlink.nlattr('class.htb.rate', type=str)
@property
+ @netlink.nlattr(type=str)
def rate(self):
rate = capi.rtnl_htb_get_rate(self._class._rtnl_class)
return util.Rate(rate)
def rate(self, value):
capi.rtnl_htb_set_rate(self._class._rtnl_class, int(value))
- @netlink.nlattr('class.htb.ceil', type=str)
@property
+ @netlink.nlattr(type=str)
def ceil(self):
ceil = capi.rtnl_htb_get_ceil(self._class._rtnl_class)
return util.Rate(ceil)
def ceil(self, value):
capi.rtnl_htb_set_ceil(self._class._rtnl_class, int(value))
- @netlink.nlattr('class.htb.burst', type=str)
@property
+ @netlink.nlattr(type=str)
def burst(self):
burst = capi.rtnl_htb_get_rbuffer(self._class._rtnl_class)
return util.Size(burst)
def burst(self, value):
capi.rtnl_htb_set_rbuffer(self._class._rtnl_class, int(value))
- @netlink.nlattr('class.htb.ceil_burst', type=str)
@property
+ @netlink.nlattr(type=str)
def ceil_burst(self):
burst = capi.rtnl_htb_get_cbuffer(self._class._rtnl_class)
return util.Size(burst)
def ceil_burst(self, value):
capi.rtnl_htb_set_cbuffer(self._class._rtnl_class, int(value))
- @netlink.nlattr('class.htb.prio', type=int)
@property
+ @netlink.nlattr(type=int)
def prio(self):
return capi.rtnl_htb_get_prio(self._class._rtnl_class)
def prio(self, value):
capi.rtnl_htb_set_prio(self._class._rtnl_class, int(value))
- @netlink.nlattr('class.htb.quantum', type=int)
@property
+ @netlink.nlattr(type=int)
def quantum(self):
return capi.rtnl_htb_get_quantum(self._class._rtnl_class)
def quantum(self, value):
capi.rtnl_htb_set_quantum(self._class._rtnl_class, int(value))
- @netlink.nlattr('class.htb.level', type=int)
@property
+ @netlink.nlattr(type=int)
def level(self):
return capi.rtnl_htb_get_level(self._class._rtnl_class)
capi.rtnl_tc_set_linktype(self._rtnl_tc, int(value))
@property
+ @netlink.nlattr(fmt=util.handle)
def handle(self):
return Handle(capi.rtnl_tc_get_handle(self._rtnl_tc))
capi.rtnl_tc_set_handle(self._rtnl_tc, int(value))
@property
+ @netlink.nlattr(fmt=util.handle)
def parent(self):
return Handle(capi.rtnl_tc_get_parent(self._rtnl_tc))
capi.rtnl_tc_set_parent(self._rtnl_tc, int(value))
@property
+ @netlink.nlattr(fmt=util.bold)
def kind(self):
return capi.rtnl_tc_get_kind(self._rtnl_tc)
self._rtnl_qdisc = self._obj2type(self._nl_object)
self._rtnl_tc = capi.obj2tc(self._nl_object)
- netlink.add_attr('qdisc.handle', fmt=util.handle)
- netlink.add_attr('qdisc.parent', fmt=util.handle)
- netlink.add_attr('qdisc.kind', fmt=util.bold)
-
if self.kind:
self._tc_module_lookup()
self._rtnl_class = self._obj2type(self._nl_object)
self._rtnl_tc = capi.obj2tc(self._nl_object)
- netlink.add_attr('class.handle', fmt=util.handle)
- netlink.add_attr('class.parent', fmt=util.handle)
- netlink.add_attr('class.kind', fmt=util.bold)
-
if self.kind:
self._tc_module_lookup()
self._rtnl_cls = self._obj2type(self._nl_object)
self._rtnl_tc = capi.obj2tc(self._nl_object)
- netlink.add_attr('cls.handle', fmt=util.handle)
- netlink.add_attr('cls.parent', fmt=util.handle)
- netlink.add_attr('cls.kind', fmt=util.bold)
-
@classmethod
def from_capi(cls, obj):
return cls(capi.cls2obj(obj))
self._indent = indent
def _nlattr(self, key):
- value = getattr(self._obj, key)
- title_ = None
+ value = getattr(self._obj.__class__, key)
+ if not isinstance(value, property):
+ raise ValueError('Invalid formatting string {0}'.format(key))
- if isinstance(value, types.MethodType):
- value = value()
+ d = getattr(value.fget, 'formatinfo', dict())
- try:
- d = netlink.attrs[self._obj._name + '.' + key]
+ # value = value.fget() is exactly the same
+ value = getattr(self._obj, key)
- if 'fmt' in d:
- value = d['fmt'](value)
+ if 'fmt' in d:
+ value = d['fmt'](value)
- if 'title' in d:
- title_ = d['title']
- except KeyError:
- pass
- except AttributeError:
- pass
+ title_ = d.get('title', None)
return title_, str(value)