This is an archive of the discontinued Mercurial Phabricator instance.

Differential D11974

simplemerge: make merge_lines() a free function
ClosedPublic

Authored by martinvonz on Jan 12 2022, 1:55 AM.

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHGc6649c53073f: simplemerge: make merge_lines() a free function
rHG912668d93604: simplemerge: make merge_lines() a free function
Summary

IMO, the rendering of a conflict as text with merge markers should be
separate from the code for resolving conflicts. The latter is what
Merge3Text is mostly about already.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Revision Contents
Changeset List

Diff 31668

mercurial/debugcommands.py

p1 = repo[ps[0]] p1 = repo[ps[0]]
if len(ps) > 1: if len(ps) > 1:
p2 = repo[ps[1]] p2 = repo[ps[1]]
pa = p1.ancestor(p2) pa = p1.ancestor(p2)
base, local, other = [ base, local, other = [
x[fn].data() for x in (pa, p1, p2) x[fn].data() for x in (pa, p1, p2)
] ]
m3 = simplemerge.Merge3Text(base, local, other) m3 = simplemerge.Merge3Text(base, local, other)
ml = [l.strip() for l in m3.merge_lines()[0]] ml = [l.strip() for l in simplemerge.merge_lines(m3)[0]]
ml.append(b"") ml.append(b"")
elif at > 0: elif at > 0:
ml = p1[fn].data().split(b"\n") ml = p1[fn].data().split(b"\n")
else: else:
ml = initialmergedlines ml = initialmergedlines
ml[id * linesperrev] += b" r%i" % id ml[id * linesperrev] += b" r%i" % id
mergedtext = b"\n".join(ml) mergedtext = b"\n".join(ml)
files.append(fn) files.append(fn)

mercurial/simplemerge.py

if a is None: if a is None:
a = mdiff.splitnewlines(atext) a = mdiff.splitnewlines(atext)
if b is None: if b is None:
b = mdiff.splitnewlines(btext) b = mdiff.splitnewlines(btext)
self.base = base self.base = base
self.a = a self.a = a
self.b = b self.b = b
def merge_lines(
self,
name_a=None,
name_b=None,
name_base=None,
start_marker=b'<<<<<<<',
mid_marker=b'=======',
end_marker=b'>>>>>>>',
base_marker=None,
minimize=False,
):
"""Return merge in cvs-like form."""
conflicts = False
newline = b'\n'
if len(self.a) > 0:
if self.a[0].endswith(b'\r\n'):
newline = b'\r\n'
elif self.a[0].endswith(b'\r'):
newline = b'\r'
if name_a and start_marker:
start_marker = start_marker + b' ' + name_a
if name_b and end_marker:
end_marker = end_marker + b' ' + name_b
if name_base and base_marker:
base_marker = base_marker + b' ' + name_base
merge_groups = self.merge_groups()
if minimize:
merge_groups = self.minimize(merge_groups)
lines = []
for what, group_lines in merge_groups:
if what == b'conflict':
base_lines, a_lines, b_lines = group_lines
conflicts = True
if start_marker is not None:
lines.append(start_marker + newline)
lines.extend(a_lines)
if base_marker is not None:
lines.append(base_marker + newline)
lines.extend(base_lines)
if mid_marker is not None:
lines.append(mid_marker + newline)
lines.extend(b_lines)
if end_marker is not None:
lines.append(end_marker + newline)
else:
lines.extend(group_lines)
return lines, conflicts
def merge_groups(self): def merge_groups(self):
"""Yield sequence of line groups. Each one is a tuple: """Yield sequence of line groups. Each one is a tuple:
'unchanged', lines 'unchanged', lines
Lines unchanged from base Lines unchanged from base
'a', lines 'a', lines
Lines taken from a Lines taken from a
if len(overrides) > 3: if len(overrides) > 3:
raise error.Abort(_(b"can only specify three labels.")) raise error.Abort(_(b"can only specify three labels."))
result = defaults[:] result = defaults[:]
for i, override in enumerate(overrides): for i, override in enumerate(overrides):
result[i] = override result[i] = override
return result return result
def merge_lines(
m3,
name_a=None,
name_b=None,
name_base=None,
start_marker=b'<<<<<<<',
mid_marker=b'=======',
end_marker=b'>>>>>>>',
base_marker=None,
minimize=False,
):
"""Return merge in cvs-like form."""
conflicts = False
newline = b'\n'
if len(m3.a) > 0:
if m3.a[0].endswith(b'\r\n'):
newline = b'\r\n'
elif m3.a[0].endswith(b'\r'):
newline = b'\r'
if name_a and start_marker:
start_marker = start_marker + b' ' + name_a
if name_b and end_marker:
end_marker = end_marker + b' ' + name_b
if name_base and base_marker:
base_marker = base_marker + b' ' + name_base
merge_groups = m3.merge_groups()
if minimize:
merge_groups = m3.minimize(merge_groups)
lines = []
for what, group_lines in merge_groups:
if what == b'conflict':
base_lines, a_lines, b_lines = group_lines
conflicts = True
if start_marker is not None:
lines.append(start_marker + newline)
lines.extend(a_lines)
if base_marker is not None:
lines.append(base_marker + newline)
lines.extend(base_lines)
if mid_marker is not None:
lines.append(mid_marker + newline)
lines.extend(b_lines)
if end_marker is not None:
lines.append(end_marker + newline)
else:
lines.extend(group_lines)
return lines, conflicts
def _mergediff(m3, name_a, name_b, name_base): def _mergediff(m3, name_a, name_b, name_base):
lines = [] lines = []
conflicts = False conflicts = False
for what, group_lines in m3.merge_groups(): for what, group_lines in m3.merge_groups():
if what == b'conflict': if what == b'conflict':
base_lines, a_lines, b_lines = group_lines base_lines, a_lines, b_lines = group_lines
base_text = b''.join(base_lines) base_text = b''.join(base_lines)
b_blocks = list( b_blocks = list(
else: else:
extrakwargs = { extrakwargs = {
'minimize': True, 'minimize': True,
} }
if mode == b'merge3': if mode == b'merge3':
extrakwargs['base_marker'] = b'|||||||' extrakwargs['base_marker'] = b'|||||||'
extrakwargs['name_base'] = name_base extrakwargs['name_base'] = name_base
extrakwargs['minimize'] = False extrakwargs['minimize'] = False
lines, conflicts = m3.merge_lines( lines, conflicts = merge_lines(
name_a=name_a, name_b=name_b, **extrakwargs m3, name_a=name_a, name_b=name_b, **extrakwargs
) )
mergedtext = b''.join(lines) mergedtext = b''.join(lines)
if opts.get('print'): if opts.get('print'):
ui.fout.write(mergedtext) ui.fout.write(mergedtext)
else: else:
# localctx.flags() already has the merged flags (done in # localctx.flags() already has the merged flags (done in
# mergestate.resolve()) # mergestate.resolve())
localctx.write(mergedtext, localctx.flags()) localctx.write(mergedtext, localctx.flags())
if conflicts: if conflicts:
return 1 return 1

tests/test-simplemerge.py

def test_null_insert(self): def test_null_insert(self):
m3 = Merge3([], [b'aaa', b'bbb'], []) m3 = Merge3([], [b'aaa', b'bbb'], [])
# todo: should use a sentinel at end as from get_matching_blocks # todo: should use a sentinel at end as from get_matching_blocks
# to match without zz # to match without zz
self.assertEqual(list(m3.find_sync_regions()), [(0, 0, 2, 2, 0, 0)]) self.assertEqual(list(m3.find_sync_regions()), [(0, 0, 2, 2, 0, 0)])
self.assertEqual(list(m3.merge_regions()), [(b'a', 0, 2)]) self.assertEqual(list(m3.merge_regions()), [(b'a', 0, 2)])
self.assertEqual(m3.merge_lines(), ([b'aaa', b'bbb'], False)) self.assertEqual(simplemerge.merge_lines(m3), ([b'aaa', b'bbb'], False))
def test_no_conflicts(self): def test_no_conflicts(self):
"""No conflicts because only one side changed""" """No conflicts because only one side changed"""
m3 = Merge3( m3 = Merge3(
[b'aaa', b'bbb'], [b'aaa', b'111', b'bbb'], [b'aaa', b'bbb'] [b'aaa', b'bbb'], [b'aaa', b'111', b'bbb'], [b'aaa', b'bbb']
) )
self.assertEqual( self.assertEqual(
list(m3.find_sync_regions()), list(m3.find_sync_regions()),
[(0, 1, 0, 1, 0, 1), (1, 2, 2, 3, 1, 2), (2, 2, 3, 3, 2, 2)], [(0, 1, 0, 1, 0, 1), (1, 2, 2, 3, 1, 2), (2, 2, 3, 3, 2, 2)],
) )
self.assertEqual( self.assertEqual(
list(m3.merge_regions()), list(m3.merge_regions()),
[(b'unchanged', 0, 1), (b'a', 1, 2), (b'unchanged', 1, 2)], [(b'unchanged', 0, 1), (b'a', 1, 2), (b'unchanged', 1, 2)],
) )
def test_append_a(self): def test_append_a(self):
m3 = Merge3( m3 = Merge3(
[b'aaa\n', b'bbb\n'], [b'aaa\n', b'bbb\n'],
[b'aaa\n', b'bbb\n', b'222\n'], [b'aaa\n', b'bbb\n', b'222\n'],
[b'aaa\n', b'bbb\n'], [b'aaa\n', b'bbb\n'],
) )
self.assertEqual(b''.join(m3.merge_lines()[0]), b'aaa\nbbb\n222\n') self.assertEqual(
b''.join(simplemerge.merge_lines(m3)[0]), b'aaa\nbbb\n222\n'
)
def test_append_b(self): def test_append_b(self):
m3 = Merge3( m3 = Merge3(
[b'aaa\n', b'bbb\n'], [b'aaa\n', b'bbb\n'],
[b'aaa\n', b'bbb\n'], [b'aaa\n', b'bbb\n'],
[b'aaa\n', b'bbb\n', b'222\n'], [b'aaa\n', b'bbb\n', b'222\n'],
) )
self.assertEqual(b''.join(m3.merge_lines()[0]), b'aaa\nbbb\n222\n') self.assertEqual(
b''.join(simplemerge.merge_lines(m3)[0]), b'aaa\nbbb\n222\n'
)
def test_append_agreement(self): def test_append_agreement(self):
m3 = Merge3( m3 = Merge3(
[b'aaa\n', b'bbb\n'], [b'aaa\n', b'bbb\n'],
[b'aaa\n', b'bbb\n', b'222\n'], [b'aaa\n', b'bbb\n', b'222\n'],
[b'aaa\n', b'bbb\n', b'222\n'], [b'aaa\n', b'bbb\n', b'222\n'],
) )
self.assertEqual(b''.join(m3.merge_lines()[0]), b'aaa\nbbb\n222\n') self.assertEqual(
b''.join(simplemerge.merge_lines(m3)[0]), b'aaa\nbbb\n222\n'
)
def test_append_clash(self): def test_append_clash(self):
m3 = Merge3( m3 = Merge3(
[b'aaa\n', b'bbb\n'], [b'aaa\n', b'bbb\n'],
[b'aaa\n', b'bbb\n', b'222\n'], [b'aaa\n', b'bbb\n', b'222\n'],
[b'aaa\n', b'bbb\n', b'333\n'], [b'aaa\n', b'bbb\n', b'333\n'],
) )
ml, conflicts = m3.merge_lines( ml, conflicts = simplemerge.merge_lines(
m3,
name_a=b'a', name_a=b'a',
name_b=b'b', name_b=b'b',
start_marker=b'<<', start_marker=b'<<',
mid_marker=b'--', mid_marker=b'--',
end_marker=b'>>', end_marker=b'>>',
) )
self.assertEqual( self.assertEqual(
b''.join(ml), b''.join(ml),
b'aaa\n' b'bbb\n' b'<< a\n' b'222\n' b'--\n' b'333\n' b'>> b\n', b'aaa\n' b'bbb\n' b'<< a\n' b'222\n' b'--\n' b'333\n' b'>> b\n',
) )
def test_insert_agreement(self): def test_insert_agreement(self):
m3 = Merge3( m3 = Merge3(
[b'aaa\n', b'bbb\n'], [b'aaa\n', b'bbb\n'],
[b'aaa\n', b'222\n', b'bbb\n'], [b'aaa\n', b'222\n', b'bbb\n'],
[b'aaa\n', b'222\n', b'bbb\n'], [b'aaa\n', b'222\n', b'bbb\n'],
) )
ml, conflicts = m3.merge_lines( ml, conflicts = simplemerge.merge_lines(
m3,
name_a=b'a', name_a=b'a',
name_b=b'b', name_b=b'b',
start_marker=b'<<', start_marker=b'<<',
mid_marker=b'--', mid_marker=b'--',
end_marker=b'>>', end_marker=b'>>',
) )
self.assertEqual(b''.join(ml), b'aaa\n222\nbbb\n') self.assertEqual(b''.join(ml), b'aaa\n222\nbbb\n')
list(m3.merge_groups()), list(m3.merge_groups()),
[ [
(b'unchanged', [b'aaa\n']), (b'unchanged', [b'aaa\n']),
(b'conflict', ([], [b'111\n'], [b'222\n'])), (b'conflict', ([], [b'111\n'], [b'222\n'])),
(b'unchanged', [b'bbb\n']), (b'unchanged', [b'bbb\n']),
], ],
) )
ml, conflicts = m3.merge_lines( ml, conflicts = simplemerge.merge_lines(
m3,
name_a=b'a', name_a=b'a',
name_b=b'b', name_b=b'b',
start_marker=b'<<', start_marker=b'<<',
mid_marker=b'--', mid_marker=b'--',
end_marker=b'>>', end_marker=b'>>',
) )
self.assertEqual( self.assertEqual(
b''.join(ml), b''.join(ml),
self.assertEqual( self.assertEqual(
list(m3.find_sync_regions()), list(m3.find_sync_regions()),
[(0, 1, 0, 1, 0, 1), (3, 4, 4, 5, 5, 6), (4, 4, 5, 5, 6, 6)], [(0, 1, 0, 1, 0, 1), (3, 4, 4, 5, 5, 6), (4, 4, 5, 5, 6, 6)],
) )
def test_merge_poem(self): def test_merge_poem(self):
"""Test case from diff3 manual""" """Test case from diff3 manual"""
m3 = Merge3(TZU, LAO, TAO) m3 = Merge3(TZU, LAO, TAO)
ml, conflicts = m3.merge_lines(b'LAO', b'TAO') ml, conflicts = simplemerge.merge_lines(m3, b'LAO', b'TAO')
self.log(b'merge result:') self.log(b'merge result:')
self.log(b''.join(ml)) self.log(b''.join(ml))
self.assertEqual(ml, MERGED_RESULT) self.assertEqual(ml, MERGED_RESULT)
def test_binary(self): def test_binary(self):
with self.assertRaises(error.Abort): with self.assertRaises(error.Abort):
Merge3([b'\x00'], [b'a'], [b'b']) Merge3([b'\x00'], [b'a'], [b'b'])
def test_dos_text(self): def test_dos_text(self):
base_text = b'a\r\n' base_text = b'a\r\n'
this_text = b'b\r\n' this_text = b'b\r\n'
other_text = b'c\r\n' other_text = b'c\r\n'
m3 = Merge3( m3 = Merge3(
base_text.splitlines(True), base_text.splitlines(True),
other_text.splitlines(True), other_text.splitlines(True),
this_text.splitlines(True), this_text.splitlines(True),
) )
m_lines, conflicts = m3.merge_lines(b'OTHER', b'THIS') m_lines, conflicts = simplemerge.merge_lines(m3, b'OTHER', b'THIS')
self.assertEqual( self.assertEqual(
b'<<<<<<< OTHER\r\nc\r\n=======\r\nb\r\n' b'<<<<<<< OTHER\r\nc\r\n=======\r\nb\r\n'
b'>>>>>>> THIS\r\n'.splitlines(True), b'>>>>>>> THIS\r\n'.splitlines(True),
m_lines, m_lines,
) )
def test_mac_text(self): def test_mac_text(self):
base_text = b'a\r' base_text = b'a\r'
this_text = b'b\r' this_text = b'b\r'
other_text = b'c\r' other_text = b'c\r'
m3 = Merge3( m3 = Merge3(
base_text.splitlines(True), base_text.splitlines(True),
other_text.splitlines(True), other_text.splitlines(True),
this_text.splitlines(True), this_text.splitlines(True),
) )
m_lines, conflicts = m3.merge_lines(b'OTHER', b'THIS') m_lines, conflicts = simplemerge.merge_lines(m3, b'OTHER', b'THIS')
self.assertEqual( self.assertEqual(
b'<<<<<<< OTHER\rc\r=======\rb\r' b'<<<<<<< OTHER\rc\r=======\rb\r'
b'>>>>>>> THIS\r'.splitlines(True), b'>>>>>>> THIS\r'.splitlines(True),
m_lines, m_lines,
) )
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)