tests/test-lock.py
author Kyle Lippincott <spectral@google.com>
Wed, 27 Jan 2021 10:29:21 -0800
branchstable
changeset 46415 8deab876fb59
parent 45942 89a2afe31e82
child 48875 6000f5b25c9b
permissions -rw-r--r--
wix: tell ComponentSearch that it is finding a directory (not a file) This is to fix an issue we've noticed where fresh installations start at `C:\Program Files\Mercurial`, and then upgrades "walk up" the tree and end up in `C:\Program Files` and finally `C:\` (where they stay). ComponentSearch defaults to finding files, which I think means "it produces a string like `C:\Program Files\Mercurial`", whereas with the type being explicitly a directory, it would return `C:\Program Files\Mercurial\` (note the final trailing backslash). Presumably, a latter step then tries to turn that file name into a proper directory, by removing everything after the last `\`. This could likely also be fixed by actually searching for the component for hg.exe itself. That seemed a lot more complicated, as the GUID for hg.exe isn't known in this file (it's one of the "auto-derived" ones). We could also consider adding a Condition that I think could check the Property and ensure it's either empty or ends in a trailing slash, but that would be an installer runtime check and I'm not convinced it'd actually be useful. This will *not* cause existing installations that are in one of the bad directories to fix themselves. Doing that would require a fair amount more understanding of wix and windows installer than I have, and it *probably* wouldn't be possible to be 100% correct about it either (there's nothing preventing a user from intentionally installing it in C:\, though I don't know why they would do so). If someone wants to tackle fixing existing installations, I think that the first installation is actually the only one that shows up in "Add or Remove Programs", and that its registry keys still exist. You might be able to find something under HKEY_USERS that lists both the "good" and the "bad" InstallDirs. Mine was under `HKEY_USERS\S-1-5-18\Software\Mercurial\InstallDir` (C:\), and `HKEY_USERS\S-1-5-21-..numbers..\Software\Mercurial\InstallDir` (C:\Program Files\Mercurial). If you find exactly two, with one being the default path, and the other being a prefix of it, the user almost certainly hit this bug :D We had originally thought that this bug might be due to unattended installations/upgrades, but I no longer think that's the case. We were able to reproduce the issue by uninstalling all copies of Mercurial I could find, installing one version (it chose the correct location), and then starting the installer for a different version (higher or lower didn't matter). I did not need to deal with an unattended or headless installation/upgrade to trigger the issue, but it's possible that my system was "primed" for this bug to happen because of a previous unattended installation/upgrade. Differential Revision: https://phab.mercurial-scm.org/D9891

from __future__ import absolute_import

import copy
import errno
import tempfile
import types
import unittest

import silenttestrunner

from mercurial import (
    encoding,
    error,
    lock,
    vfs as vfsmod,
)

testlockname = b'testlock'

# work around http://bugs.python.org/issue1515
if types.MethodType not in copy._deepcopy_dispatch:

    def _deepcopy_method(x, memo):
        return type(x)(x.__func__, copy.deepcopy(x.__self__, memo), x.im_class)

    copy._deepcopy_dispatch[types.MethodType] = _deepcopy_method


class lockwrapper(lock.lock):
    def __init__(self, pidoffset, *args, **kwargs):
        # lock.lock.__init__() calls lock(), so the pidoffset assignment needs
        # to be earlier
        self._pidoffset = pidoffset
        super(lockwrapper, self).__init__(*args, **kwargs)

    def _getpid(self):
        return super(lockwrapper, self)._getpid() + self._pidoffset


class teststate(object):
    def __init__(self, testcase, dir, pidoffset=0):
        self._testcase = testcase
        self._acquirecalled = False
        self._releasecalled = False
        self._postreleasecalled = False
        self.vfs = vfsmod.vfs(dir, audit=False)
        self._pidoffset = pidoffset

    def makelock(self, *args, **kwargs):
        l = lockwrapper(
            self._pidoffset,
            self.vfs,
            testlockname,
            releasefn=self.releasefn,
            acquirefn=self.acquirefn,
            *args,
            **kwargs
        )
        l.postrelease.append(self.postreleasefn)
        return l

    def acquirefn(self):
        self._acquirecalled = True

    def releasefn(self):
        self._releasecalled = True

    def postreleasefn(self, success):
        self._postreleasecalled = True

    def assertacquirecalled(self, called):
        self._testcase.assertEqual(
            self._acquirecalled,
            called,
            'expected acquire to be %s but was actually %s'
            % (
                self._tocalled(called),
                self._tocalled(self._acquirecalled),
            ),
        )

    def resetacquirefn(self):
        self._acquirecalled = False

    def assertreleasecalled(self, called):
        self._testcase.assertEqual(
            self._releasecalled,
            called,
            'expected release to be %s but was actually %s'
            % (
                self._tocalled(called),
                self._tocalled(self._releasecalled),
            ),
        )

    def assertpostreleasecalled(self, called):
        self._testcase.assertEqual(
            self._postreleasecalled,
            called,
            'expected postrelease to be %s but was actually %s'
            % (
                self._tocalled(called),
                self._tocalled(self._postreleasecalled),
            ),
        )

    def assertlockexists(self, exists):
        actual = self.vfs.lexists(testlockname)
        self._testcase.assertEqual(
            actual,
            exists,
            'expected lock to %s but actually did %s'
            % (
                self._toexists(exists),
                self._toexists(actual),
            ),
        )

    def _tocalled(self, called):
        if called:
            return 'called'
        else:
            return 'not called'

    def _toexists(self, exists):
        if exists:
            return 'exist'
        else:
            return 'not exist'


class testlock(unittest.TestCase):
    def testlock(self):
        state = teststate(self, tempfile.mkdtemp(dir=encoding.getcwd()))
        lock = state.makelock()
        state.assertacquirecalled(True)
        lock.release()
        state.assertreleasecalled(True)
        state.assertpostreleasecalled(True)
        state.assertlockexists(False)

    def testrecursivelock(self):
        state = teststate(self, tempfile.mkdtemp(dir=encoding.getcwd()))
        lock = state.makelock()
        state.assertacquirecalled(True)

        state.resetacquirefn()
        lock.lock()
        # recursive lock should not call acquirefn again
        state.assertacquirecalled(False)

        lock.release()  # brings lock refcount down from 2 to 1
        state.assertreleasecalled(False)
        state.assertpostreleasecalled(False)
        state.assertlockexists(True)

        lock.release()  # releases the lock
        state.assertreleasecalled(True)
        state.assertpostreleasecalled(True)
        state.assertlockexists(False)

    def testlockfork(self):
        state = teststate(self, tempfile.mkdtemp(dir=encoding.getcwd()))
        lock = state.makelock()
        state.assertacquirecalled(True)

        # fake a fork
        forklock = copy.copy(lock)
        forklock._pidoffset = 1
        forklock.release()
        state.assertreleasecalled(False)
        state.assertpostreleasecalled(False)
        state.assertlockexists(True)

        # release the actual lock
        lock.release()
        state.assertreleasecalled(True)
        state.assertpostreleasecalled(True)
        state.assertlockexists(False)

    def testfrequentlockunlock(self):
        """This tests whether lock acquisition fails as expected, even if
        (1) lock can't be acquired (makelock fails by EEXIST), and
        (2) locker info can't be read in (readlock fails by ENOENT) while
        retrying 5 times.
        """

        d = tempfile.mkdtemp(dir=encoding.getcwd())
        state = teststate(self, d)

        def emulatefrequentlock(*args):
            raise OSError(errno.EEXIST, "File exists")

        def emulatefrequentunlock(*args):
            raise OSError(errno.ENOENT, "No such file or directory")

        state.vfs.makelock = emulatefrequentlock
        state.vfs.readlock = emulatefrequentunlock

        try:
            state.makelock(timeout=0)
            self.fail("unexpected lock acquisition")
        except error.LockHeld as why:
            self.assertTrue(why.errno == errno.ETIMEDOUT)
            self.assertTrue(why.locker == b"")
            state.assertlockexists(False)


if __name__ == '__main__':
    silenttestrunner.main(__name__)