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