我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用errno.EINVAL。
def test_offset_overflow(self): # specify an offset > file size offset = len(self.DATA) + 4096 try: sent = os.sendfile(self.sockno, self.fileno, offset, 4096) except OSError as e: # Solaris can raise EINVAL if offset >= file length, ignore. if e.errno != errno.EINVAL: raise else: self.assertEqual(sent, 0) self.client.shutdown(socket.SHUT_RDWR) self.client.close() self.server.wait() data = self.server.handler_instance.get_data() self.assertEqual(data, b'')
def truncate(self, size=None): """Truncate the file's size. If the optional size argument is present, the file is truncated to (at most) that size. The size defaults to the current position. The current file position is not changed unless the position is beyond the new file size. If the specified size exceeds the file's current size, the file remains unchanged. """ _complain_ifclosed(self.closed) if size is None: size = self.pos elif size < 0: raise IOError(EINVAL, "Negative size not allowed") elif size < self.pos: self.pos = size self.buf = self.getvalue()[:size] self.len = size
def live(self): if not self.proc_path: return self.info("Proc path: %s" % self.proc_path) key = choice(self.proc_keys) filename = path_join(self.proc_path, key) data = self.generator.createValue() self.info("Write data in %s: (len=%s) %r" % (filename, len(data), data)) try: output = open(filename, 'w') output.write(data) output.close() except IOError, err: if err.errno in (EINVAL, EPERM): pass elif err.errno in (ENOENT, EACCES): self.error("Unable to write %s: %s" % (filename, err)) self.removeKey(key) else: raise
def _setgroups_until_success(l): while(1): # NASTY NASTY HACK (but glibc does it so it must be okay): # In case sysconfig didn't give the right answer, find the limit # on max groups by just looping, trying to set fewer and fewer # groups each time until it succeeds. try: setgroups(l) except ValueError: # This exception comes from python itself restricting # number of groups allowed. if len(l) > 1: del l[-1] else: raise except OSError, e: if e.errno == errno.EINVAL and len(l) > 1: # This comes from the OS saying too many groups del l[-1] else: raise else: # Success, yay! return
def get_proc_inodes(self, pid): inodes = defaultdict(list) for fd in os.listdir("%s/%s/fd" % (self._procfs_path, pid)): try: inode = readlink("%s/%s/fd/%s" % (self._procfs_path, pid, fd)) except OSError as err: # ENOENT == file which is gone in the meantime; # os.stat('/proc/%s' % self.pid) will be done later # to force NSP (if it's the case) if err.errno in (errno.ENOENT, errno.ESRCH): continue elif err.errno == errno.EINVAL: # not a link continue else: raise else: if inode.startswith('socket:['): # the process is using a socket inode = inode[8:][:-1] inodes[inode].append((pid, int(fd))) return inodes
def cmdline(self): if OPENBSD and self.pid == 0: return None # ...else it crashes elif NETBSD: # XXX - most of the times the underlying sysctl() call on Net # and Open BSD returns a truncated string. # Also /proc/pid/cmdline behaves the same so it looks # like this is a kernel bug. try: return cext.proc_cmdline(self.pid) except OSError as err: if err.errno == errno.EINVAL: if not pid_exists(self.pid): raise NoSuchProcess(self.pid, self._name) else: raise ZombieProcess(self.pid, self._name, self._ppid) else: raise else: return cext.proc_cmdline(self.pid)
def cwd(self): """Return process current working directory.""" # sometimes we get an empty string, in which case we turn # it into None if OPENBSD and self.pid == 0: return None # ...else it would raise EINVAL elif NETBSD: with wrap_exceptions_procfs(self): return os.readlink("/proc/%s/cwd" % self.pid) elif hasattr(cext, 'proc_open_files'): # FreeBSD < 8 does not support functions based on # kinfo_getfile() and kinfo_getvmmap() return cext.proc_cwd(self.pid) or None else: raise NotImplementedError( "supported only starting from FreeBSD 8" if FREEBSD else "")
def _setecho(fd, state): errmsg = 'setecho() may not be called on this platform' try: attr = termios.tcgetattr(fd) except termios.error as err: if err.args[0] == errno.EINVAL: raise IOError(err.args[0], '%s: %s.' % (err.args[1], errmsg)) raise if state: attr[3] = attr[3] | termios.ECHO else: attr[3] = attr[3] & ~termios.ECHO try: # I tried TCSADRAIN and TCSAFLUSH, but these were inconsistent and # blocked on some platforms. TCSADRAIN would probably be ideal. termios.tcsetattr(fd, termios.TCSANOW, attr) except IOError as err: if err.args[0] == errno.EINVAL: raise IOError(err.args[0], '%s: %s.' % (err.args[1], errmsg)) raise
def getecho(self): '''This returns the terminal echo mode. This returns True if echo is on or False if echo is off. Child applications that are expecting you to enter a password often set ECHO False. See waitnoecho(). Not supported on platforms where ``isatty()`` returns False. ''' try: attr = termios.tcgetattr(self.fd) except termios.error as err: errmsg = 'getecho() may not be called on this platform' if err.args[0] == errno.EINVAL: raise IOError(err.args[0], '%s: %s.' % (err.args[1], errmsg)) raise self.echo = bool(attr[3] & termios.ECHO) return self.echo
def test_accept(self, tco): with pytest.raises(nfc.llcp.Error) as excinfo: tco.accept() assert excinfo.value.errno == errno.EINVAL tco.setsockopt(nfc.llcp.SO_RCVMIU, 1000) tco.setsockopt(nfc.llcp.SO_RCVBUF, 2) tco.listen(backlog=1) assert tco.state.LISTEN is True tco.enqueue(nfc.llcp.pdu.Connect(tco.addr, 17, 500, 15)) dlc = tco.accept() assert isinstance(dlc, nfc.llcp.tco.DataLinkConnection) assert dlc.state.ESTABLISHED is True assert dlc.getsockopt(nfc.llcp.SO_RCVMIU) == 1000 assert dlc.getsockopt(nfc.llcp.SO_SNDMIU) == 500 assert dlc.getsockopt(nfc.llcp.SO_RCVBUF) == 2 assert tco.dequeue(128, 4) == \ nfc.llcp.pdu.ConnectionComplete(17, tco.addr, 1000, 2) threading.Timer(0.01, tco.close).start() with pytest.raises(nfc.llcp.Error) as excinfo: tco.accept() assert excinfo.value.errno == errno.EPIPE with pytest.raises(nfc.llcp.Error) as excinfo: tco.accept() assert excinfo.value.errno == errno.ESHUTDOWN
def test_accept_connect(self, llc, ldl, dlc, peer_miu, send_miu): with pytest.raises(nfc.llcp.Error) as excinfo: llc.accept(object()) assert excinfo.value.errno == errno.ENOTSOCK with pytest.raises(nfc.llcp.Error) as excinfo: llc.accept(ldl) assert excinfo.value.errno == errno.EOPNOTSUPP with pytest.raises(nfc.llcp.Error) as excinfo: llc.accept(dlc) assert excinfo.value.errno == errno.EINVAL connect_pdu = nfc.llcp.pdu.Connect(4, 32, peer_miu) threading.Timer(0.01, llc.dispatch, (connect_pdu,)).start() llc.bind(dlc, b'urn:nfc:sn:snep') llc.listen(dlc, 0) sock = llc.accept(dlc) assert isinstance(sock, nfc.llcp.tco.DataLinkConnection) assert llc.getsockopt(sock, nfc.llcp.SO_SNDMIU) == send_miu assert llc.getpeername(sock) == 32 assert llc.getsockname(sock) == 4
def poll(self, event, timeout): if self.state.SHUTDOWN: raise err.Error(errno.ESHUTDOWN) if event == "recv": if self.state.ESTABLISHED or self.state.CLOSE_WAIT: rcvd_pdu = super(DataLinkConnection, self).poll(event, timeout) if self.state.ESTABLISHED or self.state.CLOSE_WAIT: return isinstance(rcvd_pdu, pdu.Information) elif event == "send": if self.state.ESTABLISHED: if super(DataLinkConnection, self).poll(event, timeout): return self.state.ESTABLISHED return False elif event == "acks": with self.acks_ready: if not self.acks_recvd > 0: self.acks_ready.wait(timeout) if self.acks_recvd > 0: self.acks_recvd = self.acks_recvd - 1 return True return False else: raise err.Error(errno.EINVAL)
def cpu_affinity_set(self, cpus): # Pre-emptively check if CPUs are valid because the C # function has a weird behavior in case of invalid CPUs, # see: https://github.com/giampaolo/psutil/issues/586 allcpus = tuple(range(len(per_cpu_times()))) for cpu in cpus: if cpu not in allcpus: raise ValueError("invalid CPU #%i (choose between %s)" % (cpu, allcpus)) try: cext.proc_cpu_affinity_set(self.pid, cpus) except OSError as err: # 'man cpuset_setaffinity' about EDEADLK: # <<the call would leave a thread without a valid CPU to run # on because the set does not overlap with the thread's # anonymous mask>> if err.errno in (errno.EINVAL, errno.EDEADLK): for cpu in cpus: if cpu not in allcpus: raise ValueError( "invalid CPU #%i (choose between %s)" % ( cpu, allcpus)) raise
def _find_device_root(device_book): ''' Given the full path to a book on the device, return the path to the Kindle device eg. "C:\", "/Volumes/Kindle" ''' device_root = None if sys.platform == "win32": device_root = os.path.join(device_book.split(os.sep)[0], os.sep) elif sys.platform == "darwin" or "linux" in sys.platform: # Find "documents" in path hierarchy, we want to include the first os.sep so slice one further than we find it index = device_book.index("%sdocuments%s" % (os.sep, os.sep)) device_root = device_book[:index+1] else: raise EnvironmentError(errno.EINVAL, "Unknown platform %s" % (sys.platform)) magic_file = os.path.join(device_root, "system", "version.txt") if os.path.exists(magic_file) and open(magic_file).readline().startswith("Kindle"): return device_root raise EnvironmentError(errno.ENOENT, "Kindle device not found (%s)" % (device_root))
def run_server(self): """Run a WSGI server.""" eventlet.wsgi.HttpProtocol.default_request_version = "HTTP/1.0" eventlet.hubs.use_hub('poll') eventlet.patcher.monkey_patch(all=False, socket=True) self.pool = eventlet.GreenPool(size=self.threads) socket_timeout = cfg.CONF.eventlet_opts.client_socket_timeout or None try: eventlet.wsgi.server( self.sock, self.application, custom_pool=self.pool, url_length_limit=URL_LENGTH_LIMIT, log=self._wsgi_logger, debug=cfg.CONF.debug, keepalive=cfg.CONF.eventlet_opts.wsgi_keep_alive, socket_timeout=socket_timeout) except socket.error as err: if err[0] != errno.EINVAL: raise self.pool.waitall()
def set_options(self, sock, bound=False): sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if hasattr(socket, 'SO_REUSEPORT'): # pragma: no cover try: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except socket.error as err: if err[0] not in (errno.ENOPROTOOPT, errno.EINVAL): raise if not bound: self.bind(sock) sock.setblocking(0) # make sure that the socket can be inherited if hasattr(sock, "set_inheritable"): sock.set_inheritable(True) sock.listen(self.conf.backlog) return sock
def test_open_files_file_gone(self): # simulates a file which gets deleted during open_files() # execution p = psutil.Process() files = p.open_files() with tempfile.NamedTemporaryFile(): # give the kernel some time to see the new file call_until(p.open_files, "len(ret) != %i" % len(files)) with mock.patch('psutil._pslinux.os.readlink', side_effect=OSError(errno.ENOENT, "")) as m: files = p.open_files() assert not files assert m.called # also simulate the case where os.readlink() returns EINVAL # in which case psutil is supposed to 'continue' with mock.patch('psutil._pslinux.os.readlink', side_effect=OSError(errno.EINVAL, "")) as m: self.assertEqual(p.open_files(), []) assert m.called # --- mocked tests
def test_isfile_strict(self): from psutil._common import isfile_strict this_file = os.path.abspath(__file__) assert isfile_strict(this_file) assert not isfile_strict(os.path.dirname(this_file)) with mock.patch('psutil._common.os.stat', side_effect=OSError(errno.EPERM, "foo")): self.assertRaises(OSError, isfile_strict, this_file) with mock.patch('psutil._common.os.stat', side_effect=OSError(errno.EACCES, "foo")): self.assertRaises(OSError, isfile_strict, this_file) with mock.patch('psutil._common.os.stat', side_effect=OSError(errno.EINVAL, "foo")): assert not isfile_strict(this_file) with mock.patch('psutil._common.stat.S_ISREG', return_value=False): assert not isfile_strict(this_file)
def cmdline(self): if OPENBSD and self.pid == 0: return [] # ...else it crashes elif NETBSD: # XXX - most of the times the underlying sysctl() call on Net # and Open BSD returns a truncated string. # Also /proc/pid/cmdline behaves the same so it looks # like this is a kernel bug. try: return cext.proc_cmdline(self.pid) except OSError as err: if err.errno == errno.EINVAL: if not pid_exists(self.pid): raise NoSuchProcess(self.pid, self._name) else: raise ZombieProcess(self.pid, self._name, self._ppid) else: raise else: return cext.proc_cmdline(self.pid)
def io_transfer(iosock,flag,ifreq): """ send & recieve an ifreq struct :param iosock: io socket :param flag: sockios control call :param ifreq: ifreq to send :returns: an the ifreq struct recieved """ try: return ioctl(iosock.fileno(),flag,ifreq) except (AttributeError,struct.error) as e: # either sock is not valid or a bad value passed to ifreq if e.message.find('fileno'): raise error(errno.ENOTSOCK,"Bad socket") else: raise error(errno.EINVAL,e) except IOError as e: # generally device cannot be found sort but can also be # permissions etc, catch and reraise as our own if e.errno is not None: # just in case we have a none 2-tuple error raise error(e.errno,e.strerror) else: raise error(-1,e) except Exception as e: # blanket catchall raise error(-1,e.args[0])
def stop(self): if not os.path.isfile(self.pidfile): print('No "twistd.pid" file found in {}'.format(self.nodedir)) return with open(self.pidfile, 'r') as f: pid = f.read() pid = int(pid) print("Trying to kill PID {}...".format(pid)) try: os.kill(pid, signal.SIGTERM) except OSError as err: print(err) if err.errno == errno.ESRCH or err.errno == errno.EINVAL: os.remove(self.pidfile) else: raise
def accept(self): with self.lock: if self.state.SHUTDOWN: raise Error(errno.EBADF) if not self.state.LISTEN: self.err("accept() but socket state is {0}".format(self.state)) raise Error(errno.EINVAL) self.recv_buf += 1 try: pdu = super(DataLinkConnection, self).recv() except IndexError: raise Error(errno.EPIPE) self.recv_buf -= 1 if isinstance(pdu, Connect): dlc = DataLinkConnection(self.recv_miu, self.recv_win) dlc.addr = self.addr dlc.peer = pdu.ssap dlc.send_miu = pdu.miu dlc.send_win = pdu.rw pdu = ConnectionComplete(dlc.peer, dlc.addr) pdu.miu, pdu.rw = dlc.recv_miu, dlc.recv_win log.info("accepting CONNECT from SAP %d" % dlc.peer) dlc.state.ESTABLISHED = True self.send_queue.append(pdu) return dlc raise RuntimeError("only CONNECT expected, not "+ pdu.name)
def poll(self, event, timeout): if self.state.SHUTDOWN: raise Error(errno.EBADF) if not event in ("recv", "send", "acks"): raise Error(errno.EINVAL) if event == "recv": if self.state.ESTABLISHED or self.state.CLOSE_WAIT: ptype = super(DataLinkConnection, self).poll(event, timeout) if self.state.ESTABLISHED or self.state.CLOSE_WAIT: return ptype == ProtocolDataUnit.Information else: return False if event == "send": if self.state.ESTABLISHED: if super(DataLinkConnection, self).poll(event, timeout): return self.state.ESTABLISHED if event == "acks": with self.acks_ready: while not self.acks_recvd > 0: self.acks_ready.wait(timeout) if self.acks_recvd > 0: self.acks_recvd = self.acks_recvd - 1 return True return False
def test_does_not_crash(self): """Check if get_terminal_size() returns a meaningful value. There's no easy portable way to actually check the size of the terminal, so let's check if it returns something sensible instead. """ try: size = os.get_terminal_size() except OSError as e: if sys.platform == "win32" or e.errno in (errno.EINVAL, errno.ENOTTY): # Under win32 a generic OSError can be thrown if the # handle cannot be retrieved self.skipTest("failed to query terminal size") raise self.assertGreaterEqual(size.columns, 0) self.assertGreaterEqual(size.lines, 0)
def test_stty_match(self): """Check if stty returns the same results stty actually tests stdin, so get_terminal_size is invoked on stdin explicitly. If stty succeeded, then get_terminal_size() should work too. """ try: size = subprocess.check_output(['stty', 'size']).decode().split() except (FileNotFoundError, subprocess.CalledProcessError): self.skipTest("stty invocation failed") expected = (int(size[1]), int(size[0])) # reversed order try: actual = os.get_terminal_size(sys.__stdin__.fileno()) except OSError as e: if sys.platform == "win32" or e.errno in (errno.EINVAL, errno.ENOTTY): # Under win32 a generic OSError can be thrown if the # handle cannot be retrieved self.skipTest("failed to query terminal size") raise self.assertEqual(expected, actual)
def test_mknod_dir_fd(self): # Test using mknodat() to create a FIFO (the only use specified # by POSIX). support.unlink(support.TESTFN) mode = stat.S_IFIFO | stat.S_IRUSR | stat.S_IWUSR f = posix.open(posix.getcwd(), posix.O_RDONLY) try: posix.mknod(support.TESTFN, mode, 0, dir_fd=f) except OSError as e: # Some old systems don't allow unprivileged users to use # mknod(), or only support creating device nodes. self.assertIn(e.errno, (errno.EPERM, errno.EINVAL)) else: self.assertTrue(stat.S_ISFIFO(posix.stat(support.TESTFN).st_mode)) finally: posix.close(f)
def connectThread(self): while not self.connected: try: self.client.connect((HOST, PORT)) except socket.error as e: pass finally: if e.errno == errno.EINPROGRESS: time.sleep(1.0) elif e.errno == errno.ECONNREFUSED: time.sleep(1.0) elif e.errno == errno.EALREADY: time.sleep(1.0) elif e.errno == errno.EINVAL: break elif e.errno == errno.EISCONN: logger.log("[button]: Connected to Artoo.") self.connected = True self.buttonsInitialized = False else: logger.log("[button]: Unexpected socket exception: %s" % e) time.sleep(1.0)
def communicate(self, input=None): """Interact with process: Send data to stdin. Read data from stdout and stderr, until end-of-file is reached. Wait for process to terminate. The optional input argument should be a string to be sent to the child process, or None, if no data should be sent to the child. communicate() returns a tuple (stdout, stderr).""" # Optimization: If we are only using one pipe, or no pipe at # all, using select() or threads is unnecessary. if [self.stdin, self.stdout, self.stderr].count(None) >= 2: stdout = None stderr = None if self.stdin: if input: try: self.stdin.write(input) except IOError as e: if e.errno != errno.EPIPE and e.errno != errno.EINVAL: raise self.stdin.close() elif self.stdout: stdout = self.stdout.read() self.stdout.close() elif self.stderr: stderr = self.stderr.read() self.stderr.close() self.wait() return (stdout, stderr) return self._communicate(input)
def bind(self, *pos, **kw): """ Implements proxy connection for UDP sockets, which happens during the bind() phase. """ proxy_type, proxy_addr, proxy_port, rdns, username, password = self.proxy if not proxy_type or self.type != socket.SOCK_DGRAM: return _orig_socket.bind(self, *pos, **kw) if self._proxyconn: raise socket.error(EINVAL, "Socket already bound to an address") if proxy_type != SOCKS5: msg = "UDP only supported by SOCKS5 proxy type" raise socket.error(EOPNOTSUPP, msg) super(socksocket, self).bind(*pos, **kw) # Need to specify actual local port because # some relays drop packets if a port of zero is specified. # Avoid specifying host address in case of NAT though. _, port = self.getsockname() dst = ("0", port) self._proxyconn = _orig_socket() proxy = self._proxy_addr() self._proxyconn.connect(proxy) UDP_ASSOCIATE = b"\x03" _, relay = self._SOCKS5_request(self._proxyconn, UDP_ASSOCIATE, dst) # The relay is most likely on the same host as the SOCKS proxy, # but some proxies return a private IP address (10.x.y.z) host, _ = proxy _, port = relay super(socksocket, self).connect((host, port)) super(socksocket, self).settimeout(self._timeout) self.proxy_sockname = ("0.0.0.0", 0) # Unknown
def open_if_exists(filename, mode='rb'): """Returns a file descriptor for the filename if that file exists, otherwise `None`. """ try: return open(filename, mode) except IOError as e: if e.errno not in (errno.ENOENT, errno.EISDIR, errno.EINVAL): raise
