tests/test-sshserver.py

localrepo: stop doing special dirstate backup at transaction open

Since the dirstate writes are already managed by the transaction, we already do
a backup of the dirstate when necessary (and even trigger one to keep `hg
rollback` happy).

We needs some special code to deal with the initial empty checkout, but it is
not too complicated.

Managing variable filename (as dirstate-v2 uses) at the "journalfile" level, is
complex and fragile (which is consistent with the fact these files are not
journal…). If we no longer do it, our life is significantly simpler.

In some sense, we apply the xkcd-1134¹ solution to our savebackup/restorebackup
problem.

[1] https://xkcd.com/1134/

(the change to test-hardlink are expect as decreasing the number of duplicated
backup drive the hardlink count down)

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:
    def __init__(self, ui):
        self.ui = ui


class mockui:
    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__)