Mercurial > hg
view tests/test-wireproto.py @ 46326:3e23794b9e1c
run-tests: work around the Windows firewall popup for server processes
Windows doesn't have a `python3` executable, so cc0b332ab9fc attempted to work
around the issue by copying the current python to `python3.exe`. That put it in
`_tmpbindir` because of failures in `test-run-tests.t` when using `_bindir`,
which looked like a process was trying to open it to write out a copy while it
was in use. (Interestingly, I couldn't reproduce this running the test by
itself in a loop for a couple of hours, but it happens constantly when running
all tests.) The problem with using `_tmpbindir` is that it is the randomly
generated path for the test run, and instead of Windows Firewall remembering the
executable signature or image hash when allowing the process to open a server
port, it apparently remembers the image path. That means every run will trigger
a popup to allow it, which is bad for firing off a test run and walking away.
I tried to symlink to the python executable, but that currently requires admin
priviledges[1]. This will prompt the first time if the underlying python binary
has never opened a server port, but appears to avoid it on subsequent runs.
[1] https://bugs.python.org/issue40687
Differential Revision: https://phab.mercurial-scm.org/D9815
author | Matt Harbison <matt_harbison@yahoo.com> |
---|---|
date | Mon, 18 Jan 2021 00:50:01 -0500 |
parents | 2372284d9457 |
children | c424ff4807e6 |
line wrap: on
line source
from __future__ import absolute_import, print_function import sys from mercurial import ( error, pycompat, ui as uimod, util, wireprototypes, wireprotov1peer, wireprotov1server, ) from mercurial.utils import stringutil stringio = util.stringio class proto(object): def __init__(self, args): self.args = args self.name = 'dummyproto' def getargs(self, spec): args = self.args args.setdefault(b'*', {}) names = spec.split() return [args[n] for n in names] def checkperm(self, perm): pass wireprototypes.TRANSPORTS['dummyproto'] = { 'transport': 'dummy', 'version': 1, } class clientpeer(wireprotov1peer.wirepeer): def __init__(self, serverrepo, ui): self.serverrepo = serverrepo self.ui = ui def url(self): return b'test' def local(self): return None def peer(self): return self def canpush(self): return True def close(self): pass def capabilities(self): return [b'batch'] def _call(self, cmd, **args): args = pycompat.byteskwargs(args) res = wireprotov1server.dispatch(self.serverrepo, proto(args), cmd) if isinstance(res, wireprototypes.bytesresponse): return res.data elif isinstance(res, bytes): return res else: raise error.Abort('dummy client does not support response type') def _callstream(self, cmd, **args): return stringio(self._call(cmd, **args)) @wireprotov1peer.batchable def greet(self, name): f = wireprotov1peer.future() yield {b'name': mangle(name)}, f yield unmangle(f.value) class serverrepo(object): def __init__(self, ui): self.ui = ui def greet(self, name): return b"Hello, " + name def filtered(self, name): return self def mangle(s): return b''.join(pycompat.bytechr(ord(c) + 1) for c in pycompat.bytestr(s)) def unmangle(s): return b''.join(pycompat.bytechr(ord(c) - 1) for c in pycompat.bytestr(s)) def greet(repo, proto, name): return mangle(repo.greet(unmangle(name))) wireprotov1server.commands[b'greet'] = (greet, b'name') srv = serverrepo(uimod.ui()) clt = clientpeer(srv, uimod.ui()) def printb(data, end=b'\n'): out = getattr(sys.stdout, 'buffer', sys.stdout) out.write(data + end) out.flush() printb(clt.greet(b"Foobar")) with clt.commandexecutor() as e: fgreet1 = e.callcommand(b'greet', {b'name': b'Fo, =;:<o'}) fgreet2 = e.callcommand(b'greet', {b'name': b'Bar'}) printb( stringutil.pprint([f.result() for f in (fgreet1, fgreet2)], bprefix=True) )