view tests/test-ctxmanager.py @ 28240:1ac8ce137377

changegroup: fix treemanifests on merges

The current code for generating treemanifest revisions takes the list
of files in the changeset and finds the directories from them. This
does not work for merges, since a merge may pick file A from one side
and file B from another and neither of them would appear in the
changeset's "files" list, but the manifest would still change.

Fix this by instead walking the root manifest log for all needed
revisions, storing all needed file and subdirectory revisions, then
recursively visiting the subdirectories. This also turns out to be
faster: cloning a version of hg core converted to treemanifests went
from ~28s to ~19s (timing somewhat unfair: before this patch, timed
until crash; after this patch, timed until manifests complete).

The new algorithm is used only on treemanifest repos. Although it
works equally well on flat manifests, we leave the iteration over
files in the changeset for flat manifests for now.

from __future__ import absolute_import

import silenttestrunner
import unittest

from mercurial.util import ctxmanager

class contextmanager(object):
    def __init__(self, name, trace):
        self.name = name
        self.entered = False
        self.exited = False
        self.trace = trace

    def __enter__(self):
        self.entered = True
        self.trace(('enter', self.name))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.exited = exc_type, exc_val, exc_tb
        self.trace(('exit', self.name))

    def __repr__(self):
        return '<ctx %r>' % self.name

class ctxerror(Exception):
    pass

class raise_on_enter(contextmanager):
    def __enter__(self):
        self.trace(('raise', self.name))
        raise ctxerror(self.name)

class raise_on_exit(contextmanager):
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.trace(('raise', self.name))
        raise ctxerror(self.name)

def ctxmgr(name, trace):
    return lambda: contextmanager(name, trace)

class test_ctxmanager(unittest.TestCase):
    def test_basics(self):
        trace = []
        addtrace = trace.append
        with ctxmanager(ctxmgr('a', addtrace), ctxmgr('b', addtrace)) as c:
            a, b = c.enter()
            c.atexit(addtrace, ('atexit', 'x'))
            c.atexit(addtrace, ('atexit', 'y'))
        self.assertEqual(trace, [('enter', 'a'), ('enter', 'b'),
                                 ('atexit', 'y'), ('atexit', 'x'),
                                 ('exit', 'b'), ('exit', 'a')])

    def test_raise_on_enter(self):
        trace = []
        addtrace = trace.append
        def go():
            with ctxmanager(ctxmgr('a', addtrace),
                           lambda: raise_on_enter('b', addtrace)) as c:
                c.enter()
                addtrace('unreachable')
        self.assertRaises(ctxerror, go)
        self.assertEqual(trace, [('enter', 'a'), ('raise', 'b'), ('exit', 'a')])

    def test_raise_on_exit(self):
        trace = []
        addtrace = trace.append
        def go():
            with ctxmanager(ctxmgr('a', addtrace),
                           lambda: raise_on_exit('b', addtrace)) as c:
                c.enter()
                addtrace('running')
        self.assertRaises(ctxerror, go)
        self.assertEqual(trace, [('enter', 'a'), ('enter', 'b'), 'running',
                                 ('raise', 'b'), ('exit', 'a')])

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