Skip to content
Snippets Groups Projects
Commit c8377fdf6dca authored by Georges Racinet's avatar Georges Racinet
Browse files

Merge from heptapod-0-7 for pull-force-topic command

No related branches found
No related tags found
No related merge requests found
4787dfb1dff1676e2eb5d494d15e33a2895db068 0.7.0
2c7cd4b77da16348144696b8570f5a8014a16785 0.7.1
......@@ -58,6 +58,11 @@
ui = make_ui(base_ui, config=config)
return cls(hg.repository(ui, path))
def command(self, name, *args, **kwargs):
cmd = cmdutil.findcmd(name, commands.table)[1][0]
repo = self.repo
return cmd(repo.ui, repo, *args, **kwargs)
def write_commit(self, rpath,
content=None, message=None,
return_ctx=False,
......
......@@ -6,5 +6,5 @@
Server side Heptapod extension.
This extension should enclose all Mercurial modifications and commands
needed in the Heptapod context.
needed for Heptapod server operations.
"""
......@@ -10,1 +10,136 @@
"""
import os
import tempfile
from mercurial.i18n import _
from mercurial import (
bundlerepo,
bundle2,
cmdutil,
error,
exchange,
hg,
registrar,
scmutil,
util,
)
from hgext3rd.topic import _changetopics
if util.safehasattr(registrar, 'configitem'):
configtable = {}
configitem = registrar.configitem(configtable)
configitem('heptapod', 'repositories-root')
configitem('heptapod', 'gitlab-shell')
configitem('heptapod', 'mirror-path')
cmdtable = {}
command = registrar.command(cmdtable)
@command(
"pull-force-topic",
[('f', 'force', None, _('run even when remote repository is unrelated')),
('r', 'rev', [], _('a remote changeset intended to be imported'),
_('REV')),
] + cmdutil.remoteopts,
_('[-r] [-f] TARGET_TOPIC')
)
def pull_force_topic(ui, repo, topic, source="default",
force=False, **opts):
"""Pull changesets from remote, forcing them to drafts with given topic.
This is intended to import pull requests from an external system, such
as Bitbucket. In many case, the changesets to import would have been
made in a private fork, and could be public, most commonly shadowing the
default branch.
TARGET_TOPIC is the topic to set on the pulled changesets
"""
pull_rev = opts.get('rev')
ui.status("Pulling%s from %r, forcing new changesets to drafts with "
"topic %r\n" % ('' if pull_rev is None else ' %r' % pull_rev,
source, topic))
topic = topic.strip()
if not topic:
raise error.Abort(
_("topic name cannot consist entirely of whitespace"))
scmutil.checknewlabel(repo, topic, 'topic')
other = hg.peer(repo, opts, source)
branches = (None, ()) # we want to work on precise revs only
revs, checkout = hg.addbranchrevs(repo, other, branches, pull_rev)
if revs:
revs = [other.lookup(rev) for rev in revs]
fd, bundlepath = tempfile.mkstemp(prefix='pull-force-topic')
# it's a bit unfortunate that we wouldn't get a bundlerepo without
# 'bundlename' for local repos as it wouldn't work on tests or be
# really different than real usage.
other, inc_nodes, cleanupfn = bundlerepo.getremotechanges(
ui, repo, other,
onlyheads=revs,
force=opts.get('force'),
bundlename=bundlepath,
)
if not inc_nodes:
ui.status(_("Empty pull, nothing to do\n"))
cleanupfn()
return
try:
bundle = os.fdopen(fd, 'rb')
with repo.lock():
with repo.transaction('pull-force-topic') as txn:
unbundle(ui, repo, bundle, txn)
change_topic(ui, repo, topic, inc_nodes)
finally:
cleanupfn()
cleanup_tmp_bundle(ui, bundle, bundlepath)
def cleanup_tmp_bundle(ui, fobj, path):
try:
os.unlink(path)
fobj.close()
except Exception as exc:
ui.warn("Got an error while cleaning up %r: %r\n" % (
path, exc))
def unbundle(ui, repo, bundle, transaction):
"""Unbundler for pull-force-topic.
inspired from `commands.unbundle()`, much simplified to run
in an already provided transaction, and have only what we need.
:param bundle: file-like object
:param txn: Mercurial transaction
"""
# we probaly already made sure in negociation with remote
# that we don't risk getting BundleUnknownFeatureError
# on a bundle we generated ourselves
gen = exchange.readbundle(ui, bundle, None)
op = bundle2.applybundle(repo, gen, transaction,
source='pull-force-topic')
bundle2.combinechangegroupresults(op)
# commands.postincoming is for user feedback and wdir update
def non_obsolete_revs(repo, nodes):
rev = repo.changelog.rev
for node in nodes:
try:
yield rev(node)
except error.FilteredLookupError:
pass
def change_topic(ui, repo, topic, nodes):
revset = repo.revs('%ld', non_obsolete_revs(repo, nodes))
if repo.revs('%ld and public()', revset):
raise error.Abort("can't change topic of a public change")
_changetopics(ui, repo, revset, topic)
# python package
from __future__ import absolute_import
import pytest
from mercurial import (
error,
node,
)
from heptapod.testhelpers import (
LocalRepoWrapper,
make_ui,
)
import hgext3rd.heptapod
from .. import (
change_topic,
cleanup_tmp_bundle,
)
HGEXT_HEPTA_SOURCE = hgext3rd.heptapod.__file__.replace('.pyc', '.py')
def make_main_repo(path):
"""Make repo with 2 public revs; return wrapper, ctx of rev 0
The returned ctx is for the first changeset because we'll use it as
a branching point, hence more often than the second.
"""
wrapper = LocalRepoWrapper.init(
path,
config=dict(
phases=dict(publish=False),
extensions=dict(heptapod=HGEXT_HEPTA_SOURCE,
topic='',
evolve='',
)
))
ctx = wrapper.write_commit('foo', content='foo0', message="default0",
return_ctx=True)
wrapper.write_commit('foo', content='foo1', message="default1")
wrapper.set_phase('public', ['.'])
return wrapper, ctx
def test_force_draft(tmpdir):
"""Case where the remote changesets are public."""
main_path = tmpdir.join('main')
fork_path = tmpdir.join('fork')
main, base_ctx = make_main_repo(main_path)
fork = LocalRepoWrapper.init(fork_path)
fork.command('pull', str(main_path), rev=[base_ctx.hex()])
fork_node = fork.write_commit('bar', content='bar1', message="in fork 1",
parent=base_ctx.node())
fork.set_phase('public', ['.'])
main.command('pull-force-topic', 'zetop',
source=str(fork_path), rev=[node.hex(fork_node)])
# TODO better lookup ?
imported = main.repo['tip']
assert imported.topic() == 'zetop' # implied draft
assert imported.description() == 'in fork 1'
def test_change_topic_sanity(tmpdir):
wrapper, base_ctx = make_main_repo(tmpdir)
repo = wrapper.repo
# early validation:
with pytest.raises(error.Abort) as exc_info:
wrapper.command('pull-force-topic', ' ')
assert 'entirely of whitespace' in exc_info.value.args[0]
with pytest.raises(error.Abort) as exc_info:
wrapper.command('pull-force-topic', 'tip')
assert 'reserved' in exc_info.value.args[0]
# inner check done in transaction
with pytest.raises(error.Abort) as exc_info:
change_topic(repo.ui, repo, 'sometopic', [base_ctx.node()])
assert 'public' in exc_info.value.args[0]
def test_cleanup_tmp_bundle_exc(tmpdir):
ui = make_ui(None)
path = tmpdir.join('bundle')
# no file at given path
cleanup_tmp_bundle(ui, None, str(path))
path.write('foo')
# close fails (of course, fobj is None)
cleanup_tmp_bundle(ui, None, str(path))
def test_empy_pull(tmpdir):
"""Case where the pull is actually empty
This can happen if a remote pull-request is still open
while effectively merged, or if undetected duplicates are imported one
after the other
"""
main_path = tmpdir.join('main')
fork_path = tmpdir.join('fork')
main, base_ctx = make_main_repo(main_path)
fork = LocalRepoWrapper.init(fork_path)
fork.command('pull', str(main_path), rev=[base_ctx.hex()])
main.command('pull-force-topic', 'zetop', source=str(fork_path))
def test_already_obsolete(tmpdir):
"""Case where a remote changeset is locally obsolete
This can happen in practice if two remote pulls are stacked and
we already pulled one, making it actually obsolete because of the
topic forcing.
"""
main_path = tmpdir.join('main')
fork_path = tmpdir.join('fork')
main = make_main_repo(main_path)[0]
amended = main.write_commit('foo', content='foo2', return_ctx=True)
fork = LocalRepoWrapper.init(fork_path)
fork.command('pull', str(main_path), rev=[amended.hex()])
main_path.join('foo').write('amended')
main.command('amend', message="required in CI environment")
main.command('pull-force-topic', 'zetop',
source=str(fork_path), rev=[amended.hex()])
from __future__ import absolute_import
import pytest
from mercurial import (
error,
node,
)
from heptapod.testhelpers import (
LocalRepoWrapper,
make_ui,
)
import hgext3rd.heptapod
from .. import (
change_topic,
cleanup_tmp_bundle,
)
HGEXT_HEPTA_SOURCE = hgext3rd.heptapod.__file__.replace('.pyc', '.py')
def make_main_repo(path):
"""Make repo with 2 public revs; return wrapper, ctx of rev 0
The returned ctx is for the first changeset because we'll use it as
a branching point, hence more often than the second.
"""
wrapper = LocalRepoWrapper.init(
path,
config=dict(
phases=dict(publish=False),
extensions=dict(heptapod=HGEXT_HEPTA_SOURCE,
topic='',
evolve='',
)
))
ctx = wrapper.write_commit('foo', content='foo0', message="default0",
return_ctx=True)
wrapper.write_commit('foo', content='foo1', message="default1")
wrapper.set_phase('public', ['.'])
return wrapper, ctx
def test_force_draft(tmpdir):
"""Case where the remote changesets are public."""
main_path = tmpdir.join('main')
fork_path = tmpdir.join('fork')
main, base_ctx = make_main_repo(main_path)
fork = LocalRepoWrapper.init(fork_path)
fork.command('pull', str(main_path), rev=[base_ctx.hex()])
fork_node = fork.write_commit('bar', content='bar1', message="in fork 1",
parent=base_ctx.node())
fork.set_phase('public', ['.'])
main.command('pull-force-topic', 'zetop',
source=str(fork_path), rev=[node.hex(fork_node)])
# TODO better lookup ?
imported = main.repo['tip']
assert imported.topic() == 'zetop' # implied draft
assert imported.description() == 'in fork 1'
def test_change_topic_sanity(tmpdir):
wrapper, base_ctx = make_main_repo(tmpdir)
repo = wrapper.repo
# early validation:
with pytest.raises(error.Abort) as exc_info:
wrapper.command('pull-force-topic', ' ')
assert 'entirely of whitespace' in exc_info.value.args[0]
with pytest.raises(error.Abort) as exc_info:
wrapper.command('pull-force-topic', 'tip')
assert 'reserved' in exc_info.value.args[0]
# inner check done in transaction
with pytest.raises(error.Abort) as exc_info:
change_topic(repo.ui, repo, 'sometopic', [base_ctx.node()])
assert 'public' in exc_info.value.args[0]
def test_cleanup_tmp_bundle_exc(tmpdir):
ui = make_ui(None)
path = tmpdir.join('bundle')
# no file at given path
cleanup_tmp_bundle(ui, None, str(path))
path.write('foo')
# close fails (of course, fobj is None)
cleanup_tmp_bundle(ui, None, str(path))
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment