Replace more string with types.

makefile

@@ -25,11 +25,14 @@ clean:
 	rm -rf *.pyc dist build pynetsnmp.egg-info
 
 .PHONY: test
-HOST ?= 127.0.0.1
 test:
-	docker run --rm -v $(PWD):/mnt -w /mnt $(TAG) \
-    bash -c "python setup.py bdist_wheel \
-    && pip install dist/pynetsnmp*py2-none-any.whl ipaddr Twisted==20.3.0 \
-    && cd test \
-    && python test_runner.py --host $(HOST) \
-    && chown -R $(UID):$(GID) /mnt" ;
+	@$(DOCKER_COMMAND) bash -c "pip --no-python-version-warning install -q .; cd tests; python -m unittest discover"
+
+# HOST ?= 127.0.0.1
+# test:
+# 	docker run --rm -v $(PWD):/mnt -w /mnt $(TAG) \
+#     bash -c "python setup.py bdist_wheel \
+#     && pip install dist/pynetsnmp*py2-none-any.whl ipaddr Twisted==20.3.0 \
+#     && cd test \
+#     && python test_runner.py --host $(HOST) \
+#     && chown -R $(UID):$(GID) /mnt" ;

pynetsnmp/errors.py

@@ -0,0 +1,61 @@
+from __future__ import absolute_import
+
+from . import oids
+
+
+class SnmpTimeoutError(Exception):
+    pass
+
+
+class ArgumentParseError(Exception):
+    pass
+
+
+class TransportError(Exception):
+    pass
+
+
+class SnmpNameError(Exception):
+    def __init__(self, oid):
+        Exception.__init__(self, "Bad Name", oid)
+
+
+class SnmpError(Exception):
+    def __init__(self, message, *args, **kwargs):
+        self.message = message
+
+    def __str__(self):
+        return self.message
+
+    def __repr__(self):
+        return self.message
+
+
+class SnmpUsmError(SnmpError):
+    pass
+
+
+class SnmpUsmStatsError(SnmpUsmError):
+    def __init__(self, mesg, oid):
+        super(SnmpUsmStatsError, self).__init__(mesg)
+        self.oid = oid
+
+
+_stats_oid_error_map = {
+    oids.WrongDigest: SnmpUsmStatsError(
+        "unexpected authentication digest", oids.WrongDigest
+    ),
+    oids.UnknownUserName: SnmpUsmStatsError(
+        "unknown user", oids.UnknownUserName
+    ),
+    oids.UnknownSecurityLevel: SnmpUsmStatsError(
+        "unknown or unavailable security level", oids.UnknownSecurityLevel
+    ),
+    oids.DecryptionError: SnmpUsmStatsError(
+        "privacy decryption error", oids.DecryptionError
+    ),
+}
+
+
+def get_stats_error(oid):
+    return _stats_oid_error_map.get(oid)

pynetsnmp/netsnmp.py

@@ -87,6 +87,7 @@
     USM_AUTH_KU_LEN,
     USM_PRIV_KU_LEN,
 )
+from .errors import ArgumentParseError, SnmpTimeoutError
 
 
 def _getLogger(name):
@@ -304,7 +305,7 @@ class netsnmp_container_s(Structure):
         ("securityAuthLocalKeyLen", c_size_t),
         ("securityPrivProto", POINTER(oid)),
         ("securityPrivProtoLen", c_size_t),
-        ("securityPrivKey", c_char * USM_PRIV_KU_LEN),
+        ("securityPrivKey", u_char * USM_PRIV_KU_LEN),
         ("securityPrivKeyLen", c_size_t),
         ("securityPrivLocalKey", c_char_p),
         ("securityPrivLocalKeyLen", c_size_t),
@@ -638,16 +639,12 @@ def getResult(pdu, log):
     return result
 
 
-class SnmpError(Exception):
+class NetSnmpError(Exception):
     def __init__(self, why):
         lib.snmp_perror(why)
         Exception.__init__(self, why)
 
 
-class SnmpTimeoutError(Exception):
-    pass
-
-
 sessionMap = {}
 
 
@@ -676,14 +673,6 @@ def _callback(operation, sp, reqid, pdu, magic):
 _callback = netsnmp_callback(_callback)
 
 
-class ArgumentParseError(Exception):
-    pass
-
-
-class TransportError(Exception):
-    pass
-
-
 def _doNothingProc(argc, argv, arg):
     return 0
 
@@ -807,7 +796,7 @@ def open(self):
         ref = byref(sess)
         self.sess = lib.snmp_open(ref)
         if not self.sess:
-            raise SnmpError("snmp_open")
+            raise NetSnmpError("snmp_open")
 
     def awaitTraps(
         self, peername, fileno=-1, pre_parse_callback=None, debug=False
@@ -834,7 +823,7 @@ def awaitTraps(
         lib.setup_engineID(None, None)
         transport = lib.netsnmp_tdomain_transport(peername, 1, "udp")
         if not transport:
-            raise SnmpError(
+            raise NetSnmpError(
                 "Unable to create transport {peername}".format(
                     peername=peername
                 )
@@ -861,7 +850,7 @@ def awaitTraps(
         sess.isAuthoritative = SNMP_SESS_UNKNOWNAUTH
         rc = lib.snmp_add(self.sess, transport, pre_parse_callback, None)
         if not rc:
-            raise SnmpError("snmp_add")
+            raise NetSnmpError("snmp_add")
 
     def create_users(self, users):
         self._log.debug("create_users: Creating %s users.", len(users))
@@ -991,7 +980,7 @@ def _handle_send_status(self, req, send_status, send_type):
             lib.snmp_free_pdu(req)
             if snmperr.value == SNMPERR_TIMEOUT:
                 raise SnmpTimeoutError()
-            raise SnmpError(msg_fmt % msg_args)
+            raise NetSnmpError(msg_fmt % msg_args)
 
     def get(self, oids):
         req = self._create_request(SNMP_MSG_GET)

pynetsnmp/oids.py

@@ -0,0 +1,94 @@
+from __future__ import absolute_import
+
+from .conversions import asOidStr
+
+
+class OID(object):
+    __slots__ = ("oid",)
+
+    def __init__(self, oid):
+        super(OID, self).__setattr__("oid", oid)
+
+    def __setattr__(self, key, value):
+        if key in OID.__slots__:
+            raise AttributeError(
+                "can't set attribute '{}' on 'OID' object".format(key)
+            )
+        super(OID, self).__setattr__(key, value)
+
+    def __eq__(this, that):
+        if isinstance(that, (tuple, list)):
+            return this.oid == that
+        if isinstance(that, OID):
+            return this.oid == that.oid
+        return NotImplemented
+
+    def __ne__(this, that):
+        if isinstance(that, (tuple, list)):
+            return this.oid != that
+        if isinstance(that, OID):
+            return this.oid != that.oid
+        return NotImplemented
+
+    def __hash__(self):
+        return hash(self.oid)
+
+    def __repr__(self):
+        return "<{0.__module__}.{0.__class__.__name__} {1}>".format(
+            self, asOidStr(self.oid)
+        )
+
+    def __str__(self):
+        return asOidStr(self.oid)
+
+
+_base_status_oid = (1, 3, 6, 1, 6, 3, 15, 1, 1)
+
+
+class UnknownSecurityLevel(OID):
+    __slots__ = ()
+
+
+UnknownSecurityLevel = UnknownSecurityLevel(_base_status_oid + (1, 0))
+
+
+class NotInTimeWindow(OID):
+    __slots__ = ()
+
+
+NotInTimeWindow = NotInTimeWindow(_base_status_oid + (2, 0))
+
+
+class UnknownUserName(OID):
+    __slots__ = ()
+
+
+UnknownUserName = UnknownUserName(_base_status_oid + (3, 0))
+
+
+class UnknownEngineId(OID):
+    __slots__ = ()
+
+
+UnknownEngineId = UnknownEngineId(_base_status_oid + (4, 0))
+
+
+class WrongDigest(OID):
+    __slots__ = ()
+
+
+WrongDigest = WrongDigest(_base_status_oid + (5, 0))
+
+
+class DecryptionError(OID):
+    __slots__ = ()
+
+
+DecryptionError = DecryptionError(_base_status_oid + (6, 0))
+
+
+class SysDescr(OID):
+    __slots__ = ()
+
+
+SysDescr = SysDescr((1, 3, 6, 1, 2, 1, 1, 1, 0))