runtest

#!/usr/bin/env python3
import sys
import codecs
import glob
import tempfile
import pathlib
import subprocess
import atexit
import argparse

def reconfigure(stream):
	return codecs.getwriter(stream.encoding)(stream.detach(), errors='surrogateescape')

sys.stdout = reconfigure(sys.stdout)
sys.stderr = reconfigure(sys.stderr)

def parse_args():
	parser = argparse.ArgumentParser()
	parser.add_argument('testfile', metavar='TEST_FILE', type=pathlib.Path)
	parser.add_argument('-u', '--update', action='store_const', const=True)
	parser.add_argument('-v', '--verbose', action='store_const', const=True)
	parser.add_argument('-k', '--keep', action='store_const', const=True)
	parser.add_argument('--reference', action='store_const', const=True)
	parser.add_argument('--no-setrlimit', action='store_const', const=True)
	return parser.parse_args()

args = parse_args()

testfile = args.testfile.open('rt')

in_source    = []
in_data      = bytearray()
in_data_at   = 0
in_choices   = []
expect_msg   = []
expect_data  = in_data
expect_diswarn = False
expect_exit  = 0
test_dis     = not args.reference
test_skipout = False

mode = 'source'
for line in testfile:
	if line.startswith(';;;#'):
		continue
	if line.startswith(';;;;'):
		line = line[len(';;;;'):].strip()
		sect, _, param = line.partition(' ')

		if sect == 'source':
			mode = 'source'
		elif sect == 'output':
			mode = 'output'
		elif sect == 'data':
			mode = 'data'
		elif sect == 'data-at':
			in_data_at = int(param.strip(), 0)
		elif sect == 'target':
			mode = 'target'
			if expect_data is in_data:
				expect_data = bytearray()
		elif sect == 'comment':
			mode = 'comment'
		elif sect == 'choose':
			in_choices.append(int(param.strip(), 0))
		elif sect == 'diswarn':
			if param == 'ignore':
				expect_diswarn = None
			else:
				expect_diswarn = True
		elif sect == 'skipoutref':
			test_skipout = args.reference
		elif sect == 'exit':
			param = param.strip()
			if param == 'fatal':
				expect_exit = None
			else:
				expect_exit = int(param.strip(), 0)
		else:
			print("=== Invalid section: {} ===".format(repr(line)))
			sys.exit(1)
		continue

	if mode == 'comment':
		continue
	elif mode == 'source':
		in_source.append(line)
	elif mode == 'output':
		expect_msg.append(line)
	elif mode == 'data':
		in_data += bytes.fromhex(line.strip())
	elif mode == 'target':
		expect_data += bytes.fromhex(line.strip())
	else:
		print("=== No mode ===")
		sys.exit(1)

if expect_msg is not None:
	expect_msg = ''.join(expect_msg)

in_source = ''.join(in_source).encode('utf-8')

fname_tmpdir = pathlib.Path(tempfile.mkdtemp(prefix='bsptest-'))
fname_bsp0   = fname_tmpdir / 'bsp0'
fname_input  = fname_tmpdir / 'input'
fname_target = fname_tmpdir / 'output'
fname_dis    = fname_tmpdir / 'dis'
fname_bsp1   = fname_tmpdir / 'bsp1'

discrepancy = False

# compile
if args.verbose:
	print("=== Compiling ===")

compiler = subprocess.Popen(
	['bspcomp/bspcomp', '/dev/stdin', str(fname_bsp0)],
	stdin=subprocess.PIPE
)
compiler.stdin.write(in_source)
compiler.stdin.close()
if compiler.wait(timeout=10) != 0:
	print("=== Compilation error ===")
	sys.exit(1)

if test_dis:
	if args.verbose:
		print("=== Disassembling ===")

	disassembler = subprocess.Popen(
		['./bspdis', '-n', '-o', str(fname_dis), str(fname_bsp0)],
		stdout=subprocess.PIPE, stderr=subprocess.PIPE
	)
	(_, dis_err) = disassembler.communicate(timeout=10)

	if dis_err != b'':
		if expect_diswarn:
			if args.verbose:
				print("=== Expected disassembler warnings emitted ===")
		elif expect_diswarn is None:
			if args.verbose:
				print("=== Disassembler warnings (ignored) ===")
		else:
			print("=== Unexpected disassembler warnings ===")
			discrepancy = True
		if args.verbose or discrepancy:
			print(dis_err.decode('utf-8', 'surrogateescape'))
	elif expect_diswarn:
		print("=== Expected disassembler warnings not emitted ===")
		discrepancy = True

	if args.verbose:
		print("=== Disassembly ===")
		with fname_dis.open('r', errors='surrogateescape') as dis:
			print(dis.read())
		print("=== Compiling again ===")

	compiler = subprocess.Popen(
		['bspcomp/bspcomp', str(fname_dis), str(fname_bsp1)]
	)
	if compiler.wait(timeout=10) != 0:
		print("=== Second-pass compilation error ===")
		discrepancy = True
	else:
		if args.verbose:
			print("=== Verifying ===")

		with fname_bsp0.open('rb') as bsp0:
			with fname_bsp1.open('rb') as bsp1:
				data0 = bsp0.read()
				data1 = bsp1.read()
				if data0 != data1:
					print("=== Disassembly verification failed ===")
					print("first : {}".format(data0))
					print("second: {}".format(data1))
					discrepancy = True

# prepare input
with fname_input.open('wb') as f:
	f.seek(in_data_at)
	if in_data:
		f.write(in_data)
	else:
		f.truncate()

if args.verbose:
	print("=== Running ===")

import resource

def set_limits():
	as_limit = 1024 * 1024 * 1024
	resource.setrlimit(resource.RLIMIT_AS, (as_limit, as_limit))

if args.reference:
	argv = ['node', 'bspcomp/patcher.js', str(fname_bsp0), str(fname_input), str(fname_target)]
else:
	argv = ['./bsp', '-N16', '-I8192', '-dttk' if args.verbose else '-dk', str(fname_bsp0), str(fname_input), str(fname_target)]

interpreter = subprocess.Popen(argv,
	stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
	preexec_fn=None if args.no_setrlimit else set_limits
)

with interpreter:
	try:
		(actual_msg, actual_err) = interpreter.communicate(
			input=(''.join('{}\n'.format(ch) for ch in in_choices) + 'q\n').encode('ascii'),
			timeout=10)
	except subprocess.TimeoutExpired as e:
		interpreter.kill()
		(actual_msg, actual_err) = interpreter.communicate()
		actual_err = actual_err.decode('utf-8', 'surrogateescape')
		actual_msg = actual_msg.decode('utf-8')
		if actual_msg and not actual_msg.endswith('\n'):
			actual_msg += '\n'
		actual_exit = None
	else:
		actual_err = actual_err.decode('utf-8', 'surrogateescape')
		actual_msg = actual_msg.decode('utf-8')
		if actual_msg and not actual_msg.endswith('\n'):
			actual_msg += '\n'
		if args.reference:
			if interpreter.returncode != 0:
				actual_msg_lines = actual_msg.rstrip('\n').split('\n')
				last_line = actual_msg_lines.pop()
				if last_line.startswith('Error: '):
					actual_exit = None
					actual_msg = '\n'.join(actual_msg_lines)
				elif last_line.startswith('Patch exited with exit status '):
					actual_exit = int(last_line[len('Patch exited with exit status '):])
					actual_msg = '\n'.join(actual_msg_lines)
				else:
					actual_exit = None
			else:
				actual_exit = 0
		else:
			if interpreter.returncode == 255:
				actual_exit = None
			elif interpreter.returncode == 254:
				for line in reversed(actual_err.splitlines()):
					left, sep, line = line.partition(': ')
					if sep != ': ':
						continue
					left, sep, line = line.partition(': ')
					if sep != ': ':
						continue
					if line.startswith('patching failed, exit code '):
						actual_exit = int(line[len('patching failed, exit code '):])
						break
				else:
					raise ValueError(actual_err)
			else:
				actual_exit = interpreter.returncode

try:
	with fname_target.open('rb') as f:
		f.seek(in_data_at)
		actual_data = f.read()
except FileNotFoundError:
	actual_data = None

def rm_rf(node):
	if node.is_dir():
		for child in node.iterdir():
			rm_rf(child)
		node.rmdir()
	else:
		node.unlink()

if args.keep:
	print("=== Files kept in {} ===".format(fname_tmpdir))
else:
	rm_rf(fname_tmpdir)

if expect_exit != actual_exit:
	discrepancy = True
if not test_skipout and expect_msg != actual_msg:
	discrepancy = True
if actual_data is not None and expect_data != actual_data:
	discrepancy = True

if discrepancy or args.verbose:
	print("=== Standard error ===")
	print(actual_err)

	if expect_msg is not None:
		print("=== Expected output ===")
		print(expect_msg)
	else:
		print("=== No expected output ===")
	if expect_msg != actual_msg:
		print("=== Actual output ===")
		print(actual_msg)

	print("=== Expected final data ===")
	print(expect_data.hex())
	if actual_data is not None:
		if expect_data != actual_data:
			print("=== Actual final data ===")
			print(actual_data.hex())
		else:
			print("=== Actual final data matches ===")
	else:
		print("=== No actual final data ===")

	if expect_exit is not None:
		print("=== Expected exit code: {} ===".format(expect_exit))
	else:
		print("=== Expected a fatal error ===")

	if expect_exit != actual_exit:
		if actual_exit is not None:
			print("=== Actual exit code: {} ===".format(actual_exit))
	else:
		if actual_exit is not None:
			print("=== Actual exit code matches ===")
		else:
			print("=== Expected fatal error raised ===")

_OUTPUT_SECTIONS = ('output', 'target', 'exit')

if discrepancy:
	if args.update:
		import os
		newname = args.testfile.with_suffix('.newtest')
		emitted = set()

		def emit(sect):
			if sect in emitted:
				return
			emitted.add(sect)

			if sect == 'output':
				if actual_msg:
					newfile.write(';;;; output\n')
					newfile.write(actual_msg)
			elif sect == 'target':
				if actual_data is not None and actual_data != in_data:
					newfile.write(';;;; target\n')
					newfile.write(actual_data.hex() + '\n')
			elif sect == 'exit':
				if actual_exit is None:
					newfile.write(';;;; exit fatal\n')
				elif actual_exit:
					newfile.write(';;;; exit %u\n' % (actual_exit))

		with args.testfile.open('rt') as testfile, newname.open('wt') as newfile:
			skip = False
			for line in testfile:
				if line.startswith(';;;;'):
					sect, _, param = line[len(';;;;'):].strip().partition(' ')
					skip = sect in _OUTPUT_SECTIONS
					if skip:
						emit(sect)
				if not skip:
					newfile.write(line)

			for sect in _OUTPUT_SECTIONS:
				emit(sect)

		newname.replace(args.testfile)

	sys.exit(1)