comparison mercurial/ui.py @ 40593:7bffbbe03e90

ui: hide fin/fout/ferr attributes behind @property functions

This allows keeping references to fout/ferr/fin which are updated when these
properties are changed. See the next patch.

40592:d8997c5ce2ff

        self._terminfoparams = {}
        self._styles = {}
        self._uninterruptible = False

        if src:
            self.fout = src.fout
            self.ferr = src.ferr
            self.fin = src.fin
            self._finoutredirected = src._finoutredirected
            self.pageractive = src.pageractive
            self._disablepager = src._disablepager
            self._tweaked = src._tweaked

            self.fixconfig()

            self.httppasswordmgrdb = src.httppasswordmgrdb
            self._blockedtimes = src._blockedtimes
        else:
            self.fout = procutil.stdout
            self.ferr = procutil.stderr
            self.fin = procutil.stdin
            self._finoutredirected = False
            self.pageractive = False
            self._disablepager = False
            self._tweaked = False

    @util.propertycache
    def paths(self):
        return paths(self)

    def pushbuffer(self, error=False, subproc=False, labeled=False):
        """install a buffer to capture standard output of the ui object

        If error is True, the error output will be captured too.

            self._bufferapplylabels = None

        return "".join(self._buffers.pop())

    def _isbuffered(self, dest):
        if dest is self.fout:
            return bool(self._buffers)
        if dest is self.ferr:
            return bool(self._bufferstates and self._bufferstates[-1][0])
        return False

    def canwritewithoutlabels(self):
        '''check if write skips the label'''

        When labeling output for a specific command, a label of
        "cmdname.type" is recommended. For example, status issues
        a label of "status.modified" for modified files.
        '''
        self._write(self.fout, *args, **opts)

    def write_err(self, *args, **opts):
        self._write(self.ferr, *args, **opts)

    def _write(self, dest, *args, **opts):
        if self._isbuffered(dest):
            if self._bufferapplylabels:
                label = opts.get(r'label', '')

        msg = b''.join(args)

        # opencode timeblockedsection because this is a critical path
        starttime = util.timer()
        try:
            if dest is self.ferr and not getattr(self.fout, 'closed', False):
                self.fout.flush()
            if self._colormode == 'win32':
                # windows color printing is its own can of crab, defer to
                # the color module and that is it.
                color.win32print(self, dest.write, msg, **opts)
            else:

                    label = opts.get(r'label', '')
                    msg = self.label(msg, label)
                dest.write(msg)
            # stderr may be buffered under win32 when redirected to files,
            # including stdout.
            if dest is self.ferr and not getattr(self.ferr, 'closed', False):
                dest.flush()
        except IOError as err:
            if (dest is self.ferr
                and err.errno in (errno.EPIPE, errno.EIO, errno.EBADF)):
                # no way to report the error, so ignore it
                return
            raise error.StdioError(err)
        finally:

    def flush(self):
        # opencode timeblockedsection because this is a critical path
        starttime = util.timer()
        try:
            try:
                self.fout.flush()
            except IOError as err:
                if err.errno not in (errno.EPIPE, errno.EIO, errno.EBADF):
                    raise error.StdioError(err)
            finally:
                try:
                    self.ferr.flush()
                except IOError as err:
                    if err.errno not in (errno.EPIPE, errno.EIO, errno.EBADF):
                        raise error.StdioError(err)
        finally:
            self._blockedtimes['stdio_blocked'] += \

        '''
        i = self.configbool("ui", "interactive")
        if i is None:
            # some environments replace stdin without implementing isatty
            # usually those are non-interactive
            return self._isatty(self.fin)

        return i

    def termwidth(self):
        '''how wide is the terminal in columns?

        i = self.configbool("ui", "formatted")
        if i is None:
            # some environments replace stdout without implementing isatty
            # usually those are non-interactive
            return self._isatty(self.fout)

        return i

    def _readline(self):
        # Replacing stdin/stdout temporarily is a hard problem on Python 3
        # because they have to be text streams with *no buffering*. Instead,
        # we use rawinput() only if call_readline() will be invoked by
        # PyOS_Readline(), so no I/O will be made at Python layer.
        usereadline = (self._isatty(self.fin) and self._isatty(self.fout)
                       and procutil.isstdin(self.fin)
                       and procutil.isstdout(self.fout))
        if usereadline:
            try:
                # magically add command line editing support, where
                # available
                import readline

                # When stdin is in binary mode on Windows, it can cause
                # raw_input() to emit an extra trailing carriage return
                if pycompat.oslinesep == b'\r\n' and line.endswith(b'\r'):
                    line = line[:-1]
            else:
                self.fout.write(b' ')
                self.fout.flush()
                line = self.fin.readline()
                if not line:
                    raise EOFError
                line = line.rstrip(pycompat.oslinesep)

        return line

        """
        if not self.interactive():
            self.write(msg, ' ', label='ui.prompt')
            self.write(default or '', "\n", label='ui.promptecho')
            return default
        self._writenobuf(self.fout, msg, label='ui.prompt')
        self.flush()
        try:
            r = self._readline()
            if not r:
                r = default

            self.write_err(self.label(prompt or _('password: '), 'ui.prompt'))
            # disable getpass() only if explicitly specified. it's still valid
            # to interact with tty even if fin is not a tty.
            with self.timeblockedsection('stdio'):
                if self.configbool('ui', 'nontty'):
                    l = self.fin.readline()
                    if not l:
                        raise EOFError
                    return l.rstrip('\n')
                else:
                    return getpass.getpass('')

        if blockedtag is None:
            # Long cmds tend to be because of an absolute path on cmd. Keep
            # the tail end instead
            cmdsuffix = cmd.translate(None, _keepalnum)[-85:]
            blockedtag = 'unknown_system_' + cmdsuffix
        out = self.fout
        if any(s[1] for s in self._bufferstates):
            out = self
        with self.timeblockedsection(blockedtag):
            rc = self._runsystem(cmd, environ=environ, cwd=cwd, out=out)
        if rc and onerr:

            if config is None or not self.configbool('devel', config):
                return
        msg = 'devel-warn: ' + msg
        stacklevel += 1 # get in develwarn
        if self.tracebackflag:
            util.debugstacktrace(msg, stacklevel, self.ferr, self.fout)
            self.log('develwarn', '%s at:\n%s' %
                     (msg, ''.join(util.getstackframes(stacklevel))))
        else:
            curframe = inspect.currentframe()
            calframe = inspect.getouterframes(curframe, 2)

40593:7bffbbe03e90

        self._terminfoparams = {}
        self._styles = {}
        self._uninterruptible = False

        if src:
            self._fout = src._fout
            self._ferr = src._ferr
            self._fin = src._fin
            self._finoutredirected = src._finoutredirected
            self.pageractive = src.pageractive
            self._disablepager = src._disablepager
            self._tweaked = src._tweaked

            self.fixconfig()

            self.httppasswordmgrdb = src.httppasswordmgrdb
            self._blockedtimes = src._blockedtimes
        else:
            self._fout = procutil.stdout
            self._ferr = procutil.stderr
            self._fin = procutil.stdin
            self._finoutredirected = False
            self.pageractive = False
            self._disablepager = False
            self._tweaked = False

    @util.propertycache
    def paths(self):
        return paths(self)

    @property
    def fout(self):
        return self._fout

    @fout.setter
    def fout(self, f):
        self._fout = f

    @property
    def ferr(self):
        return self._ferr

    @ferr.setter
    def ferr(self, f):
        self._ferr = f

    @property
    def fin(self):
        return self._fin

    @fin.setter
    def fin(self, f):
        self._fin = f

    def pushbuffer(self, error=False, subproc=False, labeled=False):
        """install a buffer to capture standard output of the ui object

        If error is True, the error output will be captured too.

            self._bufferapplylabels = None

        return "".join(self._buffers.pop())

    def _isbuffered(self, dest):
        if dest is self._fout:
            return bool(self._buffers)
        if dest is self._ferr:
            return bool(self._bufferstates and self._bufferstates[-1][0])
        return False

    def canwritewithoutlabels(self):
        '''check if write skips the label'''

        When labeling output for a specific command, a label of
        "cmdname.type" is recommended. For example, status issues
        a label of "status.modified" for modified files.
        '''
        self._write(self._fout, *args, **opts)

    def write_err(self, *args, **opts):
        self._write(self._ferr, *args, **opts)

    def _write(self, dest, *args, **opts):
        if self._isbuffered(dest):
            if self._bufferapplylabels:
                label = opts.get(r'label', '')

        msg = b''.join(args)

        # opencode timeblockedsection because this is a critical path
        starttime = util.timer()
        try:
            if dest is self._ferr and not getattr(self._fout, 'closed', False):
                self._fout.flush()
            if self._colormode == 'win32':
                # windows color printing is its own can of crab, defer to
                # the color module and that is it.
                color.win32print(self, dest.write, msg, **opts)
            else:

                    label = opts.get(r'label', '')
                    msg = self.label(msg, label)
                dest.write(msg)
            # stderr may be buffered under win32 when redirected to files,
            # including stdout.
            if dest is self._ferr and not getattr(self._ferr, 'closed', False):
                dest.flush()
        except IOError as err:
            if (dest is self._ferr
                and err.errno in (errno.EPIPE, errno.EIO, errno.EBADF)):
                # no way to report the error, so ignore it
                return
            raise error.StdioError(err)
        finally:

    def flush(self):
        # opencode timeblockedsection because this is a critical path
        starttime = util.timer()
        try:
            try:
                self._fout.flush()
            except IOError as err:
                if err.errno not in (errno.EPIPE, errno.EIO, errno.EBADF):
                    raise error.StdioError(err)
            finally:
                try:
                    self._ferr.flush()
                except IOError as err:
                    if err.errno not in (errno.EPIPE, errno.EIO, errno.EBADF):
                        raise error.StdioError(err)
        finally:
            self._blockedtimes['stdio_blocked'] += \

        '''
        i = self.configbool("ui", "interactive")
        if i is None:
            # some environments replace stdin without implementing isatty
            # usually those are non-interactive
            return self._isatty(self._fin)

        return i

    def termwidth(self):
        '''how wide is the terminal in columns?

        i = self.configbool("ui", "formatted")
        if i is None:
            # some environments replace stdout without implementing isatty
            # usually those are non-interactive
            return self._isatty(self._fout)

        return i

    def _readline(self):
        # Replacing stdin/stdout temporarily is a hard problem on Python 3
        # because they have to be text streams with *no buffering*. Instead,
        # we use rawinput() only if call_readline() will be invoked by
        # PyOS_Readline(), so no I/O will be made at Python layer.
        usereadline = (self._isatty(self._fin) and self._isatty(self._fout)
                       and procutil.isstdin(self._fin)
                       and procutil.isstdout(self._fout))
        if usereadline:
            try:
                # magically add command line editing support, where
                # available
                import readline

                # When stdin is in binary mode on Windows, it can cause
                # raw_input() to emit an extra trailing carriage return
                if pycompat.oslinesep == b'\r\n' and line.endswith(b'\r'):
                    line = line[:-1]
            else:
                self._fout.write(b' ')
                self._fout.flush()
                line = self._fin.readline()
                if not line:
                    raise EOFError
                line = line.rstrip(pycompat.oslinesep)

        return line

        """
        if not self.interactive():
            self.write(msg, ' ', label='ui.prompt')
            self.write(default or '', "\n", label='ui.promptecho')
            return default
        self._writenobuf(self._fout, msg, label='ui.prompt')
        self.flush()
        try:
            r = self._readline()
            if not r:
                r = default

            self.write_err(self.label(prompt or _('password: '), 'ui.prompt'))
            # disable getpass() only if explicitly specified. it's still valid
            # to interact with tty even if fin is not a tty.
            with self.timeblockedsection('stdio'):
                if self.configbool('ui', 'nontty'):
                    l = self._fin.readline()
                    if not l:
                        raise EOFError
                    return l.rstrip('\n')
                else:
                    return getpass.getpass('')

        if blockedtag is None:
            # Long cmds tend to be because of an absolute path on cmd. Keep
            # the tail end instead
            cmdsuffix = cmd.translate(None, _keepalnum)[-85:]
            blockedtag = 'unknown_system_' + cmdsuffix
        out = self._fout
        if any(s[1] for s in self._bufferstates):
            out = self
        with self.timeblockedsection(blockedtag):
            rc = self._runsystem(cmd, environ=environ, cwd=cwd, out=out)
        if rc and onerr:

            if config is None or not self.configbool('devel', config):
                return
        msg = 'devel-warn: ' + msg
        stacklevel += 1 # get in develwarn
        if self.tracebackflag:
            util.debugstacktrace(msg, stacklevel, self._ferr, self._fout)
            self.log('develwarn', '%s at:\n%s' %
                     (msg, ''.join(util.getstackframes(stacklevel))))
        else:
            curframe = inspect.currentframe()
            calframe = inspect.getouterframes(curframe, 2)