Skip to content

Commit

Permalink
testing: some cleanups
Browse files Browse the repository at this point in the history
ran: cherry-picked and adapted slightly.
  • Loading branch information
RonnyPfannschmidt authored and bluetech committed Apr 26, 2023
1 parent 844b3da commit 35e03ea
Showing 6 changed files with 178 additions and 218 deletions.
49 changes: 17 additions & 32 deletions testing/conftest.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import pathlib
import shutil
import subprocess
import sys
from functools import lru_cache
from typing import Callable
from typing import Iterator

import execnet
import pytest
@@ -23,10 +24,15 @@ def pytest_runtest_setup(item):


@pytest.fixture
def makegateway(request):
def group_function() -> Iterator[execnet.Group]:
group = execnet.Group()
request.addfinalizer(lambda: group.terminate(0.5))
return group.makegateway
yield group
group.terminate(0.5)


@pytest.fixture
def makegateway(group_function) -> Callable[[str], execnet.gateway.Gateway]:
return group_function.makegateway


pytest_plugins = ["pytester", "doctest"]
@@ -101,37 +107,16 @@ def pytest_generate_tests(metafunc):
else:
gwtypes = ["popen", "socket", "ssh", "proxy"]
metafunc.parametrize("gw", gwtypes, indirect=True)
elif "anypython" in metafunc.fixturenames:
metafunc.parametrize(
"anypython",
indirect=True,
argvalues=("sys.executable", "pypy3"),
)


def getexecutable(name, cache={}):
try:
return cache[name]
except KeyError:
if name == "sys.executable":
return pathlib.Path(sys.executable)
path = shutil.which(name)
executable = pathlib.Path(path) if path is not None else None
if executable:
if name == "jython":
popen = subprocess.Popen(
[str(executable), "--version"],
universal_newlines=True,
stderr=subprocess.PIPE,
)
out, err = popen.communicate()
if not err or "2.5" not in err:
executable = None
cache[name] = executable
return executable
@lru_cache()
def getexecutable(name):
if name == "sys.executable":
return sys.executable
return shutil.which(name)


@pytest.fixture
@pytest.fixture(params=("sys.executable", "pypy3"))
def anypython(request):
name = request.param
executable = getexecutable(name)
223 changes: 114 additions & 109 deletions testing/test_basics.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,14 @@
from __future__ import annotations

import inspect
import os
import subprocess
import sys
import textwrap
from dataclasses import dataclass
from io import BytesIO
from pathlib import Path
from typing import Any

import execnet
import pytest
@@ -21,19 +26,18 @@
)


@pytest.mark.parametrize("val", ["123", 42, [1, 2, 3], ["23", 25]])
class TestSerializeAPI:
pytestmark = [pytest.mark.parametrize("val", ["123", 42, [1, 2, 3], ["23", 25]])]

def test_serializer_api(self, val):
dumped = execnet.dumps(val)
val2 = execnet.loads(dumped)
assert val == val2

def test_mmap(self, tmp_path, val):
mmap = pytest.importorskip("mmap").mmap
p = tmp_path / "data"
with p.open("wb") as f:
f.write(execnet.dumps(val))
p = tmp_path / "data.bin"

p.write_bytes(execnet.dumps(val))
with p.open("r+b") as f:
m = mmap(f.fileno(), 0)
val2 = execnet.load(m)
@@ -112,67 +116,83 @@ def read_write_loop():
break


def test_io_message(anypython, tmp_path, execmodel):
check = tmp_path / "check.py"
check.write_text(
inspect.getsource(gateway_base)
+ textwrap.dedent(
"""
from io import BytesIO
import tempfile
temp_out = BytesIO()
temp_in = BytesIO()
io = Popen2IO(temp_out, temp_in, get_execmodel({backend!r}))
for i, handler in enumerate(Message._types):
print ("checking %s %s" %(i, handler))
for data in "hello", "hello".encode('ascii'):
msg1 = Message(i, i, dumps(data))
msg1.to_io(io)
x = io.outfile.getvalue()
io.outfile.truncate(0)
io.outfile.seek(0)
io.infile.seek(0)
io.infile.write(x)
io.infile.seek(0)
msg2 = Message.from_io(io)
assert msg1.channelid == msg2.channelid, (msg1, msg2)
assert msg1.data == msg2.data, (msg1.data, msg2.data)
assert msg1.msgcode == msg2.msgcode
print ("all passed")
""".format(
backend=execmodel.backend
),
IO_MESSAGE_EXTRA_SOURCE = """
import sys
backend = sys.argv[1]
try:
from io import BytesIO
except ImportError:
from StringIO import StringIO as BytesIO
import tempfile
temp_out = BytesIO()
temp_in = BytesIO()
io = Popen2IO(temp_out, temp_in, get_execmodel(backend))
for i, handler in enumerate(Message._types):
print ("checking", i, handler)
for data in "hello", "hello".encode('ascii'):
msg1 = Message(i, i, dumps(data))
msg1.to_io(io)
x = io.outfile.getvalue()
io.outfile.truncate(0)
io.outfile.seek(0)
io.infile.seek(0)
io.infile.write(x)
io.infile.seek(0)
msg2 = Message.from_io(io)
assert msg1.channelid == msg2.channelid, (msg1, msg2)
assert msg1.data == msg2.data, (msg1.data, msg2.data)
assert msg1.msgcode == msg2.msgcode
print ("all passed")
"""


@dataclass
class Checker:
python: str
path: Path
idx: int = 0

def run_check(
self, script: str, *extra_args: str, **process_args: Any
) -> subprocess.CompletedProcess[str]:
self.idx += 1
check_path = self.path / f"check{self.idx}.py"
check_path.write_text(script)
return subprocess.run(
[self.python, os.fspath(check_path), *extra_args],
capture_output=True,
text=True,
check=True,
**process_args,
)


@pytest.fixture
def checker(anypython: str, tmp_path: Path) -> Checker:
return Checker(python=anypython, path=tmp_path)


def test_io_message(checker, execmodel):
out = checker.run_check(
inspect.getsource(gateway_base) + IO_MESSAGE_EXTRA_SOURCE, execmodel.backend
)
out = subprocess.run(
[str(anypython), str(check)], text=True, capture_output=True, check=True
).stdout
print(out)
assert "all passed" in out
print(out.stdout)
assert "all passed" in out.stdout


def test_popen_io(anypython, tmp_path, execmodel):
check = tmp_path / "check.py"
check.write_text(
def test_popen_io(checker, execmodel):
out = checker.run_check(
inspect.getsource(gateway_base)
+ textwrap.dedent(
f"""
io = init_popen_io(get_execmodel({execmodel.backend!r}))
io.write("hello".encode('ascii'))
s = io.read(1)
assert s == "x".encode('ascii')
"""
),
+ f"""
io = init_popen_io(get_execmodel({execmodel.backend!r}))
io.write(b"hello")
s = io.read(1)
assert s == b"x"
""",
input="x",
)
from subprocess import Popen, PIPE

args = [str(anypython), str(check)]
proc = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE)
proc.stdin.write(b"x")
stdout, stderr = proc.communicate()
print(stderr)
proc.wait()
assert b"hello" in stdout
print(out.stderr)
assert "hello" in out.stdout


def test_popen_io_readloop(monkeypatch, execmodel):
@@ -190,60 +210,45 @@ def newread(numbytes):
assert result == b"tes"


def test_rinfo_source(anypython, tmp_path):
check = tmp_path / "check.py"
check.write_text(
textwrap.dedent(
"""
class Channel:
def send(self, data):
assert eval(repr(data), {}) == data
channel = Channel()
"""
)
+ inspect.getsource(gateway.rinfo_source)
+ textwrap.dedent(
"""
print ('all passed')
"""
)
def test_rinfo_source(checker):
out = checker.run_check(
f"""
class Channel:
def send(self, data):
assert eval(repr(data), {{}}) == data
channel = Channel()
{inspect.getsource(gateway.rinfo_source)}
print ('all passed')
"""
)
out = subprocess.run(
[str(anypython), str(check)], text=True, capture_output=True, check=True
).stdout
print(out)
assert "all passed" in out

print(out.stdout)
assert "all passed" in out.stdout

def test_geterrortext(anypython, tmp_path):
check = tmp_path / "check.py"
check.write_text(

def test_geterrortext(checker):
out = checker.run_check(
inspect.getsource(gateway_base)
+ textwrap.dedent(
"""
class Arg:
pass
errortext = geterrortext((Arg, "1", 4))
assert "Arg" in errortext
import sys
try:
raise ValueError("17")
except ValueError:
excinfo = sys.exc_info()
s = geterrortext(excinfo)
assert "17" in s
print ("all passed")
+ """
class Arg:
pass
errortext = geterrortext((Arg, "1", 4))
assert "Arg" in errortext
import sys
try:
raise ValueError("17")
except ValueError:
excinfo = sys.exc_info()
s = geterrortext(excinfo)
assert "17" in s
print ("all passed")
"""
)
)
out = subprocess.run(
[str(anypython), str(check)], text=True, capture_output=True, check=True
).stdout
print(out)
assert "all passed" in out
print(out.stdout)
assert "all passed" in out.stdout


@pytest.mark.skipif(not hasattr(os, "dup"), reason="no os.dup")
@pytest.mark.skipif("not hasattr(os, 'dup')")
def test_stdouterrin_setnull(execmodel, capfd):
gateway_base.init_popen_io(execmodel)
os.write(1, b"hello")
@@ -297,15 +302,15 @@ def test_wire_protocol(self):
class TestPureChannel:
@pytest.fixture
def fac(self, execmodel):
class Gateway:
class FakeGateway:
def _trace(self, *args):
pass

def _send(self, *k):
pass

Gateway.execmodel = execmodel
return ChannelFactory(Gateway())
FakeGateway.execmodel = execmodel
return ChannelFactory(FakeGateway())

def test_factory_create(self, fac):
chan1 = fac.new()
Loading

0 comments on commit 35e03ea

Please sign in to comment.