Mercurial > hg
view tests/test-sshserver.py @ 49991:3997c64eff8b
test: use a more direct approach to test racy mutation
The previous test was relying on implementation details and harder to maintain.
The new version is closer to the initial intend : "What happens the file get overwritten from under the current process"
This change is motivated by further changes around the transaction and dirstate
logic that would break the fragile equilibrium that existed before this patch.
Making this change early make it easier to review on its own and remove noise
in future larger changes.
author | Pierre-Yves David <pierre-yves.david@octobus.net> |
---|---|
date | Tue, 07 Feb 2023 13:14:59 +0100 |
parents | 642e31cb55f0 |
children | 13c004b54cbe |
line wrap: on
line source
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__)