All new accounts created on Gitlab now require administrator approval. If you invite any collaborators, please let Flux staff know so they can approve the accounts.

Commit 23d76639 authored by Peter V. Saveliev's avatar Peter V. Saveliev

sock_diag: `get_sock_stats()` method

Bug-Url: https://github.com/svinota/pyroute2/issues/291
parent 9a3a6da7
...@@ -78,33 +78,38 @@ class MarshalDiag(Marshal): ...@@ -78,33 +78,38 @@ class MarshalDiag(Marshal):
class DiagSocket(NetlinkSocket): class DiagSocket(NetlinkSocket):
'''
Usage::
from pyroute2 import DiagSocket
with DiagSocket() as ds:
ds.bind()
sstats = ds.get_sock_stats()
'''
def __init__(self, fileno=None): def __init__(self, fileno=None):
super(DiagSocket, self).__init__(NETLINK_SOCK_DIAG) super(DiagSocket, self).__init__(NETLINK_SOCK_DIAG)
self.marshal = MarshalDiag() self.marshal = MarshalDiag()
def test(self): def get_sock_stats(self,
family=AF_UNIX,
states=SS_ALL,
show=(UDIAG_SHOW_NAME |
UDIAG_SHOW_VFS |
UDIAG_SHOW_PEER |
UDIAG_SHOW_ICONS)):
''' '''
This function is here only as an example and for Get sockets statistics.
debugging purposes. It will be removed in the release.
Usage::
from pprint import pprint
from pyroute2.netlink.diag import DiagSocket
ds = DiagSocket()
ds.bind()
pprint(ds.test())
''' '''
req = unix_diag_req() if family == AF_UNIX:
req['sdiag_family'] = AF_UNIX req = unix_diag_req()
req['udiag_states'] = SS_ALL else:
req['udiag_show'] = UDIAG_SHOW_NAME |\ raise NotImplementedError()
UDIAG_SHOW_VFS |\ req['sdiag_family'] = family
UDIAG_SHOW_PEER |\ req['udiag_states'] = states
UDIAG_SHOW_ICONS req['udiag_show'] = show
return self.nlm_request(req, SOCK_DIAG_BY_FAMILY, return self.nlm_request(req, SOCK_DIAG_BY_FAMILY,
NLM_F_REQUEST | NLM_F_ROOT | NLM_F_MATCH) NLM_F_REQUEST | NLM_F_ROOT | NLM_F_MATCH)
from socket import AF_UNIX
from pyroute2 import DiagSocket from pyroute2 import DiagSocket
class TestDiag(object): class TestDiag(object):
def test_basic(self): def test_basic(self):
sstats_set = set()
pstats_set = set()
sstats = None
fd = None
with DiagSocket() as ds: with DiagSocket() as ds:
ds.bind() ds.bind()
assert len(ds.test()) > 0 sstats = ds.get_sock_stats(family=AF_UNIX)
for s in sstats:
sstats_set.add(s['udiag_ino'])
with open('/proc/net/unix') as fd:
for l in fd.readlines():
l = l.split()
if l[0][:2] != 'ff':
continue
pstats_set.add(int(l[6]))
assert sstats_set == pstats_set
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment