test-sshserver.py
47 lines
| 1.2 KiB
| text/x-python
|
PythonLexer
/ tests / test-sshserver.py
|
r35770 | from __future__ import absolute_import, print_function | ||
import io | ||||
import unittest | ||||
import silenttestrunner | ||||
from mercurial import ( | ||||
|
r35790 | util, | ||
|
r35770 | wireproto, | ||
|
r35877 | wireprotoserver, | ||
|
r35770 | ) | ||
class SSHServerGetArgsTests(unittest.TestCase): | ||||
def testparseknown(self): | ||||
tests = [ | ||||
|
r36230 | (b'* 0\nnodes 0\n', [b'', {}]), | ||
(b'* 0\nnodes 40\n1111111111111111111111111111111111111111\n', | ||||
[b'1111111111111111111111111111111111111111', {}]), | ||||
|
r35770 | ] | ||
for input, expected in tests: | ||||
|
r36230 | self.assertparse(b'known', input, expected) | ||
|
r35770 | |||
def assertparse(self, cmd, input, expected): | ||||
server = mockserver(input) | ||||
_func, spec = wireproto.commands[cmd] | ||||
|
r36082 | self.assertEqual(server._proto.getargs(spec), expected) | ||
|
r35770 | |||
def mockserver(inbytes): | ||||
ui = mockui(inbytes) | ||||
repo = mockrepo(ui) | ||||
|
r35877 | return wireprotoserver.sshserver(ui, repo) | ||
|
r35770 | |||
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() | ||||
if __name__ == '__main__': | ||||
|
r35790 | # Don't call into msvcrt to set BytesIO to binary mode | ||
util.setbinary = lambda fp: True | ||||
|
r35770 | silenttestrunner.main(__name__) | ||