snmpclient.py 文件源码

def set(self, *oidvalues):
        assert self.alive is True
        oidvalues_trans = []
        for oid, value in oidvalues:
            if isinstance(oid, tuple):
                has_str = False
                for entry in oid:
                    if isinstance(entry, str):
                        has_str = True
                        break
                if has_str:  # if oid is a tuple containing strings, assume translation using cmdgen.MibVariable.
                    # value must then be a Python type
                    assert isinstance(value, int) or isinstance(value, str) or isinstance(value, bool)
                    oidvalues_trans.append((cmdgen.MibVariable(*oid), value))
                else:
                    # value must be a rfc1902/pyasn1 type
                    if not oid[-1] == 0:
                        assert isinstance(value, univ.Integer) or isinstance(value, univ.OctetString) or isinstance(value, univ.ObjectIdentifier)
                    oidvalues_trans.append((oid, value))
            elif isinstance(oid, str):  # if oid is a string, assume nodeid lookup
                # value must then be a rfc1902/pyasn1 type, if oid is not a scalar
                if not oid.endswith(".0"):
                    assert isinstance(value, univ.Integer) or isinstance(value, univ.OctetString) or isinstance(value, univ.ObjectIdentifier)
                oidvalues_trans.append((nodeid(oid), value))

        (error_indication, error_status, error_index, varbinds) = \
            cmdgen.CommandGenerator().setCmd(self.auth, cmdgen.UdpTransportTarget((self.host, self.port), timeout=self.timeout, retries=self.retries), *oidvalues_trans)  # pylint: disable=W0612
        if error_indication or error_status:
            self.__set_error(error_indication, error_status, error_index, varbinds)
            raise SnmpError("SNMP set command on %s of oid values %r failed" % (self.host, oidvalues_trans), error_indication, error_status, error_index, varbinds)
        return SnmpVarBinds(varbinds)