ipng/vpp-snmp-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Format with black

Browse Source
This commit is contained in:
Pim van Pelt
2023-01-08 13:05:42 +01:00
parent 72e9cf3503
commit 5e11539b44
8 changed files with 540 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
agentx/network.py
View File

@ -17,15 +17,17 @@ class NullHandler(logging.Handler):
def emit(self, record):
pass
logger = logging.getLogger('agentx.network')
logger = logging.getLogger("agentx.network")
logger.addHandler(NullHandler())
class NetworkError(Exception):
pass
class Network():
def __init__(self, server_address = '/var/agentx/master'):
class Network:
def __init__(self, server_address="/var/agentx/master"):
self.session_id = 0
self.transaction_id = 0
@ -35,8 +37,7 @@ class Network():
self.data_idx = []
self._connected = False
self._server_address = server_address
self._timeout = 0.1 # Seconds
self._timeout = 0.1 # Seconds
def connect(self):
if self._connected:
@ -44,12 +45,12 @@ class Network():
try:
logger.info("Connecting to %s" % self._server_address)
if self._server_address.startswith('/'):
if self._server_address.startswith("/"):
self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.socket.connect(self._server_address)
else:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
host, port=self._server_address.split(':')
host, port = self._server_address.split(":")
self.socket.connect((host, int(port)))
self.socket.settimeout(self._timeout)
@ -76,7 +77,9 @@ class Network():
self.data = newdata.copy()
del self.data_idx
self.data_idx = sorted(self.data.keys(), key=lambda k: tuple(int(part) for part in k.split('.')))
self.data_idx = sorted(
self.data.keys(), key=lambda k: tuple(int(part) for part in k.split("."))
)
def new_pdu(self, type):
pdu = PDU(type)
@ -93,15 +96,18 @@ class Network():
return pdu
def send_pdu(self, pdu):
if self.debug: pdu.dump()
if self.debug:
pdu.dump()
self.socket.send(pdu.encode())
def recv_pdu(self):
buf = self.socket.recv(100000)
if not buf: return None
if not buf:
return None
pdu = PDU()
pdu.decode(buf)
if self.debug: pdu.dump()
if self.debug:
pdu.dump()
return pdu
# =========================================
@ -109,7 +115,7 @@ class Network():
def _get_next_oid(self, oid, endoid):
if oid in self.data:
# Exact match found
#logger.debug('get_next_oid, exact match of %s' % oid)
# logger.debug('get_next_oid, exact match of %s' % oid)
idx = self.data_idx.index(oid)
if idx == (len(self.data_idx) - 1):
# Last Item in MIB, No match!
@ -117,11 +123,11 @@ class Network():
return self.data_idx[idx + 1]
else:
# No exact match, find prefix
#logger.debug('get_next_oid, no exact match of %s' % oid)
slist = oid.split('.')
elist = endoid.split('.')
# logger.debug('get_next_oid, no exact match of %s' % oid)
slist = oid.split(".")
elist = endoid.split(".")
for tmp_oid in self.data_idx:
tlist = tmp_oid.split('.')
tlist = tmp_oid.split(".")
for i in range(len(tlist)):
try:
sok = int(slist[i]) <= int(tlist[i])
@ -166,13 +172,13 @@ class Network():
def is_connected(self):
return self._connected
def run(self, timeout = 0.1):
def run(self, timeout=0.1):
if not self._connected:
raise NetworkError('Not connected')
raise NetworkError("Not connected")
if timeout != self._timeout:
self.socket.settimeout(timeout)
self._timeout = timeout
self.socket.settimeout(timeout)
self._timeout = timeout
try:
request = self.recv_pdu()
@ -195,11 +201,13 @@ class Network():
response.values.append(self.data[oid])
else:
logger.debug("OID Not Found!")
response.values.append({
'type': agentx.TYPE_NOSUCHOBJECT,
'name': rvalue[0],
'value': 0
})
response.values.append(
{
"type": agentx.TYPE_NOSUCHOBJECT,
"name": rvalue[0],
"value": 0,
}
)
elif request.type == agentx.AGENTX_GETNEXT_PDU:
logger.debug("Received GET_NEXT PDU")
@ -209,11 +217,13 @@ class Network():
if oid:
response.values.append(self.data[oid])
else:
response.values.append({
'type': agentx.TYPE_ENDOFMIBVIEW,
'name': rvalue[0],
'value': 0
})
response.values.append(
{
"type": agentx.TYPE_ENDOFMIBVIEW,
"name": rvalue[0],
"value": 0,
}
)
else:
logger.warn("Received unsupported PDU %d" % request.type)