Differential D2155 Diff 6223 test/functional/test_runner.py

Changeset View

Standalone View

test/functional/test_runner.py

Show All 25 Lines
import subprocess		import subprocess
import tempfile		import tempfile
import re		import re
import logging		import logging
import xml.etree.ElementTree as ET		import xml.etree.ElementTree as ET
import json		import json
import threading		import threading
import multiprocessing		import multiprocessing
from queue import Queue, Empty		import importlib
		import importlib.util
		import multiprocessing as mp
		from queue import Full, Empty

		from test_framework.test_framework import BitcoinTestFramework
		from test_framework.cdefs import get_srcdir

# Formatting. Default colors to empty strings.		# Formatting. Default colors to empty strings.
BOLD, BLUE, RED, GREY = ("", ""), ("", ""), ("", ""), ("", "")		BOLD, BLUE, RED, GREY = ("", ""), ("", ""), ("", ""), ("", "")
try:		try:
# Make sure python thinks it can write unicode to its stdout		# Make sure python thinks it can write unicode to its stdout
"\u2713".encode("utf_8").decode(sys.stdout.encoding)		"\u2713".encode("utf_8").decode(sys.stdout.encoding)
TICK = "✓ "		TICK = "✓ "
CROSS = "✖ "		CROSS = "✖ "
▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Lines	def run(self, portseed_offset):
stdout=log_stdout,		stdout=log_stdout,
stderr=log_stderr)		stderr=log_stderr)

process.wait()		process.wait()
log_stdout.seek(0), log_stderr.seek(0)		log_stdout.seek(0), log_stderr.seek(0)
[stdout, stderr] = [l.read().decode('utf-8')		[stdout, stderr] = [l.read().decode('utf-8')
for l in (log_stdout, log_stderr)]		for l in (log_stdout, log_stderr)]
log_stdout.close(), log_stderr.close()		log_stdout.close(), log_stderr.close()

if process.returncode == TEST_EXIT_PASSED and stderr == "":		if process.returncode == TEST_EXIT_PASSED and stderr == "":
status = "Passed"		status = "Passed"
elif process.returncode == TEST_EXIT_SKIPPED:		elif process.returncode == TEST_EXIT_SKIPPED:
status = "Skipped"		status = "Skipped"
else:		else:
status = "Failed"		status = "Failed"

return TestResult(name, status, int(time.time() - time0), stdout, stderr)		return TestResult(name, status, int(time.time() - time0), stdout, stderr)
▲ Show 20 Lines • Show All 206 Lines • ▼ Show 20 Lines	def run_tests(test_list, build_dir, tests_dir, junitouput, exeext, tmpdir, num_jobs, enable_coverage=False, args=[], build_timings=None):
if not os.listdir(tmpdir):		if not os.listdir(tmpdir):
os.rmdir(tmpdir)		os.rmdir(tmpdir)

all_passed = all(		all_passed = all(
map(lambda test_result: test_result.status == "Passed", test_results))		map(lambda test_result: test_result.status == "Passed", test_results))

sys.exit(not all_passed)		sys.exit(not all_passed)


##		##
# Define some helper functions we will need for threading.		# Define some helper functions we will need for threading.
##		##

def handle_message(message, running_jobs):
		def handle_message(message, running_jobs, results_queue):
"""		"""
handle_message handles a single message from handle_test_cases		handle_message handles a single message from handle_test_cases
"""		"""
if isinstance(message, TestCase):		if isinstance(message, TestCase):
running_jobs.add(message.test_case)		running_jobs.add(message.test_case)
print("{}{}{} started".format(BOLD[1], message.test_case, BOLD[0]))		print("{}{}{} started".format(BOLD[1], message.test_case, BOLD[0]))
return		return
if isinstance(message, TestResult):		if isinstance(message, TestResult):
test_result = message		test_result = message
running_jobs.remove(test_result.name)		running_jobs.remove(test_result.name)
test_results.append(test_result)		results_queue.put(test_result)
if test_result.status == "Passed":		if test_result.status == "Passed":
print("%s%s%s passed, Duration: %s s" % (		print("%s%s%s passed, Duration: %s s" % (
BOLD[1], test_result.name, BOLD[0], test_result.time))		BOLD[1], test_result.name, BOLD[0], test_result.time))
elif test_result.status == "Skipped":		elif test_result.status == "Skipped":
print("%s%s%s skipped" %		print("%s%s%s skipped" %
(BOLD[1], test_result.name, BOLD[0]))		(BOLD[1], test_result.name, BOLD[0]))
else:		else:
print("%s%s%s failed, Duration: %s s\n" %		print("%s%s%s failed, Duration: %s s\n" %
(BOLD[1], test_result.name, BOLD[0], test_result.time))		(BOLD[1], test_result.name, BOLD[0], test_result.time))
print(BOLD[1] + 'stdout:' + BOLD[0])		print(BOLD[1] + 'stdout:' + BOLD[0])
print(test_result.stdout)		print(test_result.stdout)
print(BOLD[1] + 'stderr:' + BOLD[0])		print(BOLD[1] + 'stderr:' + BOLD[0])
print(test_result.stderr)		print(test_result.stderr)
return		return
assert False, "we should not be here"		assert False, "we should not be here"


def handle_update_messages():		def handle_update_messages(update_queue, results_queue):
"""		"""
handle_update_messages waits for messages to be sent from handle_test_cases via the		handle_update_messages waits for messages to be sent from handle_test_cases via the
update_queue. It serializes the results so we can print nice status update messages.		update_queue. It serializes the results so we can print nice status update messages.
"""		"""
printed_status = False		printed_status = False
running_jobs = set()		running_jobs = set()
		poll_timeout = 10 # seconds

while True:		while True:
message = None		message = None
try:		try:
message = update_queue.get(True, poll_timeout)		message = update_queue.get(True, poll_timeout)
if message is None:		if message is None:
break		break
# We printed a status message, need to kick to the next line		# We printed a status message, need to kick to the next line
# before printing more.		# before printing more.
if printed_status:		if printed_status:
print()		print()
printed_status = False		printed_status = False
handle_message(message, running_jobs)		handle_message(message, running_jobs, results_queue)
update_queue.task_done()		update_queue.task_done()
except Empty as e:		except Empty as e:
if not on_ci():		if not on_ci():
print("Running jobs: {}".format(		print("Running jobs: {}".format(
", ".join(running_jobs)), end="\r")		", ".join(running_jobs)), end="\r")
sys.stdout.flush()		sys.stdout.flush()
printed_status = True		printed_status = True


def handle_test_cases():		def handle_test_cases(job_queue, update_queue):
"""		"""
job_runner represents a single thread that is part of a worker pool.		job_runner represents a single thread that is part of a worker pool.
It waits for a test, then executes that test.		It waits for a test, then executes that test. It also reports start
It also reports start and result messages to handle_update_messages		and result messages to handle_update_messages.
"""		"""
		# In case there is a graveyard of zombie bitcoinds, we can apply a
		# pseudorandom offset to hopefully jump over them.
		# (625 is PORT_RANGE/MAX_NODES)
		portseed_offset = int(time.time() * 1000) % 625

while True:		while True:
test = job_queue.get()		test = job_queue.get()
if test is None:		if test is None:
break		break
# Signal that the test is starting to inform the poor waiting		# Signal that the test is starting to inform the poor waiting
# programmer		# programmer
update_queue.put(test)		update_queue.put(test)
result = test.run(portseed_offset)		result = test.run(portseed_offset)
update_queue.put(result)		update_queue.put(result)
job_queue.task_done()		job_queue.task_done()


def execute_test_processes(num_jobs, test_list, tests_dir, tmpdir, flags):		def execute_test_processes(num_jobs, test_list, tests_dir, tmpdir, flags):
update_queue = Queue()		ctx = mp.get_context('spawn')
job_queue = Queue()		update_queue = ctx.JoinableQueue()
test_results = []		job_queue = ctx.JoinableQueue()
poll_timeout = 10 # seconds		results_queue = ctx.Queue()
# In case there is a graveyard of zombie bitcoinds, we can apply a
# pseudorandom offset to hopefully jump over them.
# (625 is PORT_RANGE/MAX_NODES)
portseed_offset = int(time.time() * 1000) % 625

##		##
# Setup our threads, and start sending tasks		# Setup our threads, and start sending tasks
##		##

# Start our result collection thread.		# Start our result collection thread.
t = threading.Thread(target=handle_update_messages)		t = ctx.Process(target=handle_update_messages,
t.setDaemon(True)		args=(update_queue, results_queue,))
t.start()		t.start()

# Start some worker threads		# Start some worker threads
for j in range(num_jobs):		for j in range(num_jobs):
t = threading.Thread(target=handle_test_cases)		t = ctx.Process(target=handle_test_cases,
t.setDaemon(True)		args=(job_queue, update_queue,))
t.start()		t.start()

# Push all our test cases into the job queue.		# Push all our test cases into the job queue.
for i, t in enumerate(test_list):		for i, t in enumerate(test_list):
job_queue.put(TestCase(i, t, tests_dir, tmpdir, flags))		job_queue.put(TestCase(i, t, tests_dir, tmpdir, flags))

# Wait for all the jobs to be completed		# Wait for all the jobs to be completed
job_queue.join()		job_queue.join()

# Wait for all the results to be compiled		# Wait for all the results to be compiled
update_queue.join()		update_queue.join()

# Flush our queues so the threads exit		# Flush our queues so the threads exit
update_queue.put(None)		update_queue.put(None)
for j in range(num_jobs):		for j in range(num_jobs):
job_queue.put(None)		job_queue.put(None)

		# We've already stopped sending messages, so we can be sure when this queue is empty we are done.
		test_results = []
		while not results_queue.empty():
		test_results.append(results_queue.get())

return test_results		return test_results


def print_results(test_results, max_len_name, runtime):		def print_results(test_results, max_len_name, runtime):
results = "\n" + BOLD[1] + "%s \| %s \| %s\n\n" % (		results = "\n" + BOLD[1] + "%s \| %s \| %s\n\n" % (
"TEST".ljust(max_len_name), "STATUS ", "DURATION") + BOLD[0]		"TEST".ljust(max_len_name), "STATUS ", "DURATION") + BOLD[0]

test_results.sort(key=lambda result: result.name.lower())		test_results.sort(key=lambda result: result.name.lower())
▲ Show 20 Lines • Show All 238 Lines • Show Last 20 Lines