annotate tests/test-sshserver.py @ 42618:c17e6a3e7356

phabricator: handle local:commits time being string or int

When setting local:commits arcanist has different behaviour depending on
whether the repo is git or hg. With hg it sets the time as a number, since it
calls PHP's strtotime on the value, but with git it sets it as a string.
Normally this wouldn't be an issue since phabread wouldn't be interacting with
Phabricator Revisions for git repos, but Mozilla has a secondary workflow for
git users that uses the git-cinnabar tool to interact with their hg repos. When
a git-cinnabar user uses the moz-phab tool to submit patches for mozilla-central
it makes use of Mozilla's fork of arcanist, which works with their local git
version of m-c, and thus sets the local:commit time as a string, and then
translates the commit hashes.

Currently when encountering such DREVS phabread dies with "TypeError: %d format:
a number is required, not str".

phabsend also used to set it as a string but wouldn't have encountered the
issue with its own DREVs since it would read hg:meta first.

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

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