( )⚙ D11973 simplemerge: convert `merge_lines()` away from generator

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D11973

simplemerge: convert `merge_lines()` away from generator
ClosedPublic

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

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHGce8c82a5cd65: simplemerge: convert `merge_lines()` away from generator
rHG3f5bc446357a: simplemerge: convert `merge_lines()` away from generator
Summary

We always consume all the lines and put them in a list anyway. By
making the function not a generator, we can later make it return an
additional value (to indicate if there were conflicts).

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Revision Contents
Changeset List

Diff 31628

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()] ml = [l.strip() for l in m3.merge_lines()[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

name_base=None, name_base=None,
start_marker=b'<<<<<<<', start_marker=b'<<<<<<<',
mid_marker=b'=======', mid_marker=b'=======',
end_marker=b'>>>>>>>', end_marker=b'>>>>>>>',
base_marker=None, base_marker=None,
minimize=False, minimize=False,
): ):
"""Return merge in cvs-like form.""" """Return merge in cvs-like form."""
self.conflicts = False conflicts = False
newline = b'\n' newline = b'\n'
if len(self.a) > 0: if len(self.a) > 0:
if self.a[0].endswith(b'\r\n'): if self.a[0].endswith(b'\r\n'):
newline = b'\r\n' newline = b'\r\n'
elif self.a[0].endswith(b'\r'): elif self.a[0].endswith(b'\r'):
newline = b'\r' newline = b'\r'
if name_a and start_marker: if name_a and start_marker:
start_marker = start_marker + b' ' + name_a start_marker = start_marker + b' ' + name_a
if name_b and end_marker: if name_b and end_marker:
end_marker = end_marker + b' ' + name_b end_marker = end_marker + b' ' + name_b
if name_base and base_marker: if name_base and base_marker:
base_marker = base_marker + b' ' + name_base base_marker = base_marker + b' ' + name_base
merge_groups = self.merge_groups() merge_groups = self.merge_groups()
if minimize: if minimize:
merge_groups = self.minimize(merge_groups) merge_groups = self.minimize(merge_groups)
for what, lines in merge_groups: lines = []
for what, group_lines in merge_groups:
if what == b'conflict': if what == b'conflict':
base_lines, a_lines, b_lines = lines base_lines, a_lines, b_lines = group_lines
self.conflicts = True conflicts = True
if start_marker is not None: if start_marker is not None:
yield start_marker + newline lines.append(start_marker + newline)
for line in a_lines: lines.extend(a_lines)
yield line
if base_marker is not None: if base_marker is not None:
yield base_marker + newline lines.append(base_marker + newline)
for line in base_lines: lines.extend(base_lines)
yield line
if mid_marker is not None: if mid_marker is not None:
yield mid_marker + newline lines.append(mid_marker + newline)
for line in b_lines: lines.extend(b_lines)
yield line
if end_marker is not None: if end_marker is not None:
yield end_marker + newline lines.append(end_marker + newline)
else: else:
for line in lines: lines.extend(group_lines)
yield line 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
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 = list( lines, conflicts = m3.merge_lines(
m3.merge_lines(name_a=name_a, name_b=name_b, **extrakwargs) name_a=name_a, name_b=name_b, **extrakwargs
) )
conflicts = m3.conflicts
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(list(m3.merge_lines()), [b'aaa', b'bbb']) self.assertEqual(m3.merge_lines(), ([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()), b'aaa\nbbb\n222\n') self.assertEqual(b''.join(m3.merge_lines()[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()), b'aaa\nbbb\n222\n') self.assertEqual(b''.join(m3.merge_lines()[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()), b'aaa\nbbb\n222\n') self.assertEqual(b''.join(m3.merge_lines()[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 = m3.merge_lines( ml, conflicts = m3.merge_lines(
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 = m3.merge_lines( ml, conflicts = m3.merge_lines(
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 = m3.merge_lines( ml, conflicts = m3.merge_lines(
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 = list(m3.merge_lines(b'LAO', b'TAO')) ml, conflicts = m3.merge_lines(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 = m3.merge_lines(b'OTHER', b'THIS') m_lines, conflicts = m3.merge_lines(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),
list(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 = m3.merge_lines(b'OTHER', b'THIS') m_lines, conflicts = m3.merge_lines(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),
list(m_lines), m_lines,
) )
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)