[Git][NTPsec/ntpsec][master] 2 commits: Moved walkMIBTree from agentx_packet.py to agentx.py.
Ian Bruene
gitlab at mg.gitlab.com
Fri Mar 16 20:47:15 UTC 2018
Ian Bruene pushed to branch master at NTPsec / ntpsec
Commits:
a05e6f8b by Ian Bruene at 2018-03-16T15:55:29Z
Moved walkMIBTree from agentx_packet.py to agentx.py.
- - - - -
6cb39cc3 by Ian Bruene at 2018-03-16T20:43:57Z
Changed MIB tree initialization mechanism.
Previously the MIB tree was created manually via a Nested Dict of Doom.
Now each entry is inited via the addNode method.
- - - - -
6 changed files:
- ntpclients/ntpsnmpd.py
- pylib/agentx.py
- pylib/agentx_packet.py
- + tests/pylib/test_agentx.py
- tests/pylib/test_agentx_packet.py
- tests/wscript
Changes:
=====================================
ntpclients/ntpsnmpd.py
=====================================
--- a/ntpclients/ntpsnmpd.py
+++ b/ntpclients/ntpsnmpd.py
@@ -48,223 +48,110 @@ class DataSource(ntp.agentx.MIBControl):
# This is defined as a dict tree because it is simpler, and avoids
# certain edge cases
# OIDs are relative from ntp root
- oidTree = {
- # ntpEntNotifications
- 0: {"subids":
- # ntpEntNotifModeChange
- {1: {},
- # ntpEntNotifStratumChange
- 2: {},
- # ntpEntNotifSyspeerChange
- 3: {},
- # ntpEntNotifAddAssociation
- 4: {},
- # ntpEntNotifRemoveAsociation
- 5: {},
- # ntpEntNotifConfigChanged
- 6: {},
- # ntpEntNotifLeapSecondAnnounced
- 7: {},
- # ntpEntNotifHeartbeat
- 8: {}}},
- # ntpSnmpMIBObjects
- 1: {"subids":
- # ntpEntInfo
- {1:
- {"subids":
- # ntpNetSoftwareName utf8str
- {1: {"subids":
- {0:
- {"reader":
- (lambda oid: self.cbr_systemInfo(oid, "name"))}}},
- # ntpEntSoftwareVersion utf8str
- 2: {"subids":
- {0:
- {"reader":
- (lambda oid: self.cbr_systemInfo(oid, "version"))}}},
- # ntpEntSoftwareVendor utf8str
- 3: {"subids":
- {0:
- {"reader":
- (lambda oid: self.cbr_systemInfo(oid, "vendor"))}}},
- # ntpEntSystemType utf8str
- 4: {"subids":
- {0:
- {"reader":
- (lambda oid: self.cbr_systemInfo(oid, "system"))}}},
- # ntpEntTimeResolution uint32
- 5: {"subids":
- {0:
- {"reader": self.cbr_timeResolution}}},
- # ntpEntTimePrecision int32
- 6: {"subids":
- {0:
- {"reader": self.cbr_timePrecision}}},
- # ntpEntTimeDistance DisplayString
- 7: {"subids":
- {0:
- {"reader": self.cbr_timeDistance}}}}},
- # ntpEntStatus
- 2: {"subids":
- # ntpEntStatusCurrentMode INTEGER {...}
- {1: {"subids":
- {0:
- {"reader": self.cbr_statusCurrentMode}}},
- # ntpEntStatusStratum NtpStratum
- 2: {"subids":
- {0:
- {"reader": self.cbr_statusStratum}}},
- # ntpEntStatusActiveRefSourceId
- # uint32 (0..99999)
- 3: {"subids":
- {0:
- {"reader": self.cbr_statusActiveRefSourceID}}},
- # ntpEntStatusActiveRefSourceName utf8str
- 4: {"subids":
- {0:
- {"reader": self.cbr_statusActiveRefSourceName}}},
- # ntpEntStatusActiveOffset DisplayString
- 5: {"subids":
- {0:
- {"reader": self.cbr_statusActiveOffset}}},
- # ntpEntStatusNumberOfRefSources
- # unit32 (0..99)
- 6: {"subids":
- {0:
- {"reader": self.cbr_statusNumRefSources}}},
- # ntpEntStatusDispersion DisplayString
- 7: {"subids":
- {0:
- {"reader": self.cbr_statusDispersion}}},
- # ntpEntStatusEntityUptime TimeTicks
- 8: {"subids":
- {0:
- {"reader": self.cbr_statusEntityUptime}}},
- # ntpEntStatusDateTime NtpDateTime
- 9: {"subids":
- {0:
- {"reader": self.cbr_statusDateTime}}},
- # ntpEntStatusLeapSecond NtpDateTime
- 10: {"subids":
- {0:
- {"reader": self.cbr_statusLeapSecond}}},
- # ntpEntStatusLeapSecondDirection
- # int32 (-1..1)
- 11: {"subids":
- {0:
- {"reader": self.cbr_statusLeapSecDirection}}},
- # ntpEntStatusInPkts Counter32
- 12: {"subids":
- {0:
- {"reader": self.cbr_statusInPkts}}},
- # ntpEntStatusOutPkts Counter32
- 13: {"subids":
- {0:
- {"reader": self.cbr_statusOutPkts}}},
- # ntpEntStatusBadVersion Counter32
- 14: {"subids":
- {0:
- {"reader": self.cbr_statusBadVersion}}},
- # ntpEntStatusProtocolError Counter32
- 15: {"subids":
- {0:
- {"reader": self.cbr_statusProtocolError}}},
- # ntpEntStatusNotifications Counter32
- 16: {"subids":
- {0:
- {"reader": self.cbr_statusNotifications}}},
- # ntpEntStatPktModeTable
- # SEQUENCE of NtpEntStatPktModeEntry
- 17: {"subids":
- # ntpEntStatPktModeEntry SEQUENCE {...}
- {1: {"subids":
- # ntpEntStatPktMode INTEGER {...}
- {1: {},
- # ntpEntStatPktSent Counter32
- 2: {},
- # ntpEntStatPktRecived Counter32
- 3: {}}}}}}},
- # ntpAssociation
- 3: {"subids":
- # ntpAssociationTable
- # SEQUENCE of NtpAssociationEntry
- {1: {"subids":
- # ntpAssociationEntry SEQUENCE {...}
- {1: {"subids":
- # ntpAssocId uint32 (1..99999)
- {1: {"subids": self.sub_assocID},
- # ntpAssocName utf8str
- 2: {"subids": self.sub_assocName},
- # ntpAssocRefId DisplayString
- 3: {"subids": self.sub_assocRefID},
- # ntpAssocAddressType
- # InetAddressType
- 4: {"subids": self.sub_assocAddrType},
- # ntpAssocAddress
- # InetAddress SIZE (4|8|16|20)
- 5: {"subids": self.sub_assocAddr},
- # ntpAssocOffset DisplayString
- 6: {"subids": self.sub_assocOffset},
- # ntpAssocStratum NtpStratum
- 7: {"subids": self.sub_assocStratum},
- # ntpAssocStatusJitter
- # DisplayString
- 8: {"subids": self.sub_assocJitter},
- # ntpAssocStatusDelay
- # DisplayString
- 9: {"subids": self.sub_assocDelay},
- # ntpAssocStatusDispersion
- # DisplayString
- 10: {"subids": self.sub_assocDispersion}}}}},
- # ntpAssociationStatisticsTable
- # SEQUENCE of ntpAssociationStatisticsEntry
- 2: {"subids":
- # ntpAssociationStatisticsEntry
- # SEQUENCE {...}
- {1: {"subids":
- # ntpAssocStatInPkts Counter32
- {1: {"subids": self.sub_assocStatInPkts},
- # ntpAssocStatOutPkts Counter32
- 2: {"subids": self.sub_assocStatOutPkts},
- # ntpAssocStatProtocolError
- # Counter32
- 3: {"subids":
- self.sub_assocStatProtoErr}}}}}}},
- # ntpEntControl
- 4: {"subids":
- # ntpEntHeartbeatInterval unit32
- {1: {"subids":
- {0:
- {"reader": self.cbr_entHeartbeatInterval,
- "writer": self.cbw_entHeartbeatInterval}}},
- # ntpEntNotifBits BITS {...}
- 2: {"subids":
- {0:
- {"reader": self.cbr_entNotifBits,
- "writer": self.cbw_entNotifBits}}}}},
- # ntpEntNotifObjects
- 5: {"subids":
- # ntpEntNotifMessage utf8str
- {1: {"subids":
- {0:
- {"reader": self.cbr_entNotifMessage}}}}}}},
- # ntpEntConformance
- 2: {"subids":
- # ntpEntCompliances
- {1: {"subids":
- # ntpEntNTPCompliance
- {1: {},
- # ntpEntSNTPCompliance
- 2: {}}},
- # ntpEntGroups
- 2: {"subids":
- # ntpEntObjectsGroup1 OBJECTS {...}
- {1: {},
- # ntpEntObjectsGroup2 OBJECTS {...}
- 2: {},
- # ntpEntNotifGroup NOTIFICATIONS {...}
- 3: {}}}}}}
- ntp.agentx.MIBControl.__init__(self, oidTree, mibRoot=ntpRootOID)
+ ntp.agentx.MIBControl.__init__(self, mibRoot=ntpRootOID)
+ # MIB node init
+ # block 0
+ self.addNode((0,)) # ntpEntNotifications
+ self.addNode((0, 1)) # ntpEntNotifModeChange
+ self.addNode((0, 2)) # ntpEntNotifStratumChange
+ self.addNode((0, 3)) # ntpEntNotifSyspeerChange
+ self.addNode((0, 4)) # ntpEntNotifAddAssociation
+ self.addNode((0, 5)) # ntpEntNotifRemoveAsociation
+ self.addNode((0, 6)) # ntpEntNotifConfigChanged
+ self.addNode((0, 7)) # ntpEntNotifLeapSecondAnnounced
+ self.addNode((0, 8)) # ntpEntNotifHeartbeat
+ # block 1
+ # block 1, 1
+ self.addNode((1, 1, 1, 0), # ntpNetSoftwareName utf8str
+ (lambda oid: self.cbr_systemInfo(oid, "name")))
+ self.addNode((1, 1, 2, 0), # ntpEntSoftwareVersion utf8str
+ (lambda oid: self.cbr_systemInfo(oid, "version")))
+ self.addNode((1, 1, 3, 0), # ntpEntSoftwareVendor utf8str
+ (lambda oid: self.cbr_systemInfo(oid, "vendor")))
+ self.addNode((1, 1, 4, 0), # ntpEntSystemType utf8str
+ (lambda oid: self.cbr_systemInfo(oid, "system")))
+ self.addNode((1, 1, 5, 0), # ntpEntTimeResolution uint32
+ self.cbr_timeResolution)
+ self.addNode((1, 1, 6, 0), # ntpEntTimePrecision int32
+ self.cbr_timePrecision)
+ self.addNode((1, 1, 7, 0), # ntpEntTimeDistance DisplayString
+ self.cbr_timeDistance)
+ # block 1, 2
+ self.addNode((1, 2, 1, 0), # ntpEntStatusCurrentMode INTEGER {...}
+ self.cbr_statusCurrentMode)
+ self.addNode((1, 2, 2, 0), # ntpEntStatusStratum NtpStratum
+ self.cbr_statusStratum)
+ self.addNode((1, 2, 3, 0), # ntpEntStatusActiveRefSourceId uint32
+ self.cbr_statusActiveRefSourceID)
+ self.addNode((1, 2, 4, 0), # ntpEntStatusActiveRefSourceName utf8str
+ self.cbr_statusActiveRefSourceName)
+ self.addNode((1, 2, 5, 0), # ntpEntStatusActiveOffset DisplayString
+ self.cbr_statusActiveOffset)
+ self.addNode((1, 2, 6, 0), # ntpEntStatusNumberOfRefSources unit32
+ self.cbr_statusNumRefSources)
+ self.addNode((1, 2, 7, 0), # ntpEntStatusDispersion DisplayString
+ self.cbr_statusDispersion)
+ self.addNode((1, 2, 8, 0), # ntpEntStatusEntityUptime TimeTicks
+ self.cbr_statusEntityUptime)
+ self.addNode((1, 2, 9, 0), # ntpEntStatusDateTime NtpDateTime
+ self.cbr_statusDateTime)
+ self.addNode((1, 2, 10, 0), # ntpEntStatusLeapSecond NtpDateTime
+ self.cbr_statusLeapSecond)
+ self.addNode((1, 2, 11, 0), # ntpEntStatusLeapSecondDirection int32
+ self.cbr_statusLeapSecDirection)
+ self.addNode((1, 2, 12, 0), # ntpEntStatusInPkts Counter32
+ self.cbr_statusInPkts)
+ self.addNode((1, 2, 13, 0), # ntpEntStatusOutPkts Counter32
+ self.cbr_statusOutPkts)
+ self.addNode((1, 2, 14, 0), # ntpEntStatusBadVersion Counter32
+ self.cbr_statusBadVersion)
+ self.addNode((1, 2, 15, 0), # ntpEntStatusProtocolError Counter32
+ self.cbr_statusProtocolError)
+ self.addNode((1, 2, 16, 0), # ntpEntStatusNotifications Counter32
+ self.cbr_statusNotifications)
+ self.addNode((1, 2, 17, 1, 1)) # ntpEntStatPktMode INTEGER {...}
+ self.addNode((1, 2, 17, 1, 2)) # ntpEntStatPktSent Counter32
+ self.addNode((1, 2, 17, 1, 3)) # ntpEntStatPktRecived Counter32
+ # block 1, 3
+ self.addNode((1, 3, 1, 1, 1), # ntpAssocId uint32 (1..99999)
+ dynamic=self.sub_assocID)
+ self.addNode((1, 3, 1, 1, 2), # ntpAssocName utf8str
+ dynamic=self.sub_assocName)
+ self.addNode((1, 3, 1, 1, 3), # ntpAssocRefId DisplayString
+ dynamic=self.sub_assocRefID)
+ self.addNode((1, 3, 1, 1, 4), # ntpAssocAddressType InetAddressType
+ dynamic=self.sub_assocAddrType)
+ self.addNode((1, 3, 1, 1, 5), # ntpAssocAddress InetAddress SIZE
+ dynamic=self.sub_assocAddr)
+ self.addNode((1, 3, 1, 1, 6), # ntpAssocOffset DisplayString
+ dynamic=self.sub_assocOffset)
+ self.addNode((1, 3, 1, 1, 7), # ntpAssocStratum NtpStratum
+ dynamic=self.sub_assocStratum)
+ self.addNode((1, 3, 1, 1, 8), # ntpAssocStatusJitter DisplayString
+ dynamic=self.sub_assocJitter)
+ self.addNode((1, 3, 1, 1, 9), # ntpAssocStatusDelay DisplayString
+ dynamic=self.sub_assocDelay)
+ self.addNode((1, 3, 1, 1, 10), # ntpAssocStatusDispersion DisplayStr
+ dynamic=self.sub_assocDispersion)
+ self.addNode((1, 3, 2, 1, 1), # ntpAssocStatInPkts Counter32
+ dynamic=self.sub_assocStatInPkts)
+ self.addNode((1, 3, 2, 1, 2), # ntpAssocStatOutPkts Counter32
+ dynamic=self.sub_assocStatOutPkts)
+ self.addNode((1, 3, 2, 1, 3), # ntpAssocStatProtocolError Counter32
+ dynamic=self.sub_assocStatProtoErr)
+ # block 1, 4
+ self.addNode((1, 4, 1, 0), # ntpEntHeartbeatInterval unit32
+ self.cbr_entHeartbeatInterval,
+ self.cbw_entHeartbeatInterval)
+ self.addNode((1, 4, 2, 0), # ntpEntNotifBits BITS {...}
+ self.cbr_entNotifBits,
+ self.cbw_entNotifBits)
+ # block 1, 5
+ self.addNode((1, 5, 1, 0), # ntpEntNotifMessage utf8str
+ self.cbr_entNotifMessage)
+ # block 2 # all compliance statements
+ #print(repr(self.oidTree))
+ #print(self.oidTree[1]["subids"][1][1][0])
self.session = ntp.packet.ControlSession()
self.hostname = hostname if hostname else DEFHOST
self.session.openhost(self.hostname)
=====================================
pylib/agentx.py
=====================================
--- a/pylib/agentx.py
+++ b/pylib/agentx.py
@@ -23,7 +23,7 @@ pingTime = 60
class MIBControl:
- def __init__(self, oidTree, mibRoot=(), rangeSubid=0, upperBound=None,
+ def __init__(self, oidTree={}, mibRoot=(), rangeSubid=0, upperBound=None,
mibContext=None):
self.oidTree = oidTree # contains callbacks for the MIB
# The undo system is only for the last operation
@@ -48,8 +48,36 @@ class MIBControl:
def mib_context(self):
return self.context
+ def addNode(self, oid, reader=None, writer=None, dynamic=None):
+ if isinstance(oid, ax.OID): # get it in a mungable format
+ oid = tuple(oid.subids)
+ # dynamic is the generator for tables
+ currentLevel = self.oidTree
+ remainingOID = oid
+ while True:
+ print(remainingOID)
+ node, remainingOID = ntp.util.slicedata(remainingOID, 1)
+ node = node[0]
+ print("Nodes at this level:", currentLevel.keys())
+ if node not in currentLevel.keys():
+ print("Adding node:", node)
+ currentLevel[node] = {"reader":None, "writer":None,
+ "subids":None}
+ if len(remainingOID) == 0: # We have reached the target node
+ print("Reached target, filling it")
+ currentLevel[node]["reader"] = reader
+ currentLevel[node]["writer"] = writer
+ if dynamic is not None:
+ # can't be both dynamic and non-dynamic
+ currentLevel[node]["subids"] = dynamic
+ return
+ else:
+ if currentLevel[node]["subids"] is None:
+ currentLevel[node]["subids"] = {}
+ currentLevel = currentLevel[node]["subids"]
+
def getOID_core(self, nextP, searchoid, returnGenerator=False):
- gen = ax.walkMIBTree(self.oidTree, self.mibRoot)
+ gen = walkMIBTree(self.oidTree, self.mibRoot)
while True:
try:
oid, reader, writer = gen.next()
@@ -84,7 +112,7 @@ class MIBControl:
def getOIDsInRange(self, oidrange, firstOnly=False):
"Get a list of every (optionally the first) OID in a range"
oids = []
- gen = ax.walkMIBTree(self.oidTree, self.mibRoot)
+ gen = walkMIBTree(self.oidTree, self.mibRoot)
# Find the first OID
while True:
try:
@@ -500,3 +528,49 @@ class PacketControl:
# Ok, response with no associated packet.
# Probably something that timed out.
pass
+
+
+def walkMIBTree(tree, rootpath=()):
+ # Tree node formats:
+ # {"reader": <func>, "writer": <func>, "subids": {.blah.}}
+ # {"reader": <func>, "writer": <func>, "subids": <func>}
+ # The "subids" function in dynamic nodes must return an MIB tree
+ nodeStack = []
+ oidStack = []
+ current = tree
+ currentKeys = list(current.keys())
+ currentKeys.sort()
+ keyID = 0
+ while True:
+ if keyID >= len(currentKeys):
+ if len(nodeStack) > 0:
+ # No more nodes this level, pop higher node
+ current, currentKeys, keyID, key = nodeStack.pop()
+ oidStack.pop()
+ keyID += 1
+ continue
+ else: # Out of tree, we are done
+ return
+ key = currentKeys[keyID]
+ oid = ax.OID(rootpath + tuple(oidStack) + (key,))
+ yield (oid, current[key].get("reader"), current[key].get("writer"))
+ subs = current[key].get("subids")
+ if subs is not None:
+ # Push current node, move down a level
+ nodeStack.append((current, currentKeys, keyID, key))
+ oidStack.append(key)
+ if isinstance(subs, dict) is True:
+ current = subs
+ else:
+ current = subs() # Tree generator function
+ if current == {}: # no dynamic subids, pop
+ current, currentKeys, keyID, key = nodeStack.pop()
+ oidStack.pop()
+ keyID += 1
+ continue
+ currentKeys = list(current.keys())
+ currentKeys.sort()
+ keyID = 0
+ key = currentKeys[keyID]
+ continue
+ keyID += 1
=====================================
pylib/agentx_packet.py
=====================================
--- a/pylib/agentx_packet.py
+++ b/pylib/agentx_packet.py
@@ -1297,52 +1297,6 @@ def decode_packet(data):
return parsedPkt, newData
-def walkMIBTree(tree, rootpath=()):
- # Tree node formats:
- # {"reader": <func>, "writer": <func>, "subids": {.blah.}}
- # {"reader": <func>, "writer": <func>, "subids": <func>}
- # The "subids" function in dynamic nodes must return an MIB tree
- nodeStack = []
- oidStack = []
- current = tree
- currentKeys = list(current.keys())
- currentKeys.sort()
- keyID = 0
- while True:
- if keyID >= len(currentKeys):
- if len(nodeStack) > 0:
- # No more nodes this level, pop higher node
- current, currentKeys, keyID, key = nodeStack.pop()
- oidStack.pop()
- keyID += 1
- continue
- else: # Out of tree, we are done
- return
- key = currentKeys[keyID]
- oid = OID(rootpath + tuple(oidStack) + (key,))
- yield (oid, current[key].get("reader"), current[key].get("writer"))
- subs = current[key].get("subids")
- if subs is not None:
- # Push current node, move down a level
- nodeStack.append((current, currentKeys, keyID, key))
- oidStack.append(key)
- if isinstance(subs, dict) is True:
- current = subs
- else:
- current = subs() # Tree generator function
- if current == {}: # no dynamic subids, pop
- current, currentKeys, keyID, key = nodeStack.pop()
- oidStack.pop()
- keyID += 1
- continue
- currentKeys = list(current.keys())
- currentKeys.sort()
- keyID = 0
- key = currentKeys[keyID]
- continue
- keyID += 1
-
-
def bits2Bools(bitString, cropLength=None):
bits = []
for octet in bitString:
=====================================
tests/pylib/test_agentx.py
=====================================
--- /dev/null
+++ b/tests/pylib/test_agentx.py
@@ -0,0 +1,104 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import ntp.agentx as AX
+
+import unittest
+
+class TestAgentx(unittest.TestCase):
+ def test_walkMIBTree(self):
+ f = AX.walkMIBTree
+
+ # Test empty tree
+ self.assertEqual(tuple(f({})), ())
+ # Test flat, fully static tree
+ self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
+ "subids": None},
+ 1: {},
+ 2: {"reader": None, "writer": None,
+ "subids": None},
+ 5: {"reader": None, "writer": None,
+ "subids": None}})),
+ ((AX.OID((0,)), None, None),
+ (AX.OID((1,)), None, None),
+ (AX.OID((2,)), None, None),
+ (AX.OID((5,)), None, None)))
+ # Test nested, fully static tree
+ self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
+ "subids": None},
+ 1: {"reader": None, "writer": None,
+ "subids":
+ {0: {"reader": None, "writer": None,
+ "subids": None},
+ 1: {"reader": None, "writer": None,
+ "subids":
+ {42: {"reader": None,
+ "writer": None,
+ "subids": None}}}}},
+ 5: {"reader": None, "writer": None,
+ "subids": None}})),
+ ((AX.OID((0,)), None, None),
+ (AX.OID((1,)), None, None),
+ (AX.OID((1, 0)), None, None),
+ (AX.OID((1, 1)), None, None),
+ (AX.OID((1, 1, 42)), None, None),
+ (AX.OID((5,)), None, None)))
+ # Test nested, fully static tree, with rootpath
+ self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
+ "subids": None},
+ 1: {"reader": None, "writer": None,
+ "subids":
+ {0: {"reader": None, "writer": None,
+ "subids": None},
+ 1: {"reader": None, "writer": None,
+ "subids":
+ {42: {"reader": None,
+ "writer": None,
+ "subids": None}}}}},
+ 5: {"reader": None, "writer": None,
+ "subids": None}},
+ (23,))),
+ ((AX.OID((23, 0)), None, None),
+ (AX.OID((23, 1)), None, None),
+ (AX.OID((23, 1, 0)), None, None),
+ (AX.OID((23, 1, 1)), None, None),
+ (AX.OID((23, 1, 1, 42)), None, None),
+ (AX.OID((23, 5)), None, None)))
+ # subid lambda for dynamic tree testing
+ submaker = (lambda: {0: {"reader": None, "writer": None,
+ "subids": None},
+ 1: {"reader": None, "writer": None,
+ "subids":
+ {0: {"reader": None, "writer": None,
+ "subids": None}}},
+ 2: {"reader": None, "writer": None,
+ "subids": None}})
+ # Test tree with dynamic nodes
+ self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
+ "subids": None},
+ 1: {"reader": None, "writer": None,
+ "subids": submaker},
+ 2: {"reader": None, "writer": None,
+ "subids": None}})),
+ ((AX.OID((0,)), None, None),
+ (AX.OID((1,)), None, None),
+ (AX.OID((1, 0)), None, None),
+ (AX.OID((1, 1)), None, None),
+ (AX.OID((1, 1, 0)), None, None),
+ (AX.OID((1, 2)), None, None),
+ (AX.OID((2,)), None, None)))
+ # Test tree with dynamic nodes and root path
+ self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
+ "static": True, "subids": None},
+ 1: {"reader": None, "writer": None,
+ "static": False, "subids": submaker},
+ 2: {"reader": None, "writer": None,
+ "static": True, "subids": None}},
+ (23,))),
+ ((AX.OID((23, 0)), None, None),
+ (AX.OID((23, 1)), None, None),
+ (AX.OID((23, 1, 0)), None, None),
+ (AX.OID((23, 1, 1)), None, None),
+ (AX.OID((23, 1, 1, 0)), None, None),
+ (AX.OID((23, 1, 2)), None, None),
+ (AX.OID((23, 2)), None, None)))
=====================================
tests/pylib/test_agentx_packet.py
=====================================
--- a/tests/pylib/test_agentx_packet.py
+++ b/tests/pylib/test_agentx_packet.py
@@ -2598,103 +2598,6 @@ class TestNtpclientsNtpsnmpd(unittest.TestCase):
errored = e
self.assertEqual(errored.message, "PDU type 255 not in defined types")
- def test_walkMIBTree(self):
- f = AX.walkMIBTree
-
- # Test empty tree
- self.assertEqual(tuple(f({})), ())
- # Test flat, fully static tree
- self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
- "subids": None},
- 1: {},
- 2: {"reader": None, "writer": None,
- "subids": None},
- 5: {"reader": None, "writer": None,
- "subids": None}})),
- ((AX.OID((0,)), None, None),
- (AX.OID((1,)), None, None),
- (AX.OID((2,)), None, None),
- (AX.OID((5,)), None, None)))
- # Test nested, fully static tree
- self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
- "subids": None},
- 1: {"reader": None, "writer": None,
- "subids":
- {0: {"reader": None, "writer": None,
- "subids": None},
- 1: {"reader": None, "writer": None,
- "subids":
- {42: {"reader": None,
- "writer": None,
- "subids": None}}}}},
- 5: {"reader": None, "writer": None,
- "subids": None}})),
- ((AX.OID((0,)), None, None),
- (AX.OID((1,)), None, None),
- (AX.OID((1, 0)), None, None),
- (AX.OID((1, 1)), None, None),
- (AX.OID((1, 1, 42)), None, None),
- (AX.OID((5,)), None, None)))
- # Test nested, fully static tree, with rootpath
- self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
- "subids": None},
- 1: {"reader": None, "writer": None,
- "subids":
- {0: {"reader": None, "writer": None,
- "subids": None},
- 1: {"reader": None, "writer": None,
- "subids":
- {42: {"reader": None,
- "writer": None,
- "subids": None}}}}},
- 5: {"reader": None, "writer": None,
- "subids": None}},
- (23,))),
- ((AX.OID((23, 0)), None, None),
- (AX.OID((23, 1)), None, None),
- (AX.OID((23, 1, 0)), None, None),
- (AX.OID((23, 1, 1)), None, None),
- (AX.OID((23, 1, 1, 42)), None, None),
- (AX.OID((23, 5)), None, None)))
- # subid lambda for dynamic tree testing
- submaker = (lambda: {0: {"reader": None, "writer": None,
- "subids": None},
- 1: {"reader": None, "writer": None,
- "subids":
- {0: {"reader": None, "writer": None,
- "subids": None}}},
- 2: {"reader": None, "writer": None,
- "subids": None}})
- # Test tree with dynamic nodes
- self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
- "subids": None},
- 1: {"reader": None, "writer": None,
- "subids": submaker},
- 2: {"reader": None, "writer": None,
- "subids": None}})),
- ((AX.OID((0,)), None, None),
- (AX.OID((1,)), None, None),
- (AX.OID((1, 0)), None, None),
- (AX.OID((1, 1)), None, None),
- (AX.OID((1, 1, 0)), None, None),
- (AX.OID((1, 2)), None, None),
- (AX.OID((2,)), None, None)))
- # Test tree with dynamic nodes and root path
- self.assertEqual(tuple(f({0: {"reader": None, "writer": None,
- "static": True, "subids": None},
- 1: {"reader": None, "writer": None,
- "static": False, "subids": submaker},
- 2: {"reader": None, "writer": None,
- "static": True, "subids": None}},
- (23,))),
- ((AX.OID((23, 0)), None, None),
- (AX.OID((23, 1)), None, None),
- (AX.OID((23, 1, 0)), None, None),
- (AX.OID((23, 1, 1)), None, None),
- (AX.OID((23, 1, 1, 0)), None, None),
- (AX.OID((23, 1, 2)), None, None),
- (AX.OID((23, 2)), None, None)))
-
def test_bits2Bools(self):
bits2bool = AX.bits2Bools
=====================================
tests/wscript
=====================================
--- a/tests/wscript
+++ b/tests/wscript
@@ -127,6 +127,7 @@ def build(ctx):
os.symlink(pypath.abspath(), linkpath.abspath())
pytests = ["pylib/test_util.py",
+ "pylib/test_agentx.py",
"pylib/test_agentx_packet.py",
"pylib/test_packet.py",
"pylib/test_statfiles.py"]
View it on GitLab: https://gitlab.com/NTPsec/ntpsec/compare/806bd0d24f331d4a7cc89522d187b56815baddb7...6cb39cc37a40774f85081ef66e69869ddbf5e4bb
---
View it on GitLab: https://gitlab.com/NTPsec/ntpsec/compare/806bd0d24f331d4a7cc89522d187b56815baddb7...6cb39cc37a40774f85081ef66e69869ddbf5e4bb
You're receiving this email because of your account on gitlab.com.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.ntpsec.org/pipermail/vc/attachments/20180316/e74df514/attachment.html>
More information about the vc
mailing list