1import execnet 2import py 3import pytest 4import sys 5import subprocess 6from execnet.gateway_base import get_execmodel, WorkerPool 7 8collect_ignore = ['build', 'doc/_build'] 9 10rsyncdirs = ['conftest.py', 'execnet', 'testing', 'doc'] 11 12winpymap = { 13 'python2.7': r'C:\Python27\python.exe', 14 'python3.4': r'C:\Python34\python.exe', 15} 16 17 18@pytest.hookimpl(hookwrapper=True) 19def pytest_runtest_setup(item): 20 if item.fspath.purebasename in ('test_group', 'test_info'): 21 getspecssh(item.config) # will skip if no gx given 22 yield 23 if 'pypy' in item.keywords and not item.config.option.pypy: 24 py.test.skip("pypy tests skipped, use --pypy to run them.") 25 26 27@pytest.fixture 28def makegateway(request): 29 group = execnet.Group() 30 request.addfinalizer(lambda: group.terminate(0.5)) 31 return group.makegateway 32 33pytest_plugins = ['pytester', 'doctest'] 34 35 36# configuration information for tests 37def pytest_addoption(parser): 38 group = parser.getgroup("execnet", "execnet testing options") 39 group.addoption( 40 '--gx', action="append", dest="gspecs", default=None, 41 help="add a global test environment, XSpec-syntax. ") 42 group.addoption( 43 '--pypy', action="store_true", dest="pypy", 44 help="run some tests also against pypy") 45 group.addoption( 46 '--broken-isp', action="store_true", dest="broken_isp", 47 help=("Skips tests that assume your ISP doesn't put up a landing " 48 "page on invalid addresses")) 49 50 51@pytest.fixture 52def specssh(request): 53 return getspecssh(request.config) 54 55 56@pytest.fixture 57def specsocket(request): 58 return getsocketspec(request.config) 59 60 61def getgspecs(config=None): 62 if config is None: 63 config = py.test.config 64 return map(execnet.XSpec, config.getvalueorskip("gspecs")) 65 66 67def getspecssh(config=None): 68 xspecs = getgspecs(config) 69 for spec in xspecs: 70 if spec.ssh: 71 if not py.path.local.sysfind("ssh"): 72 py.test.skip("command not found: ssh") 73 return spec 74 py.test.skip("need '--gx ssh=...'") 75 76 77def getsocketspec(config=None): 78 xspecs = getgspecs(config) 79 for spec in xspecs: 80 if spec.socket: 81 return spec 82 py.test.skip("need '--gx socket=...'") 83 84 85def pytest_generate_tests(metafunc): 86 if 'gw' in metafunc.funcargnames: 87 assert 'anypython' not in metafunc.funcargnames, "need combine?" 88 if hasattr(metafunc.function, 'gwtypes'): 89 gwtypes = metafunc.function.gwtypes 90 elif hasattr(metafunc.cls, 'gwtype'): 91 gwtypes = [metafunc.cls.gwtype] 92 else: 93 gwtypes = ['popen', 'socket', 'ssh', 'proxy'] 94 metafunc.parametrize("gw", gwtypes, indirect=True) 95 elif 'anypython' in metafunc.funcargnames: 96 metafunc.parametrize( 97 "anypython", indirect=True, argvalues=( 98 'sys.executable', 'python2.7', 'pypy', 'jython', 99 ) 100 ) 101 102 103def getexecutable(name, cache={}): 104 try: 105 return cache[name] 106 except KeyError: 107 if name == 'sys.executable': 108 return py.path.local(sys.executable) 109 executable = py.path.local.sysfind(name) 110 if executable: 111 if name == "jython": 112 popen = subprocess.Popen( 113 [str(executable), "--version"], 114 universal_newlines=True, stderr=subprocess.PIPE) 115 out, err = popen.communicate() 116 if not err or "2.5" not in err: 117 executable = None 118 cache[name] = executable 119 return executable 120 121 122@pytest.fixture 123def anypython(request): 124 name = request.param 125 executable = getexecutable(name) 126 if executable is None: 127 if sys.platform == "win32": 128 executable = winpymap.get(name, None) 129 if executable: 130 executable = py.path.local(executable) 131 if executable.check(): 132 return executable 133 executable = None 134 py.test.skip("no {} found".format(name)) 135 if "execmodel" in request.fixturenames and name != 'sys.executable': 136 backend = request.getfixturevalue("execmodel").backend 137 if backend != "thread": 138 pytest.xfail( 139 "cannot run {!r} execmodel with bare {}".format(backend, name)) 140 return executable 141 142 143@pytest.fixture(scope='session') 144def group(): 145 g = execnet.Group() 146 yield g 147 g.terminate(timeout=1) 148 149 150@pytest.fixture 151def gw(request, execmodel, group): 152 try: 153 return group[request.param] 154 except KeyError: 155 if request.param == "popen": 156 gw = group.makegateway("popen//id=popen//execmodel=%s" 157 % execmodel.backend) 158 elif request.param == "socket": 159 # if execmodel.backend != "thread": 160 # pytest.xfail( 161 # "cannot set remote non-thread execmodel for sockets") 162 pname = 'sproxy1' 163 if pname not in group: 164 proxygw = group.makegateway("popen//id=%s" % pname) 165 # assert group['proxygw'].remote_status().receiving 166 gw = group.makegateway( 167 "socket//id=socket//installvia=%s" 168 "//execmodel=%s" % (pname, execmodel.backend)) 169 gw.proxygw = proxygw 170 assert pname in group 171 elif request.param == "ssh": 172 sshhost = request.getfixturevalue('specssh').ssh 173 # we don't use execmodel.backend here 174 # but you can set it when specifying the ssh spec 175 gw = group.makegateway("ssh={}//id=ssh".format(sshhost)) 176 elif request.param == 'proxy': 177 group.makegateway('popen//id=proxy-transport') 178 gw = group.makegateway('popen//via=proxy-transport//id=proxy' 179 '//execmodel=%s' % execmodel.backend) 180 else: 181 assert 0, "unknown execmodel: {}".format(request.param) 182 return gw 183 184 185@pytest.fixture(params=["thread", "eventlet", "gevent"], scope="session") 186def execmodel(request): 187 if request.param != "thread": 188 pytest.importorskip(request.param) 189 if sys.platform == "win32": 190 pytest.xfail("eventlet/gevent do not work onwin32") 191 return get_execmodel(request.param) 192 193 194@pytest.fixture 195def pool(execmodel): 196 return WorkerPool(execmodel=execmodel) 197