view tests/test-sshserver.py @ 45761:341e014fe0fb stable

repoview: only pin obsolete wdir parents while there are unresolved conflicts I noticed after doing an update from an obsolete revision with a dirty wdir that the obsolete commit stayed visible for no obvious reason. It was decided in 85b03b1e4715 not to clear mergestate once all of the conflicts were resolved, in order to allow re-resolving. Since the point of pinning the obsolete parents was to allow resolving in the first place (aaeccdb6e654), it makes sense to also gate it on whether or not there are any remaining files to resolve. This might result in pinning again if files are marked unresolved again, but that seems reasonable, given that it still solves the original issue. Note that this isn't purely cosmetic- pushing with a pinned obsolete revision is likely to cause complaints about pushing multiple heads or other unexpected errors. So the faster it comes out of that state, the better. Differential Revision: https://phab.mercurial-scm.org/D9248
author Matt Harbison <matt_harbison@yahoo.com>
date Fri, 23 Oct 2020 22:20:08 -0400
parents 2372284d9457
children 6000f5b25c9b
line wrap: on
line source

from __future__ import absolute_import, print_function

import io
import unittest

import silenttestrunner

from mercurial import (
    wireprotoserver,
    wireprotov1server,
)

from mercurial.utils import procutil


class SSHServerGetArgsTests(unittest.TestCase):
    def testparseknown(self):
        tests = [
            (b'* 0\nnodes 0\n', [b'', {}]),
            (
                b'* 0\nnodes 40\n1111111111111111111111111111111111111111\n',
                [b'1111111111111111111111111111111111111111', {}],
            ),
        ]
        for input, expected in tests:
            self.assertparse(b'known', input, expected)

    def assertparse(self, cmd, input, expected):
        server = mockserver(input)
        proto = wireprotoserver.sshv1protocolhandler(
            server._ui, server._fin, server._fout
        )
        _func, spec = wireprotov1server.commands[cmd]
        self.assertEqual(proto.getargs(spec), expected)


def mockserver(inbytes):
    ui = mockui(inbytes)
    repo = mockrepo(ui)
    return wireprotoserver.sshserver(ui, repo)


class mockrepo(object):
    def __init__(self, ui):
        self.ui = ui


class mockui(object):
    def __init__(self, inbytes):
        self.fin = io.BytesIO(inbytes)
        self.fout = io.BytesIO()
        self.ferr = io.BytesIO()

    def protectfinout(self):
        return self.fin, self.fout

    def restorefinout(self, fin, fout):
        pass


if __name__ == '__main__':
    # Don't call into msvcrt to set BytesIO to binary mode
    procutil.setbinary = lambda fp: True
    silenttestrunner.main(__name__)