Commit beb9d47e authored by Jérome Perrin's avatar Jérome Perrin

Treat program output as binary for python3 support

While treating output as text would not really be impossible, treating it
as bytes seems a better choice because:
 - we don't have to make assumptions about what output encoding the test
   program is using for output
 - `tee` can just read stream output bytes by bytes without having to worry
   about multi-bytes characters
 - testnode protocol uses xmlrpc.client.Binary, which uses bytes.

Because using bufsize=1 implies reading subprocess output as text, we use
bufsize=0 instead in the subprocess.Popen call, to prevent buffering.

To make manipulation of strings and bytes easier, we add a dependency on
pygolang, so that we can use its strings utility functions.

Also add a few tests to verify general functionality.
parent d829e9ca
......@@ -60,6 +60,7 @@ from subprocess import Popen, PIPE
from time import time, strftime, gmtime, localtime
import os, sys, threading, argparse, logging, traceback, re, pwd, socket
import six
from golang import b
# loadNXDTestFile loads .nxdtest file located @path.
def loadNXDTestFile(path): # -> TestEnv
......@@ -164,6 +165,13 @@ def main():
# log information about local node
system_info()
if sys.version_info < (3,):
bstdout = sys.stdout
bstderr = sys.stderr
else:
bstdout = sys.stdout.buffer
bstderr = sys.stderr.buffer
# run the tests
devnull = open(os.devnull)
while 1:
......@@ -193,29 +201,28 @@ def main():
# In addition to kw['env'], kw['envadj'] allows users to define
# only adjustments instead of providing full env dict.
# Test command is spawned with unchanged cwd. Instance wrapper cares to set cwd before running us.
# bufsize=1 means 'line buffered'
kw = t.kw.copy()
env = kw.pop('env', os.environ)
env = env.copy()
envadj = kw.pop('envadj', {})
env.update(envadj)
p = Popen(t.argv, env=env, stdin=devnull, stdout=PIPE, stderr=PIPE, bufsize=1, **kw)
p = Popen(t.argv, env=env, stdin=devnull, stdout=PIPE, stderr=PIPE, bufsize=0, **kw)
except:
stdout, stderr = '', traceback.format_exc()
sys.stderr.write(stderr)
stdout, stderr = b'', b(traceback.format_exc())
bstderr.write(stderr)
status['error_count'] += 1
else:
# tee >stdout,stderr so we can also see in testnode logs
# (explicit teeing instead of p.communicate() to be able to see incremental progress)
buf_out = []
buf_err = []
tout = threading.Thread(target=tee, args=(p.stdout, sys.stdout, buf_out))
terr = threading.Thread(target=tee, args=(p.stderr, sys.stderr, buf_err))
tout = threading.Thread(target=tee, args=(p.stdout, bstdout, buf_out))
terr = threading.Thread(target=tee, args=(p.stderr, bstderr, buf_err))
tout.start()
terr.start()
tout.join(); stdout = ''.join(buf_out)
terr.join(); stderr = ''.join(buf_err)
tout.join(); stdout = b''.join(buf_out)
terr.join(); stderr = b''.join(buf_err)
p.wait()
if p.returncode != 0:
......@@ -226,8 +233,8 @@ def main():
try:
summary = t.summaryf(stdout)
except:
bad = traceback.format_exc()
sys.stderr.write(bad)
bad = b(traceback.format_exc())
bstderr.write(bad)
stderr += bad
status['error_count'] += 1
......@@ -363,7 +370,7 @@ class PyTest:
return {}
def get(name, default=None):
m = re.search(r'\b([0-9]+) '+name+r'\b', pytail)
m = re.search(br'\b([0-9]+) ' + name.encode() + br'\b', pytail)
if m is None:
return default
return int(m.group(1))
......
......@@ -21,12 +21,13 @@
from nxdtest import _test_result_summary, PyTest
import pytest
from golang import b
# [] of (name, textout, summaryok)
# [] of (name, out, summaryok)
testv = []
def case1(name, textout, summaryok): testv.append((name, textout, summaryok))
def case1(name, out, summaryok): testv.append((name, out, summaryok))
case1('ok+xfail', """\
case1('ok+xfail', b("""\
============================= test session starts ==============================
platform linux2 -- Python 2.7.18, pytest-4.6.11, py-1.9.0, pluggy-0.13.1
rootdir: /srv/slapgrid/slappart9/srv/testnode/dfq/soft/46d349541123ed5fc6ceea58fd013a51/parts/zodbtools-dev
......@@ -39,10 +40,10 @@ zodbtools/test/test_tidrange.py ............................. [ 81%]
zodbtools/test/test_zodb.py ........ [100%]
=============== 41 passed, 2 xfailed, 1 warnings in 4.62 seconds ===============
""",
"""),
'?\ttestname\t1.000s\t# 43t ?e ?f ?s')
case1('ok+fail', """\
case1('ok+fail', b("""\
============================= test session starts ==============================
platform linux2 -- Python 2.7.18, pytest-4.6.11, py-1.9.0, pluggy-0.13.1
rootdir: /srv/slapgrid/slappart16/srv/testnode/dfj/soft/8b9988ce0aa31334c6bd56b40e4bba65/parts/pygolang-dev
......@@ -111,10 +112,10 @@ E Use -v to get the full diff
golang/time_test.py:106: AssertionError
=============== 1 failed, 98 passed, 13 skipped in 26.85 seconds ===============
""",
"""),
'?\ttestname\t1.000s\t# 112t ?e 1f 13s')
case1('ok+tailtext', """\
case1('ok+tailtext', b("""\
date: Sun, 08 Nov 2020 12:26:24 MSK
xnode: kirr@deco.navytux.spb.ru
uname: Linux deco 5.9.0-1-amd64 #1 SMP Debian 5.9.1-1 (2020-10-17) x86_64
......@@ -129,13 +130,13 @@ wcfs: 2020/11/08 12:26:38 /tmp/testdb_fs.Z9IvT0/1.fs: watcher: stat /tmp/testdb_
# unmount/stop wcfs pid39653 @ /tmp/wcfs/40cc7154ed758d6a867205e79e320c1d3b56458d
wcfs: 2020/11/08 12:26:38 /tmp/testdb_fs.B3rbby/1.fs: watcher: stat /tmp/testdb_fs.B3rbby/1.fs: use of closed file
# unmount/stop wcfs pid39595 @ /tmp/wcfs/d0b5d036a2cce47fe73003cf2d9f0b22c7043817
""",
"""),
'?\ttestname\t1.000s\t# 55t ?e ?f ?s')
@pytest.mark.parametrize("name,textout,summaryok", testv)
def test_pytest_summary(name,textout, summaryok):
@pytest.mark.parametrize("name,out,summaryok", testv)
def test_pytest_summary(name, out, summaryok):
kw = {'duration': 1.0}
kw.update(PyTest.summary(textout))
kw.update(PyTest.summary(out))
summary = _test_result_summary('testname', kw)
assert summary == summaryok
# -*- coding: utf-8 -*-
# Copyright (C) 2020 Nexedi SA and Contributors.
#
# This program is free software: you can Use, Study, Modify and Redistribute
# it under the terms of the GNU General Public License version 3, or (at your
# option) any later version, as published by the Free Software Foundation.
#
# You can also Link and Combine this program with other software covered by
# the terms of any of the Free Software licenses or any of the Open Source
# Initiative approved licenses and Convey the resulting work. Corresponding
# source of such a combination shall include the source code for all other
# software used.
#
# This program is distributed WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
# verify general functionality
import sys
import re
import pytest
from nxdtest import main
@pytest.fixture
def run_nxdtest(tmpdir):
"""Fixture which returns a function which invokes nxdtest in a temporary
directory, with the provided .nxdtest file content and with arguments
passed as `argv`.
"""
def _run_nxdtest(nxdtest_file_content, argv=("nxdtest",)):
with tmpdir.as_cwd():
with open(".nxdtest", "w") as f:
f.write(nxdtest_file_content)
sys_argv = sys.argv
sys.argv = argv
try:
main()
finally:
sys.argv = sys_argv
return _run_nxdtest
def test_main(run_nxdtest, capsys):
run_nxdtest(
"""\
TestCase('TESTNAME', ['echo', 'TEST OUPUT'])
"""
)
captured = capsys.readouterr()
output_lines = captured.out.splitlines()
assert ">>> TESTNAME" in output_lines
assert "$ echo TEST OUPUT" in output_lines
assert "TEST OUPUT" in output_lines
assert re.match("ok\tTESTNAME\t.*s\t# 1t 0e 0f 0s", output_lines[-1])
def test_error_invoking_command(run_nxdtest, capsys):
run_nxdtest(
"""\
TestCase('TESTNAME', ['not exist command'])
"""
)
captured = capsys.readouterr()
assert "No such file or directory" in captured.err
def test_error_invoking_summary(run_nxdtest, capsys):
run_nxdtest(
"""\
TestCase('TESTNAME', ['echo'], summaryf="error")
"""
)
captured = capsys.readouterr()
assert "TypeError" in captured.err
def test_run_argument(run_nxdtest, capsys):
run_nxdtest(
"""\
TestCase('TEST1', ['echo', 'TEST1'])
TestCase('TEST2', ['echo', 'TEST2'])
""",
argv=["nxdtest", "--run", "TEST1"],
)
captured = capsys.readouterr()
assert "TEST1" in captured.out
assert "TEST2" not in captured.out
......@@ -13,7 +13,7 @@ setup(
keywords = 'Nexedi testing infrastructure tool tox',
packages = find_packages(),
install_requires = ['erp5.util', 'six'],
install_requires = ['erp5.util', 'six', 'pygolang'],
extras_require = {
'test': ['pytest'],
},
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment