]>
git.ipfire.org Git - thirdparty/pdns.git/blob - regression-tests.dnsdist/test_SNMP.py
5 from pysnmp
.hlapi
import *
6 from dnsdisttests
import DNSDistTest
8 class TestSNMP(DNSDistTest
):
11 _snmpServer
= '127.0.0.1'
13 _snmpV2Community
= 'secretcommunity'
14 _snmpV3User
= 'secretuser'
15 _snmpV3AuthKey
= 'mysecretauthkey'
16 _snmpV3EncKey
= 'mysecretenckey'
17 _snmpOID
= '1.3.6.1.4.1.43315.3'
19 _config_template
= """
20 newServer{address="127.0.0.1:%s", name="servername"}
24 def _checkStatsValues(self
, results
, queriesCountersValue
):
25 for i
in list(range(1, 5)) + list(range(6, 20)) + list(range(24, 35)) + [ 35 ] :
26 oid
= self
._snmpOID
+ '.1.' + str(i
) + '.0'
27 self
.assertTrue(oid
in results
)
28 self
.assertTrue(isinstance(results
[oid
], Counter64
))
30 for i
in range(20, 23):
31 oid
= self
._snmpOID
+ '.1.' + str(i
) + '.0'
32 self
.assertTrue(isinstance(results
[oid
], OctetString
))
35 self
.assertGreater(results
['1.3.6.1.4.1.43315.3.1.24.0'], 0)
36 # check memory usage > 0
37 self
.assertGreater(results
['1.3.6.1.4.1.43315.3.1.25.0'], 0)
39 # check that the queries, responses and rdQueries counters are now at queriesCountersValue
41 oid
= self
._snmpOID
+ '.1.' + str(i
) + '.0'
42 self
.assertEquals(results
[oid
], queriesCountersValue
)
44 # the others counters (except for latency ones) should still be at 0
45 for i
in [3, 4, 6, 7, 8, 9, 10, 11, 12, 13, 26, 27, 29, 30, 31, 35, 36]:
46 oid
= self
._snmpOID
+ '.1.' + str(i
) + '.0'
47 self
.assertEquals(results
[oid
], 0)
49 # check the backend stats
53 for i
in [3, 4, 5, 6, 7, 11, 12, 13]:
54 oid
= self
._snmpOID
+ '.2.1.' + str(i
) + '.0'
55 self
.assertTrue(isinstance(results
[oid
], Counter64
))
56 for i
in [2, 8, 9, 10]:
57 oid
= self
._snmpOID
+ '.2.1.' + str(i
) + '.0'
58 self
.assertTrue(isinstance(results
[oid
], OctetString
))
61 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.2.0'], "servername")
63 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.4.0'], 1)
65 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.5.0'], 0)
67 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.6.0'], 0)
69 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.7.0'], 0)
71 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.8.0'], "up")
73 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.9.0'], ("127.0.0.1:%s" % (self
._testServerPort
)))
75 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.10.0'], "")
77 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.12.0'], queriesCountersValue
)
79 self
.assertEquals(results
['1.3.6.1.4.1.43315.3.2.1.13.0'], 1)
81 def _getSNMPStats(self
, auth
):
83 for (errorIndication
, errorStatus
, errorIndex
, varBinds
) in nextCmd(SnmpEngine(),
85 UdpTransportTarget((self
._snmpServer
, self
._snmpPort
), timeout
=self
._snmpTimeout
),
87 ObjectType(ObjectIdentity(self
._snmpOID
)),
89 self
.assertFalse(errorIndication
)
90 self
.assertFalse(errorStatus
)
91 self
.assertTrue(varBinds
)
92 for key
, value
in varBinds
:
93 keystr
= key
.prettyPrint()
94 if not keystr
.startswith(self
._snmpOID
):
96 results
[keystr
] = value
100 def _checkStats(self
, auth
, name
):
101 # wait 1s so that the uptime is > 0
104 results
= self
._getSNMPStats
(auth
)
105 self
._checkStatsValues
(results
, self
.__class
__._queriesSent
)
107 query
= dns
.message
.make_query(name
, 'A', 'IN', use_edns
=False)
108 response
= dns
.message
.make_response(query
)
109 rrset
= dns
.rrset
.from_text(name
,
114 response
.answer
.append(rrset
)
117 (receivedQuery
, receivedResponse
) = self
.sendUDPQuery(query
, response
)
118 self
.assertTrue(receivedQuery
)
119 self
.assertTrue(receivedResponse
)
120 receivedQuery
.id = query
.id
121 self
.assertEquals(query
, receivedQuery
)
122 self
.assertEquals(response
, receivedResponse
)
123 self
.__class
__._queriesSent
= self
.__class
__._queriesSent
+ 1
125 results
= self
._getSNMPStats
(auth
)
126 self
._checkStatsValues
(results
, self
.__class
__._queriesSent
)
128 def testSNMPv2Stats(self
):
130 SNMP: Retrieve statistics via SNMPv2c
133 auth
= CommunityData(self
._snmpV
2Community
, mpModel
=1)
134 name
= 'simplea.snmpv2c.tests.powerdns.com.'
135 self
._checkStats
(auth
, name
)
137 def testSNMPv3Stats(self
):
139 SNMP: Retrieve statistics via SNMPv3
142 auth
= UsmUserData(self
._snmpV
3User
,
143 authKey
=self
._snmpV
3AuthKey
,
144 privKey
=self
._snmpV
3EncKey
,
145 authProtocol
=usmHMACSHAAuthProtocol
,
146 privProtocol
=usmAesCfb128Protocol
)
147 name
= 'simplea.snmpv2.tests.powerdns.com.'
148 self
._checkStats
(auth
, name
)