openstack
/
ironic
Code Issues Proposed changes

Migrate ironic `snmp` driver to the latest pysnmp API 

Since pysnmp 4.3 (which dates back to 2015), library's high-level
SNMP API has been reworked towards better consistency and performance.
Ironic snmp driver still uses the older pysnmp API. This story
suggests migrating ironic snmp driver to the latest (e.g.
post 4.3) pysnmp API.

This bumps up ironic lower constraint on pysnmp.

Change-Id: Id3a03210e8a52dda5e6403e2f9cb040c59ca9573
Story: 2002756
Task: 22612
This commit is contained in:
Ilya Etingof 2018-06-28 22:07:28 +02:00
parent 607070569c
commit 0d03136bcc
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.
@ -533,11 +556,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)
@ -707,12 +730,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
@ -36,210 +36,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.