openstack/ironic
Code Issues Proposed changes

Merge "Migrate ironic snmp driver to the latest pysnmp API"

Browse Source
This commit is contained in:
Zuul 2018-07-16 19:47:12 +00:00 committed by Gerrit Code Review
commit 157739435e
6 changed files with 173 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
driver-requirements.txt
View File

@ -5,7 +5,7 @@
# These are available on pypi
proliantutils>=2.5.0
pysnmp
pysnmp>=4.3.0,<5.0.0
python-ironic-inspector-client>=1.5.0
python-oneviewclient<3.0.0,>=2.5.2
python-scciclient>=0.7.1

121
ironic/drivers/modules/snmp.py
View File

@ -41,24 +41,23 @@ from ironic.drivers import base
pysnmp = importutils.try_import('pysnmp')
if pysnmp:
from pysnmp.entity.rfc3413.oneliner import cmdgen
from pysnmp import error as snmp_error
from pysnmp.proto import rfc1902
from pysnmp import hlapi as snmp
snmp_auth_protocols = {
'md5': cmdgen.usmHMACMD5AuthProtocol,
'sha': cmdgen.usmHMACSHAAuthProtocol,
'none': cmdgen.usmNoAuthProtocol,
'md5': snmp.usmHMACMD5AuthProtocol,
'sha': snmp.usmHMACSHAAuthProtocol,
'none': snmp.usmNoAuthProtocol,
}
# available since pysnmp 4.4.1
try:
snmp_auth_protocols.update(
{
'sha224': cmdgen.usmHMAC128SHA224AuthProtocol,
'sha256': cmdgen.usmHMAC192SHA256AuthProtocol,
'sha384': cmdgen.usmHMAC256SHA384AuthProtocol,
'sha512': cmdgen.usmHMAC384SHA512AuthProtocol,
'sha224': snmp.usmHMAC128SHA224AuthProtocol,
'sha256': snmp.usmHMAC192SHA256AuthProtocol,
'sha384': snmp.usmHMAC256SHA384AuthProtocol,
'sha512': snmp.usmHMAC384SHA512AuthProtocol,
}
)
@ -67,20 +66,20 @@ if pysnmp:
pass
snmp_priv_protocols = {
'des': cmdgen.usmDESPrivProtocol,
'3des': cmdgen.usm3DESEDEPrivProtocol,
'aes': cmdgen.usmAesCfb128Protocol,
'aes192': cmdgen.usmAesCfb192Protocol,
'aes256': cmdgen.usmAesCfb256Protocol,
'none': cmdgen.usmNoPrivProtocol,
'des': snmp.usmDESPrivProtocol,
'3des': snmp.usm3DESEDEPrivProtocol,
'aes': snmp.usmAesCfb128Protocol,
'aes192': snmp.usmAesCfb192Protocol,
'aes256': snmp.usmAesCfb256Protocol,
'none': snmp.usmNoPrivProtocol,
}
# available since pysnmp 4.4.3
try:
snmp_priv_protocols.update(
{
'aes192blmt': cmdgen.usmAesBlumenthalCfb192Protocol,
'aes256blmt': cmdgen.usmAesBlumenthalCfb256Protocol,
'aes192blmt': snmp.usmAesBlumenthalCfb192Protocol,
'aes256blmt': snmp.usmAesBlumenthalCfb256Protocol,
}
)
@ -89,9 +88,8 @@ if pysnmp:
pass
else:
cmdgen = None
snmp = None
snmp_error = None
rfc1902 = None
snmp_auth_protocols = {
'none': None
@ -198,7 +196,7 @@ class SNMPClient(object):
user=None, auth_proto=None,
auth_key=None, priv_proto=None,
priv_key=None, context_engine_id=None, context_name=None):
if not cmdgen:
if not snmp:
raise exception.DriverLoadError(
driver=self.__class__.__name__,
reason=_("Unable to import python-pysnmp library")
@ -213,13 +211,14 @@ class SNMPClient(object):
self.auth_key = auth_key
self.priv_proto = priv_proto
self.priv_key = priv_key
self.context_engine_id = context_engine_id
self.context_name = context_name or ''
else:
self.read_community = read_community
self.write_community = write_community
self.cmd_gen = cmdgen.CommandGenerator()
self.context_engine_id = context_engine_id
self.context_name = context_name or ''
self.snmp_engine = snmp.SnmpEngine()
def _get_auth(self, write_mode=False):
"""Return the authorization data for an SNMP request.
@ -227,23 +226,22 @@ class SNMPClient(object):
:param write_mode: `True` if write class SNMP command is
executed. Default is `False`.
:returns: Either
:class:`pysnmp.entity.rfc3413.oneliner.cmdgen.CommunityData`
or :class:`pysnmp.entity.rfc3413.oneliner.cmdgen.UsmUserData`
:class:`pysnmp.hlapi.CommunityData`
or :class:`pysnmp.hlapi.UsmUserData`
object depending on SNMP version being used.
"""
if self.version == SNMP_V3:
return cmdgen.UsmUserData(
return snmp.UsmUserData(
self.user,
authKey=self.auth_key,
authProtocol=self.auth_proto,
privKey=self.priv_key,
privProtocol=self.priv_proto,
contextEngineId=self.context_engine_id,
contextName=self.context_name
privProtocol=self.priv_proto
)
else:
mp_model = 1 if self.version == SNMP_V2C else 0
return cmdgen.CommunityData(
return snmp.CommunityData(
self.write_community if write_mode else self.read_community,
mpModel=mp_model
)
@ -252,17 +250,31 @@ class SNMPClient(object):
"""Return the transport target for an SNMP request.
:returns: A :class:
`pysnmp.entity.rfc3413.oneliner.cmdgen.UdpTransportTarget` object.
:raises: snmp_error.PySnmpError if the transport address is bad.
`pysnmp.hlapi.UdpTransportTarget` object.
:raises: :class:`pysnmp.error.PySnmpError` if the transport address
is bad.
"""
# The transport target accepts timeout and retries parameters, which
# default to 1 (second) and 5 respectively. These are deemed sensible
# enough to allow for an unreliable network or slow device.
return cmdgen.UdpTransportTarget(
return snmp.UdpTransportTarget(
(self.address, self.port),
timeout=CONF.snmp.udp_transport_timeout,
retries=CONF.snmp.udp_transport_retries)
def _get_context(self):
"""Return the SNMP context for an SNMP request.
:returns: A :class:
`pysnmp.hlapi.ContextData` object.
:raises: :class:`pysnmp.error.PySnmpError` if SNMP context data
is bad.
"""
return snmp.ContextData(
contextEngineId=self.context_engine_id,
contextName=self.context_name
)
def get(self, oid):
"""Use PySNMP to perform an SNMP GET operation on a single object.
@ -271,13 +283,16 @@ class SNMPClient(object):
:returns: The value of the requested object.
"""
try:
results = self.cmd_gen.getCmd(self._get_auth(),
self._get_transport(),
oid)
snmp_gen = snmp.getCmd(self.snmp_engine,
self._get_auth(),
self._get_transport(),
self._get_context(),
snmp.ObjectType(snmp.ObjectIdentity(oid)))
except snmp_error.PySnmpError as e:
raise exception.SNMPFailure(operation="GET", error=e)
error_indication, error_status, error_index, var_binds = results
error_indication, error_status, error_index, var_binds = next(snmp_gen)
if error_indication:
# SNMP engine-level error.
@ -301,13 +316,17 @@ class SNMPClient(object):
:returns: A list of values of the requested table object.
"""
try:
results = self.cmd_gen.nextCmd(self._get_auth(),
self._get_transport(),
oid)
snmp_gen = snmp.nextCmd(self.snmp_engine,
self._get_auth(),
self._get_transport(),
self._get_context(),
snmp.ObjectType(snmp.ObjectIdentity(oid)))
except snmp_error.PySnmpError as e:
raise exception.SNMPFailure(operation="GET_NEXT", error=e)
error_indication, error_status, error_index, var_bind_table = results
(error_indication, error_status, error_index,
var_bind_table) = next(snmp_gen)
if error_indication:
# SNMP engine-level error.
@ -329,13 +348,17 @@ class SNMPClient(object):
:raises: SNMPFailure if an SNMP request fails.
"""
try:
results = self.cmd_gen.setCmd(self._get_auth(write_mode=True),
self._get_transport(),
(oid, value))
snmp_gen = snmp.setCmd(self.snmp_engine,
self._get_auth(write_mode=True),
self._get_transport(),
self._get_context(),
snmp.ObjectType(
snmp.ObjectIdentity(oid), value))
except snmp_error.PySnmpError as e:
raise exception.SNMPFailure(operation="SET", error=e)
error_indication, error_status, error_index, var_binds = results
error_indication, error_status, error_index, var_binds = next(snmp_gen)
if error_indication:
# SNMP engine-level error.
@ -565,11 +588,11 @@ class SNMPDriverSimple(SNMPDriverBase):
return power_state
def _snmp_power_on(self):
value = rfc1902.Integer(self.value_power_on)
value = snmp.Integer(self.value_power_on)
self.client.set(self.oid, value)
def _snmp_power_off(self):
value = rfc1902.Integer(self.value_power_off)
value = snmp.Integer(self.value_power_off)
self.client.set(self.oid, value)
@ -739,12 +762,12 @@ class SNMPDriverEatonPower(SNMPDriverBase):
def _snmp_power_on(self):
oid = self._snmp_oid(self.oid_poweron)
value = rfc1902.Integer(self.value_power_on)
value = snmp.Integer(self.value_power_on)
self.client.set(oid, value)
def _snmp_power_off(self):
oid = self._snmp_oid(self.oid_poweroff)
value = rfc1902.Integer(self.value_power_off)
value = snmp.Integer(self.value_power_off)
self.client.set(oid, value)

174
ironic/tests/unit/drivers/modules/test_snmp.py
View File

@ -20,8 +20,8 @@ import time
import mock
from oslo_config import cfg
from pysnmp.entity.rfc3413.oneliner import cmdgen
from pysnmp import error as snmp_error
from pysnmp import hlapi as pysnmp
from ironic.common import exception
from ironic.common import states
@ -37,210 +37,218 @@ CONF = cfg.CONF
INFO_DICT = db_utils.get_test_snmp_info()
@mock.patch.object(cmdgen, 'CommandGenerator', autospec=True)
class SNMPClientTestCase(base.TestCase):
def setUp(self):
super(SNMPClientTestCase, self).setUp()
self.address = '1.2.3.4'
self.port = '6700'
self.oid = 'oid'
self.oid = (1, 3, 6, 1, 1, 1, 0)
self.value = 'value'
def test___init__(self, mock_cmdgen):
@mock.patch.object(pysnmp, 'SnmpEngine', autospec=True)
def test___init__(self, mock_snmpengine):
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V1)
mock_cmdgen.assert_called_once_with()
mock_snmpengine.assert_called_once_with()
self.assertEqual(self.address, client.address)
self.assertEqual(self.port, client.port)
self.assertEqual(snmp.SNMP_V1, client.version)
self.assertIsNone(client.read_community)
self.assertIsNone(client.write_community)
self.assertNotIn('user', client.__dict__)
self.assertEqual(mock_cmdgen.return_value, client.cmd_gen)
self.assertEqual(mock_snmpengine.return_value, client.snmp_engine)
@mock.patch.object(cmdgen, 'CommunityData', autospec=True)
def test__get_auth_v1_read(self, mock_community, mock_cmdgen):
@mock.patch.object(pysnmp, 'CommunityData', autospec=True)
def test__get_auth_v1_read(self, mock_community):
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V1,
read_community='public',
write_community='private')
client._get_auth()
mock_cmdgen.assert_called_once_with()
mock_community.assert_called_once_with(client.read_community,
mpModel=0)
@mock.patch.object(cmdgen, 'CommunityData', autospec=True)
def test__get_auth_v1_write(self, mock_community, mock_cmdgen):
@mock.patch.object(pysnmp, 'CommunityData', autospec=True)
def test__get_auth_v1_write(self, mock_community):
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V1,
read_community='public',
write_community='private')
client._get_auth(write_mode=True)
mock_cmdgen.assert_called_once_with()
mock_community.assert_called_once_with(client.write_community,
mpModel=0)
@mock.patch.object(cmdgen, 'UsmUserData', autospec=True)
def test__get_auth_v3(self, mock_user, mock_cmdgen):
@mock.patch.object(pysnmp, 'UsmUserData', autospec=True)
def test__get_auth_v3(self, mock_user):
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
client._get_auth()
mock_cmdgen.assert_called_once_with()
mock_user.assert_called_once_with(
client.user,
authKey=client.auth_key,
authProtocol=client.auth_proto,
privKey=client.priv_key,
privProtocol=client.priv_proto,
contextEngineId=client.context_engine_id,
contextName=client.context_name
)
@mock.patch.object(cmdgen, 'UdpTransportTarget', autospec=True)
def test__get_transport(self, mock_transport, mock_cmdgen):
@mock.patch.object(pysnmp, 'ContextData', autospec=True)
def test__get_context(self, mock_context):
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V1)
client._get_context()
mock_context.assert_called_once_with(None, '')
@mock.patch.object(pysnmp, 'UdpTransportTarget', autospec=True)
def test__get_transport(self, mock_transport):
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
client._get_transport()
mock_cmdgen.assert_called_once_with()
mock_transport.assert_called_once_with(
(client.address, client.port),
retries=CONF.snmp.udp_transport_retries,
timeout=CONF.snmp.udp_transport_timeout)
@mock.patch.object(cmdgen, 'UdpTransportTarget', autospec=True)
def test__get_transport_err(self, mock_transport, mock_cmdgen):
@mock.patch.object(pysnmp, 'UdpTransportTarget', autospec=True)
def test__get_transport_err(self, mock_transport):
mock_transport.side_effect = snmp_error.PySnmpError
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
self.assertRaises(snmp_error.PySnmpError, client._get_transport)
mock_cmdgen.assert_called_once_with()
mock_transport.assert_called_once_with(
(client.address, client.port),
retries=CONF.snmp.udp_transport_retries,
timeout=CONF.snmp.udp_transport_timeout)
@mock.patch.object(cmdgen, 'UdpTransportTarget', autospec=True)
def test__get_transport_custom_timeout(self, mock_transport, mock_cmdgen):
@mock.patch.object(pysnmp, 'UdpTransportTarget', autospec=True)
def test__get_transport_custom_timeout(self, mock_transport):
self.config(udp_transport_timeout=2.0, group='snmp')
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
client._get_transport()
mock_cmdgen.assert_called_once_with()
mock_transport.assert_called_once_with((client.address, client.port),
retries=5, timeout=2.0)
@mock.patch.object(cmdgen, 'UdpTransportTarget', autospec=True)
def test__get_transport_custom_retries(self, mock_transport, mock_cmdgen):
@mock.patch.object(pysnmp, 'UdpTransportTarget', autospec=True)
def test__get_transport_custom_retries(self, mock_transport):
self.config(udp_transport_retries=10, group='snmp')
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
client._get_transport()
mock_cmdgen.assert_called_once_with()
mock_transport.assert_called_once_with((client.address, client.port),
retries=10, timeout=1.0)
@mock.patch.object(pysnmp, 'getCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_get(self, mock_auth, mock_transport, mock_cmdgen):
def test_get(self, mock_auth, mock_context, mock_transport, mock_getcmd):
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.getCmd.return_value = ("", None, 0, [var_bind])
mock_getcmd.return_value = iter([("", None, 0, [var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
val = client.get(self.oid)
self.assertEqual(var_bind[1], val)
mock_cmdgenerator.getCmd.assert_called_once_with(mock.ANY, mock.ANY,
self.oid)
self.assertEqual(1, mock_getcmd.call_count)
@mock.patch.object(pysnmp, 'nextCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_get_next(self, mock_auth, mock_transport, mock_cmdgen):
def test_get_next(self, mock_auth, mock_context, mock_transport,
mock_nextcmd):
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.nextCmd.return_value = (
"", None, 0, [[var_bind, var_bind]])
mock_nextcmd.return_value = iter([("", None, 0,
[[var_bind, var_bind]])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
val = client.get_next(self.oid)
self.assertEqual([self.value, self.value], val)
mock_cmdgenerator.nextCmd.assert_called_once_with(mock.ANY, mock.ANY,
self.oid)
self.assertEqual(1, mock_nextcmd.call_count)
@mock.patch.object(pysnmp, 'getCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_get_err_transport(self, mock_auth, mock_transport, mock_cmdgen):
def test_get_err_transport(self, mock_auth, mock_context, mock_transport,
mock_getcmd):
mock_transport.side_effect = snmp_error.PySnmpError
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.getCmd.return_value = ("engine error", None, 0,
[var_bind])
mock_getcmd.return_value = iter([("engine error", None,
0, [var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
self.assertRaises(exception.SNMPFailure, client.get, self.oid)
self.assertFalse(mock_cmdgenerator.getCmd.called)
self.assertFalse(mock_getcmd.called)
@mock.patch.object(pysnmp, 'nextCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_get_next_err_transport(self, mock_auth, mock_transport,
mock_cmdgen):
def test_get_next_err_transport(self, mock_auth, mock_context,
mock_transport, mock_nextcmd):
mock_transport.side_effect = snmp_error.PySnmpError
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.nextCmd.return_value = ("engine error", None, 0,
[[var_bind, var_bind]])
mock_nextcmd.return_value = iter([("engine error", None, 0,
[var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
self.assertRaises(exception.SNMPFailure, client.get_next, self.oid)
self.assertFalse(mock_cmdgenerator.nextCmd.called)
self.assertFalse(mock_nextcmd.called)
@mock.patch.object(pysnmp, 'getCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_get_err_engine(self, mock_auth, mock_transport, mock_cmdgen):
def test_get_err_engine(self, mock_auth, mock_context, mock_transport,
mock_getcmd):
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.getCmd.return_value = ("engine error", None, 0,
[var_bind])
mock_getcmd.return_value = iter([("engine error", None, 0,
[var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
self.assertRaises(exception.SNMPFailure, client.get, self.oid)
mock_cmdgenerator.getCmd.assert_called_once_with(mock.ANY, mock.ANY,
self.oid)
self.assertEqual(1, mock_getcmd.call_count)
@mock.patch.object(pysnmp, 'nextCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_get_next_err_engine(self, mock_auth, mock_transport, mock_cmdgen):
def test_get_next_err_engine(self, mock_auth, mock_context,
mock_transport, mock_nextcmd):
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.nextCmd.return_value = ("engine error", None, 0,
[[var_bind, var_bind]])
mock_nextcmd.return_value = iter([("engine error", None, 0,
[var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
self.assertRaises(exception.SNMPFailure, client.get_next, self.oid)
mock_cmdgenerator.nextCmd.assert_called_once_with(mock.ANY, mock.ANY,
self.oid)
self.assertEqual(1, mock_nextcmd.call_count)
@mock.patch.object(pysnmp, 'setCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_set(self, mock_auth, mock_transport, mock_cmdgen):
def test_set(self, mock_auth, mock_context, mock_transport,
mock_setcmd):
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.setCmd.return_value = ("", None, 0, [var_bind])
mock_setcmd.return_value = iter([("", None, 0,
[var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
client.set(self.oid, self.value)
mock_cmdgenerator.setCmd.assert_called_once_with(mock.ANY, mock.ANY,
var_bind)
self.assertEqual(1, mock_setcmd.call_count)
@mock.patch.object(pysnmp, 'setCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_set_err_transport(self, mock_auth, mock_transport, mock_cmdgen):
def test_set_err_transport(self, mock_auth, mock_context, mock_transport,
mock_setcmd):
mock_transport.side_effect = snmp_error.PySnmpError
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.setCmd.return_value = ("engine error", None, 0,
[var_bind])
mock_setcmd.return_value = iter([("engine error", None, 0,
[var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
self.assertRaises(exception.SNMPFailure,
client.set, self.oid, self.value)
self.assertFalse(mock_cmdgenerator.setCmd.called)
self.assertRaises(exception.SNMPFailure, client.set, self.oid,
self.value)
self.assertFalse(mock_setcmd.called)
@mock.patch.object(pysnmp, 'setCmd', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_transport', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_context', autospec=True)
@mock.patch.object(snmp.SNMPClient, '_get_auth', autospec=True)
def test_set_err_engine(self, mock_auth, mock_transport, mock_cmdgen):
def test_set_err_engine(self, mock_auth, mock_context, mock_transport,
mock_setcmd):
var_bind = (self.oid, self.value)
mock_cmdgenerator = mock_cmdgen.return_value
mock_cmdgenerator.setCmd.return_value = ("engine error", None, 0,
[var_bind])
mock_setcmd.return_value = iter([("engine error", None, 0,
[var_bind])])
client = snmp.SNMPClient(self.address, self.port, snmp.SNMP_V3)
self.assertRaises(exception.SNMPFailure,
client.set, self.oid, self.value)
mock_cmdgenerator.setCmd.assert_called_once_with(mock.ANY, mock.ANY,
var_bind)
self.assertRaises(exception.SNMPFailure, client.set, self.oid,
self.value)
self.assertEqual(1, mock_setcmd.call_count)
class SNMPValidateParametersTestCase(db_base.DbTestCase):

3
ironic/tests/unit/drivers/third_party_driver_mock_specs.py
View File

@ -71,9 +71,8 @@ PYWSMAN_SPEC = (
# pywsnmp
PYWSNMP_SPEC = (
'entity',
'hlapi',
'error',
'proto',
)
# scciclient

12
ironic/tests/unit/drivers/third_party_driver_mocks.py
View File

@ -117,24 +117,18 @@ if not dracclient:
if 'ironic.drivers.modules.drac' in sys.modules:
six.moves.reload_module(sys.modules['ironic.drivers.modules.drac'])
# attempt to load the external 'pysnmp' library, which is required by
# the optional drivers.modules.snmp module
pysnmp = importutils.try_import("pysnmp")
if not pysnmp:
pysnmp = mock.MagicMock(spec_set=mock_specs.PYWSNMP_SPEC)
sys.modules["pysnmp"] = pysnmp
sys.modules["pysnmp.entity"] = pysnmp.entity
sys.modules["pysnmp.entity.rfc3413"] = pysnmp.entity.rfc3413
sys.modules["pysnmp.entity.rfc3413.oneliner"] = (
pysnmp.entity.rfc3413.oneliner)
sys.modules["pysnmp.entity.rfc3413.oneliner.cmdgen"] = (
pysnmp.entity.rfc3413.oneliner.cmdgen)
sys.modules["pysnmp.hlapi"] = pysnmp.hlapi
sys.modules["pysnmp.error"] = pysnmp.error
pysnmp.error.PySnmpError = Exception
sys.modules["pysnmp.proto"] = pysnmp.proto
sys.modules["pysnmp.proto.rfc1902"] = pysnmp.proto.rfc1902
# Patch the RFC1902 integer class with a python int
pysnmp.proto.rfc1902.Integer = int
pysnmp.hlapi.Integer = int
# if anything has loaded the snmp driver yet, reload it now that the

5
releasenotes/notes/migrate-to-pysnmp-hlapi-477075b5e69cc5bc.yaml Normal file
View File

@ -0,0 +1,5 @@
---
upgrade:
- The minimum required version of pysnmp has been bumped to 4.3. This
pysnmp version introduces simpler, faster and more functional high-level
SNMP API on which ironic `snmp` driver has been migrated.