projects / thirdparty / pdns.git / blame
| Commit | Line | Data |
|---|---|---|
| aa7a54c9 RG |
1 | import time |
| 2 | ||
| 3 | from pysnmp.hlapi import * | |
| 4 | ||
| 5 | from recursortests import RecursorTest | |
| 6 | ||
| 7 | class TestSNMP(RecursorTest): | |
| 8 | ||
| 9 | _snmpTimeout = 2.0 | |
| 10 | _snmpServer = '127.0.0.1' | |
| 11 | _snmpPort = 161 | |
| 12 | _snmpV2Community = 'secretcommunity' | |
| 13 | _snmpV3User = 'secretuser' | |
| 14 | _snmpV3AuthKey = 'mysecretauthkey' | |
| 15 | _snmpV3EncKey = 'mysecretenckey' | |
| 16 | _snmpOID = '1.3.6.1.4.1.43315.2' | |
| 17 | _queriesSent = 0 | |
| 18 | _confdir = 'SNMP' | |
| 19 | _config_template = """ | |
| 20 | snmp-agent=yes | |
| 21 | """ | |
| 22 | ||
| 23 | def _checkStatsValues(self, results): | |
| 24 | for i in list(range(1, 93)): | |
| 25 | oid = self._snmpOID + '.1.' + str(i) + '.0' | |
| 26 | self.assertTrue(oid in results) | |
| 27 | self.assertTrue(isinstance(results[oid], Counter64)) | |
| 28 | ||
| 29 | # check uptime > 0 | |
| 30 | self.assertGreater(results['1.3.6.1.4.1.43315.2.1.75.0'], 0) | |
| 31 | # check memory usage > 0 | |
| 32 | self.assertGreater(results['1.3.6.1.4.1.43315.2.1.76.0'], 0) | |
| 33 | ||
| 34 | def _getSNMPStats(self, auth): | |
| 35 | results = {} | |
| 36 | for (errorIndication, errorStatus, errorIndex, varBinds) in nextCmd(SnmpEngine(), | |
| 37 | auth, | |
| 38 | UdpTransportTarget((self._snmpServer, self._snmpPort), timeout=self._snmpTimeout), | |
| 39 | ContextData(), | |
| 40 | ObjectType(ObjectIdentity(self._snmpOID)), | |
| 41 | lookupMib=False): | |
| 42 | self.assertFalse(errorIndication) | |
| 43 | self.assertFalse(errorStatus) | |
| 44 | self.assertTrue(varBinds) | |
| 45 | for key, value in varBinds: | |
| 46 | keystr = key.prettyPrint() | |
| 47 | if not keystr.startswith(self._snmpOID): | |
| 48 | continue | |
| 49 | results[keystr] = value | |
| 50 | ||
| 51 | return results | |
| 52 | ||
| 53 | def _checkStats(self, auth): | |
| 54 | # wait 1s so that the uptime is > 0 | |
| 55 | time.sleep(1) | |
| 56 | ||
| 57 | results = self._getSNMPStats(auth) | |
| 58 | self._checkStatsValues(results) | |
| 59 | ||
| 60 | def testSNMPv2Stats(self): | |
| 61 | """ | |
| 62 | SNMP: Retrieve statistics via SNMPv2c | |
| 63 | """ | |
| 64 | ||
| 65 | auth = CommunityData(self._snmpV2Community, mpModel=1) | |
| 66 | self._checkStats(auth) | |
| 67 | ||
| 68 | def testSNMPv3Stats(self): | |
| 69 | """ | |
| 70 | SNMP: Retrieve statistics via SNMPv3 | |
| 71 | """ | |
| 72 | ||
| 73 | auth = UsmUserData(self._snmpV3User, | |
| 74 | authKey=self._snmpV3AuthKey, | |
| 75 | privKey=self._snmpV3EncKey, | |
| 76 | authProtocol=usmHMACSHAAuthProtocol, | |
| 77 | privProtocol=usmAesCfb128Protocol) | |
| 78 | self._checkStats(auth) |