Add support for configuring PoE for cisco equipment

python/nav/portadmin/handlers.py

@@ -352,3 +352,7 @@ class POEStateNotSupportedError(ManagementError):
 
 class XMLParseError(ManagementError):
  """Raised when failing to parse XML"""
+
+
+class POEIndexNotFoundError(ManagementError):
+ """Raised when a PoE index could not be found for an interface"""

python/nav/portadmin/snmp/cisco.py

@@ -16,13 +16,21 @@
 #
 """Cisco specific PortAdmin SNMP handling"""
 import logging
+from typing import Sequence, Dict, Optional, Tuple
 
 from nav.Snmp.errors import SnmpError
 from nav.bitvector import BitVector
 from nav.oids import OID
 from nav.portadmin.snmp.base import SNMPHandler, translate_protocol_errors
 from nav.smidumps import get_mib
 from nav.enterprise.ids import VENDOR_ID_CISCOSYSTEMS
+from nav.portadmin.handlers import (
+ PoeState,
+ POEStateNotSupportedError,
+ POENotSupportedError,
+ POEIndexNotFoundError,
+)
+from nav.models import manage
 
 
 _logger = logging.getLogger(__name__)
@@ -35,6 +43,7 @@
 
  VTPNODES = get_mib('CISCO-VTP-MIB')['nodes']
  PAENODES = get_mib('CISCO-PAE-MIB')['nodes']
+ POENODES = get_mib('CISCO-POWER-ETHERNET-EXT-MIB')['nodes']
 
  VTPVLANSTATE = VTPNODES['vtpVlanState']['oid']
  VTPVLANTYPE = VTPNODES['vtpVlanType']['oid']
@@ -57,6 +66,19 @@
  DOT1X_AUTHENTICATOR = 0b10000000
  DOT1X_SUPPLICANT = 0b01000000
 
+ POEENABLE = POENODES['cpeExtPsePortEnable']['oid']
+ POE_AUTO = PoeState(state=1, name="AUTO")
+ POE_STATIC = PoeState(state=2, name="STATIC")
+ POE_LIMIT = PoeState(state=3, name="LIMIT")
+ POE_DISABLE = PoeState(state=4, name="DISABLE")
+
+ POE_OPTIONS = [
+ POE_AUTO,
+ POE_STATIC,
+ POE_LIMIT,
+ POE_DISABLE,
+ ]
+
  def __init__(self, netbox, **kwargs):
  super(Cisco, self).__init__(netbox, **kwargs)
  self.vlan_oid = '1.3.6.1.4.1.9.9.68.1.2.2.1.2'
@@ -300,5 +322,82 @@
  for oid, state in self._bulkwalk(self.dot1xPortAuth)
  }
 
+ def get_poe_state_options(self) -> Sequence[PoeState]:
+ """Returns the available options for enabling/disabling PoE on this netbox"""
+ return self.POE_OPTIONS
+
+ @translate_protocol_errors
+ def set_poe_state(self, interface: manage.Interface, state: PoeState):
+ """Set state for enabling/disabling PoE on this interface.
+ Available options should be retrieved using `get_poe_state_options`
+ """
+ unit_number, interface_number = self._get_poe_indexes_for_interface(interface)
+ oid_with_unit_number = self.POEENABLE + OID((unit_number,))
+ try:
+ self._set_netbox_value(
+ oid_with_unit_number, interface_number, 'i', state.state
+ )
+ except SnmpError as error:
+ _logger.error('Error setting poe state: %s', error)
+ raise
+ except ValueError as error:
+ _logger.error('%s is not a valid option for poe state', state)
+ raise
+
+ def _get_poe_indexes_for_interface(
+ self, interface: manage.Interface
+ ) -> Tuple[int, int]:
+ """Returns the unit number and interface number for the given interface"""
+ try:
+ poeport = manage.POEPort.objects.get(interface=interface)
+ except manage.POEPort.DoesNotExist:
+ raise POEIndexNotFoundError(
+ "This interface does not have PoE indexes defined"
+ )
+ unit_number = poeport.poegroup.index
+ interface_number = poeport.index
+ return unit_number, interface_number
+
+ def get_poe_states(
+ self, interfaces: Optional[Sequence[manage.Interface]] = None
+ ) -> Dict[str, Optional[PoeState]]:
+ """Retrieves current PoE state for interfaces on this device.
+
+ :param interfaces: Optional sequence of interfaces to filter for, as fetching
+ data for all interfaces may be a waste of time if only a
+ single interface is needed. If this parameter is omitted,
+ the default behavior is to filter on all Interface objects
+ registered for this device.
+ :returns: A dict mapping interfaces to their discovered PoE state.
+ The key matches the `ifname` attribute for the related
+ Interface object.
+ The value will be None if the interface does not support PoE.
+ """
+ if not interfaces:
+ interfaces = self.netbox.interfaces
+ states_dict = {}
+ for interface in interfaces:
+ try:
+ state = self._get_poe_state_for_single_interface(interface)
+ except POENotSupportedError:
+ state = None
+ states_dict[interface.ifname] = state
+ return states_dict
+
+ @translate_protocol_errors
+ def _get_poe_state_for_single_interface(
+ self, interface: manage.Interface
+ ) -> PoeState:
+ """Retrieves current PoE state for given the given interface"""
+ unit_number, interface_number = self._get_poe_indexes_for_interface(interface)
+ oid_with_unit_number = self.POEENABLE + OID((unit_number,))
+ state_value = self._query_netbox(oid_with_unit_number, interface_number)
+ if state_value == None:
+ raise POENotSupportedError("This interface does not support PoE")
+ for state in self.get_poe_state_options():
+ if state.state == state_value:
+ return state
+ raise POEStateNotSupportedError(f"Unknown PoE state {state_value}")
+
 
 CHARS_IN_1024_BITS = 128

tests/unittests/portadmin/conftest.py

@@ -0,0 +1,58 @@
+from mock import Mock
+
+import pytest
+
+from nav.enterprise.ids import VENDOR_ID_HEWLETT_PACKARD, VENDOR_ID_CISCOSYSTEMS
+from nav.portadmin.management import ManagementFactory
+
+
+@pytest.fixture
+def profile():
+ profile = Mock()
+ profile.snmp_version = 2
+ profile.snmp_community = "public"
+ return profile
+
+
+@pytest.fixture
+def netbox_hp(profile):
+ vendor = Mock()
+ vendor.id = u'hp'
+
+ netbox_type = Mock()
+ netbox_type.vendor = vendor
+ netbox_type.get_enterprise_id.return_value = VENDOR_ID_HEWLETT_PACKARD
+
+ netbox = Mock()
+ netbox.type = netbox_type
+ netbox.ip = '10.240.160.39'
+ netbox.get_preferred_snmp_management_profile.return_value = profile
+
+ return netbox
+
+
+@pytest.fixture
+def netbox_cisco(profile):
+ vendor = Mock()
+ vendor.id = u'cisco'
+
+ netbox_type = Mock()
+ netbox_type.vendor = vendor
+ netbox_type.get_enterprise_id.return_value = VENDOR_ID_CISCOSYSTEMS
+
+ netbox = Mock()
+ netbox.type = netbox_type
+ netbox.ip = '10.240.160.38'
+ netbox.get_preferred_snmp_management_profile.return_value = profile
+
+ return netbox
+
+
+@pytest.fixture
+def handler_hp(netbox_hp):
+ return ManagementFactory.get_instance(netbox_hp)
+
+
+@pytest.fixture
+def handler_cisco(netbox_cisco):
+ return ManagementFactory.get_instance(netbox_cisco)

tests/unittests/portadmin/portadmin_poe_cisco_test.py

@@ -0,0 +1,78 @@
+from mock import Mock, patch
+
+import pytest
+
+from nav.portadmin.snmp.cisco import Cisco
+from nav.portadmin.handlers import (
+ POEIndexNotFoundError,
+ POEStateNotSupportedError,
+)
+from nav.models import manage
+
+
+class TestGetPoeStateOptions:
+ def test_returns_correct_options(self, handler_cisco):
+ state_options = handler_cisco.get_poe_state_options()
+ assert Cisco.POE_AUTO in state_options
+ assert Cisco.POE_STATIC in state_options
+ assert Cisco.POE_LIMIT in state_options
+ assert Cisco.POE_DISABLE in state_options
+
+
+class TestGetPoeState:
+ @pytest.mark.usefixtures('poeport_get_mock')
+ def test_should_raise_exception_if_unknown_poe_state(self, handler_cisco):
+ handler_cisco._query_netbox = Mock(return_value=76)
+ interface = Mock(interface="interface")
+ with pytest.raises(POEStateNotSupportedError):
+ handler_cisco.get_poe_states([interface])
+
+ @pytest.mark.usefixtures('poeport_get_mock_error')
+ def test_should_raise_exception_if_interface_is_missing_poeport(
+ self, handler_cisco
+ ):
+ interface = Mock(interface="interface")
+ with pytest.raises(POEIndexNotFoundError):
+ handler_cisco.get_poe_states([interface])
+
+ @pytest.mark.usefixtures('poeport_get_mock')
+ def test_dict_should_give_none_if_interface_does_not_support_poe(
+ self, handler_cisco
+ ):
+ handler_cisco._query_netbox = Mock(return_value=None)
+ interface = Mock(interface="interface")
+ states = handler_cisco.get_poe_states([interface])
+ assert states[interface.ifname] is None
+
+ @pytest.mark.usefixtures('poeport_get_mock')
+ def test_returns_correct_poe_state(self, handler_cisco):
+ expected_state = Cisco.POE_AUTO
+ handler_cisco._query_netbox = Mock(return_value=expected_state.state)
+ interface = Mock(ifname="interface")
+ state = handler_cisco.get_poe_states([interface])
+ assert state[interface.ifname] == expected_state
+
+ @pytest.mark.usefixtures('poeport_get_mock')
+ def test_use_interfaces_from_db_if_empty_interfaces_arg(self, handler_cisco):
+ expected_state = Cisco.POE_AUTO
+ handler_cisco._query_netbox = Mock(return_value=expected_state.state)
+ interface = Mock(ifname="interface")
+ handler_cisco.netbox.interfaces = [interface]
+ state = handler_cisco.get_poe_states()
+ assert interface.ifname in state
+
+
+@pytest.fixture()
+def poeport_get_mock():
+ with patch("nav.portadmin.snmp.cisco.manage.POEPort.objects.get") as get_mock:
+ poegroup_mock = Mock(index=1)
+ poeport_mock = Mock(poegroup=poegroup_mock, index=1)
+ get_mock.return_value = poeport_mock
+ yield get_mock
+
+
+@pytest.fixture()
+def poeport_get_mock_error():
+ with patch("nav.portadmin.snmp.cisco.manage.POEPort.objects.get") as get_mock:
+ get_mock.side_effect = manage.POEPort.DoesNotExist
+ yield get_mock

tests/unittests/portadmin/portadmin_test.py

@@ -1,9 +1,6 @@
 from mock import Mock
 
-import pytest
-
 from nav.oids import OID
-from nav.enterprise.ids import VENDOR_ID_HEWLETT_PACKARD, VENDOR_ID_CISCOSYSTEMS
 from nav.portadmin.management import ManagementFactory
 from nav.portadmin.snmp.hp import HP
 from nav.portadmin.snmp.cisco import Cisco
@@ -77,55 +74,3 @@ def test_get_ifaliases_cisco(self, handler_cisco):
  expected = {1: 'jomar', 2: 'knut', 3: 'hjallis'}
  snmp_read_only_handler.bulkwalk = Mock(return_value=walkdata)
  assert handler_cisco._get_all_ifaliases() == expected, "getAllIfAlias failed."
-
-
-@pytest.fixture
-def profile():
- profile = Mock()
- profile.snmp_version = 2
- profile.snmp_community = "public"
- return profile
-
-
-@pytest.fixture
-def netbox_hp(profile):
- vendor = Mock()
- vendor.id = u'hp'
-
- netbox_type = Mock()
- netbox_type.vendor = vendor
- netbox_type.get_enterprise_id.return_value = VENDOR_ID_HEWLETT_PACKARD
-
- netbox = Mock()
- netbox.type = netbox_type
- netbox.ip = '10.240.160.39'
- netbox.get_preferred_snmp_management_profile.return_value = profile
-
- return netbox
-
-
-@pytest.fixture
-def netbox_cisco(profile):
- vendor = Mock()
- vendor.id = u'cisco'
-
- netbox_type = Mock()
- netbox_type.vendor = vendor
- netbox_type.get_enterprise_id.return_value = VENDOR_ID_CISCOSYSTEMS
-
- netbox = Mock()
- netbox.type = netbox_type
- netbox.ip = '10.240.160.38'
- netbox.get_preferred_snmp_management_profile.return_value = profile
-
- return netbox
-
-
-@pytest.fixture
-def handler_hp(netbox_hp):
- return ManagementFactory.get_instance(netbox_hp)
-
-
-@pytest.fixture
-def handler_cisco(netbox_cisco):
- return ManagementFactory.get_instance(netbox_cisco)