tests/test-sshserver.py

changing-files: rework the way we store changed files in side-data

We need to store new data so this is a good opportunity to rework this fully.

1) We directly store the list of affected file in the side data:

* This avoid having to fetch and parse the `files` list in the revision in
  addition to the sidedata. Making the data more self sufficient.

* This work around situation where that `files` field contains wrong
  information, and open the way to other bug fixing (eg: issue6219)

* The format (fixed initial index, sorted files) allow for fast lookup of
  filename within the structure.

* This unify the storage of affected files and copies sources and destination,
  limiting the number filename stored redundantly.

* This prepare for the fact we should drop the `files` as soon as we do any
  change affecting the revision schema.

* This rely on compression to avoid a significant increase of the changelog.d.
  More testing on this will be done before we freeze the final format.


2) We can store additional data:

* The new "merged" field,

* A future "salvaged" set recording files that might have been deleted but have
  were still present in the final result.

Differential Revision: https://phab.mercurial-scm.org/D9090

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__)