view tests/fakedirstatewritetime.py @ 46083:81c1f5d1801f

procutils: don't try to get `.buffer` if sys.stdin is None

While hunting down following test failure of test-chg.t on Python 3, I stumbled
the case when `.buffer` is not available as sys.stdin is None.

	--- /home/pulkit/repo/hg-committed/tests/test-chg.t
	+++ /home/pulkit/repo/hg-committed/tests/test-chg.t.err
	@@ -203,7 +203,31 @@
	   $ CHGDEBUG=1 chg version -q 0<&-
		 chg: debug: * stdio fds are missing (glob)
	     chg: debug: * execute original hg (glob)
		 -  Mercurial Distributed SCM * (glob)
		 +  Traceback (most recent call last):
		 +    File "/tmp/hgtests.avspvsq4/install/bin/hg", line 43, in <module>
		 +      dispatch.run()
		 +    File "/usr/lib/python3.6/importlib/util.py", line 233, in
		 __getattribute__
		 +      self.__spec__.loader.exec_module(self)
		 +    File "<frozen importlib._bootstrap_external>", line 678, in
		 exec_module
		 +    File "<frozen importlib._bootstrap>", line 219, in
		 _call_with_frames_removed
		 +    File
		 "/tmp/hgtests.avspvsq4/install/lib/python/mercurial/dispatch.py", line
		 726, in <module>
		 +      class lazyaliasentry(object):
		 +    File
		 "/tmp/hgtests.avspvsq4/install/lib/python/mercurial/dispatch.py", line
		 737, in lazyaliasentry
		 +      @util.propertycache
		 +    File "/usr/lib/python3.6/importlib/util.py", line 233, in
		 __getattribute__
		 +      self.__spec__.loader.exec_module(self)
		 +    File "<frozen importlib._bootstrap_external>", line 678, in
		 exec_module
		 +    File "<frozen importlib._bootstrap>", line 219, in
		 _call_with_frames_removed
		 +    File "/tmp/hgtests.avspvsq4/install/lib/python/mercurial/util.py",
		 line 3473, in <module>
		 +      f=procutil.stderr,
		 +    File "/usr/lib/python3.6/importlib/util.py", line 233, in
		 __getattribute__
		 +      self.__spec__.loader.exec_module(self)
		 +    File "<frozen importlib._bootstrap_external>", line 678, in
		 exec_module
		 +    File "<frozen importlib._bootstrap>", line 219, in
		 _call_with_frames_removed
		 +    File
		 "/tmp/hgtests.avspvsq4/install/lib/python/mercurial/utils/procutil.py",
		 line 127, in <module>
		 +      stdin = sys.stdin.buffer
		 +  AttributeError: 'NoneType' object has no attribute 'buffer'
		 +  [1]

		  server lifecycle
		   ----------------

Differential Revision: https://phab.mercurial-scm.org/D9500

# extension to emulate invoking 'dirstate.write()' at the time
# specified by '[fakedirstatewritetime] fakenow', only when
# 'dirstate.write()' is invoked via functions below:
#
#   - 'workingctx._poststatusfixup()' (= 'repo.status()')
#   - 'committablectx.markcommitted()'

from __future__ import absolute_import

from mercurial import (
    context,
    dirstate,
    extensions,
    policy,
    registrar,
)
from mercurial.utils import dateutil

try:
    from mercurial import rustext

    rustext.__name__  # force actual import (see hgdemandimport)
except ImportError:
    rustext = None

configtable = {}
configitem = registrar.configitem(configtable)

configitem(
    b'fakedirstatewritetime',
    b'fakenow',
    default=None,
)

parsers = policy.importmod('parsers')
rustmod = policy.importrust('parsers')


def pack_dirstate(fakenow, orig, dmap, copymap, pl, now):
    # execute what original parsers.pack_dirstate should do actually
    # for consistency
    actualnow = int(now)
    for f, e in dmap.items():
        if e[0] == 'n' and e[3] == actualnow:
            e = parsers.dirstatetuple(e[0], e[1], e[2], -1)
            dmap[f] = e

    return orig(dmap, copymap, pl, fakenow)


def fakewrite(ui, func):
    # fake "now" of 'pack_dirstate' only if it is invoked while 'func'

    fakenow = ui.config(b'fakedirstatewritetime', b'fakenow')
    if not fakenow:
        # Execute original one, if fakenow isn't configured. This is
        # useful to prevent subrepos from executing replaced one,
        # because replacing 'parsers.pack_dirstate' is also effective
        # in subrepos.
        return func()

    # parsing 'fakenow' in YYYYmmddHHMM format makes comparison between
    # 'fakenow' value and 'touch -t YYYYmmddHHMM' argument easy
    fakenow = dateutil.parsedate(fakenow, [b'%Y%m%d%H%M'])[0]

    if rustmod is not None:
        # The Rust implementation does not use public parse/pack dirstate
        # to prevent conversion round-trips
        orig_dirstatemap_write = dirstate.dirstatemap.write
        wrapper = lambda self, st, now: orig_dirstatemap_write(
            self, st, fakenow
        )
        dirstate.dirstatemap.write = wrapper

    orig_dirstate_getfsnow = dirstate._getfsnow
    wrapper = lambda *args: pack_dirstate(fakenow, orig_pack_dirstate, *args)

    orig_module = parsers
    orig_pack_dirstate = parsers.pack_dirstate

    orig_module.pack_dirstate = wrapper
    dirstate._getfsnow = lambda *args: fakenow
    try:
        return func()
    finally:
        orig_module.pack_dirstate = orig_pack_dirstate
        dirstate._getfsnow = orig_dirstate_getfsnow
        if rustmod is not None:
            dirstate.dirstatemap.write = orig_dirstatemap_write


def _poststatusfixup(orig, workingctx, status, fixup):
    ui = workingctx.repo().ui
    return fakewrite(ui, lambda: orig(workingctx, status, fixup))


def markcommitted(orig, committablectx, node):
    ui = committablectx.repo().ui
    return fakewrite(ui, lambda: orig(committablectx, node))


def extsetup(ui):
    extensions.wrapfunction(
        context.workingctx, '_poststatusfixup', _poststatusfixup
    )
    extensions.wrapfunction(context.workingctx, 'markcommitted', markcommitted)