219 lines
12 KiB
Python
219 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Native CLI-/Standalone-Parität. Vorher cargo build --release für tbc/tbrt/tb-template."""
|
|
import argparse
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
def run(args, cwd, *, data=b'', env=None):
|
|
return subprocess.run([str(a) for a in args], cwd=cwd, input=data,
|
|
capture_output=True, timeout=30, env=env)
|
|
|
|
|
|
def must(result):
|
|
assert result.returncode == 0, (result.args, result.returncode, result.stdout, result.stderr)
|
|
return result
|
|
|
|
|
|
def terminal(binary, arguments, cwd, env, abort=False):
|
|
"""Echte Unix-PTY, Eingabe/Abbruch und unveränderte Terminalattribute."""
|
|
import fcntl
|
|
import select
|
|
import signal
|
|
import termios
|
|
import time
|
|
master, slave = os.openpty()
|
|
original = termios.tcgetattr(slave)
|
|
fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', 30, 100, 0, 0))
|
|
pid = os.fork()
|
|
if pid == 0:
|
|
os.setsid()
|
|
fcntl.ioctl(slave, termios.TIOCSCTTY, 0)
|
|
for fd in (0, 1, 2): os.dup2(slave, fd)
|
|
os.close(master)
|
|
os.close(slave)
|
|
result = subprocess.run([str(binary), *map(str, arguments)], cwd=cwd, env=env, check=False)
|
|
restored = termios.tcgetattr(0) == original
|
|
os._exit(result.returncode if restored else 99)
|
|
transcript = bytearray()
|
|
sent = False
|
|
reaped = False
|
|
deadline = time.monotonic() + 15
|
|
try:
|
|
while True:
|
|
if select.select([master], [], [], .05)[0]:
|
|
transcript.extend(os.read(master, 65536))
|
|
if not sent and b'READY' in transcript:
|
|
os.write(master, b'\x03' if abort else b'Ada\r')
|
|
sent = True
|
|
child, status = os.waitpid(pid, os.WNOHANG)
|
|
if child:
|
|
reaped = True
|
|
assert os.waitstatus_to_exitcode(status) == (3 if abort else 0), bytes(transcript)
|
|
break
|
|
assert time.monotonic() < deadline, bytes(transcript[-2000:])
|
|
assert sent and b'?1049h' in transcript and b'?1049l' in transcript, transcript
|
|
assert (b'Abgebrochen' if abort else b'PTYAda') in transcript, transcript
|
|
finally:
|
|
if not reaped:
|
|
os.killpg(pid, signal.SIGKILL)
|
|
os.waitpid(pid, 0)
|
|
os.close(master)
|
|
os.close(slave)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--bin-dir', type=Path, default=Path('target/release'))
|
|
parser.add_argument('--target', required=True)
|
|
args = parser.parse_args()
|
|
binaries = args.bin_dir.resolve()
|
|
suffix = '.exe' if args.target == 'x86_64-pc-windows-msvc' else ''
|
|
tbc = binaries / ('tbc' + suffix)
|
|
rt = binaries / ('tbrt' + suffix)
|
|
prepare = binaries / ('tb-template' + suffix)
|
|
env = dict(os.environ, PATH='') # weder Cargo noch tbc/tbrt über PATH auflösbar
|
|
with tempfile.TemporaryDirectory(prefix='tb-native-') as work:
|
|
root = Path(work)
|
|
template = root / ('runtime' + suffix)
|
|
shutil.copy2(rt, template)
|
|
must(run([prepare, template, args.target], root))
|
|
empty = run([template], root, env=env)
|
|
assert empty.returncode == 1 and b'Nutzlast' in empty.stderr, empty
|
|
cases = {
|
|
'console': ('PRINT "Hallo, Welt!"\nPRINT COMMAND$\nINPUT "Name"; n$\nPRINT n$\nLPRINT "Papier"\nEND\n', b'Ada\n'),
|
|
'restart': ('OPEN "count.txt" FOR BINARY AS #1\nn%=LOF(1)\nCLOSE #1\nIF n%=0 THEN\nOPEN "count.txt" FOR OUTPUT AS #1\nPRINT #1, "einmal"\nCLOSE #1\nx%=99\nRUN\nEND IF\nPRINT x%; COMMAND$\nEND\n', b''),
|
|
'stop': ('PRINT "vor STOP"\nSTOP\nPRINT "falsch"\n', b''),
|
|
'error': ('PRINT "vor Fehler"\nERROR 6\n', b''),
|
|
'project': ('CALL Ergebnis\nEND\n', b''),
|
|
'form': ('VERSION 1.00\nBegin Form Form1\n Caption = "Native Form"\n Width = 40\n Height = 10\n Begin CommandButton Ok\n Caption = "OK"\n Default = -1\n Left = 2\n Top = 2\n Width = 10\n Height = 1\n End\nEnd\nForm1.Show\nSUB Ok_Click\nForm1.Hide\nCLS\nPRINT "FORMDONE"\nEND\nEND SUB\n', b'\n'),
|
|
'external': ('RUN "NEXT.TBC"\n', b''),
|
|
}
|
|
for name, (code, data) in cases.items():
|
|
case = root / name
|
|
sources, reference, clean = (case / p for p in ('sources', 'reference', 'clean'))
|
|
for d in (sources, reference, clean): d.mkdir(parents=True)
|
|
source = sources / ('main.frm' if name == 'form' else 'main.bas')
|
|
source.write_text(code, encoding='utf-8')
|
|
if name == 'project':
|
|
(sources / 'lib.bas').write_text("'$INCLUDE: 'OUTER.BI'\n")
|
|
(sources / 'outer.bi').write_text("'$INCLUDE: 'INNER.BI'\n")
|
|
(sources / 'inner.bi').write_text('SUB Ergebnis\nOPEN "seed.txt" FOR INPUT AS #1\nLINE INPUT #1, s$\nCLOSE #1\nOPEN "result.txt" FOR OUTPUT AS #2\nPRINT #2, s$\nCLOSE #2\nPRINT s$\nERROR 6\nEND SUB\n')
|
|
source = sources / 'app.mak'
|
|
source.write_text('MAIN.BAS\nLIB.BAS\n')
|
|
for d in (reference, clean): (d / 'seed.txt').write_text('Grüße\n', encoding='utf-8')
|
|
if name == 'form':
|
|
source.write_text(code.replace('PRINT "FORMDONE"', 'CALL Ergebnis'), encoding='utf-8')
|
|
(sources / 'lib.bas').write_text("'$INCLUDE: 'OUTER.BI'\n")
|
|
(sources / 'outer.bi').write_text("'$INCLUDE: 'INNER.BI'\n")
|
|
(sources / 'inner.bi').write_text('SUB Ergebnis\nPRINT "FORMDONE"\nEND SUB\n')
|
|
source = sources / 'app.mak'
|
|
source.write_text('MAIN.FRM\nLIB.BAS\n')
|
|
if name == 'external':
|
|
child = sources / 'next.bas'
|
|
child.write_text('PRINT "EXTERNAL"; COMMAND$\nEND\n')
|
|
must(run([tbc, 'build', child], root))
|
|
shutil.copy2(child.with_suffix('.tbc'), clean / 'next.tbc')
|
|
expected = run([tbc, 'run', source, 'alpha', 'ä beta'], reference, data=data)
|
|
exe = clean / ('program' + suffix)
|
|
build_source = source
|
|
if name == 'error':
|
|
must(run([tbc, 'build', source], root))
|
|
build_source = source.with_suffix('.tbc')
|
|
command = [tbc, 'build', build_source, '--exe', '--target', args.target, '--template', template, '-o', exe]
|
|
must(run(command, root, env=env))
|
|
original = exe.read_bytes()
|
|
conflict = run(command, root, env=env)
|
|
assert conflict.returncode == 1 and b'--force' in conflict.stderr
|
|
assert exe.read_bytes() == original
|
|
must(run(command + ['--force'], root, env=env))
|
|
if sys.platform == 'darwin':
|
|
must(run(['/usr/bin/codesign', '--verify', '--strict', exe], root))
|
|
shutil.rmtree(sources)
|
|
actual = run([exe, 'alpha', 'ä beta'], clean, data=data, env=env)
|
|
assert (actual.returncode, actual.stdout, actual.stderr) == (expected.returncode, expected.stdout, expected.stderr), (name, expected, actual)
|
|
if name == 'form': assert b'FORMDONE' in actual.stdout, actual
|
|
if name == 'restart': assert b'99' not in actual.stdout and b'alpha' in actual.stdout, actual
|
|
if name == 'project': assert b'inner.bi:9:1:' in actual.stderr.lower(), actual
|
|
for file in reference.iterdir():
|
|
assert (clean / file.name).read_bytes() == file.read_bytes(), file
|
|
assert sorted(p.name for p in clean.iterdir()) == sorted([exe.name] + [p.name for p in reference.iterdir()] + (['next.tbc'] if name == 'external' else []))
|
|
print('PASS native', name, flush=True)
|
|
if os.name == 'posix':
|
|
source = root / 'terminal.bas'
|
|
source.write_text('PRINT "READY"\nINPUT s$\nPRINT "PTY"; s$\nEND\n')
|
|
exe = root / 'terminal'
|
|
must(run([tbc, 'build', source, '--exe', '--template', template, '--target', args.target, '-o', exe], root, env=env))
|
|
for abort in (False, True):
|
|
terminal(tbc, ['run', source], root, env, abort)
|
|
source.unlink()
|
|
for abort in (False, True):
|
|
terminal(exe, [], root, env, abort)
|
|
print('PASS PTY Eingabe, Abbruch und Wiederherstellung (CLI/native)', flush=True)
|
|
if os.name == 'posix':
|
|
import signal
|
|
import time
|
|
source = root / 'cancel.bas'
|
|
source.write_text('PRINT "nicht veröffentlicht"\nEND\n')
|
|
output = root / 'cancelled'
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
output.write_bytes(b'original')
|
|
process = subprocess.Popen([str(tbc), 'build', str(source), '--exe', '--template', str(template), '--target', args.target, '-o', str(output), '--force'], cwd=root, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
try:
|
|
deadline = time.monotonic() + 10
|
|
while not list(root.glob('.tb-export-*.tmp')):
|
|
assert process.poll() is None and time.monotonic() < deadline, 'Export endete vor Abbruchprobe'
|
|
time.sleep(.001)
|
|
process.send_signal(sig)
|
|
stdout, stderr = process.communicate(timeout=15)
|
|
assert process.returncode == 1 and b'abgebrochen' in stderr, (process.returncode, stdout, stderr)
|
|
assert output.read_bytes() == b'original'
|
|
assert not list(root.glob('.tb-export-*.tmp'))
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.wait()
|
|
print('PASS Exportabbruch SIGINT/SIGTERM, Original und Bereinigung', flush=True)
|
|
# Beschädigte Nutzlast wird auf macOS neu signiert, damit der Lader statt des Kernels prüft.
|
|
valid = (root / 'console/clean' / ('program' + suffix)).read_bytes()
|
|
footer = valid.rfind(b'TBPCODE!')
|
|
assert footer >= 0
|
|
for name, mutate in (
|
|
('length', lambda b: struct.pack_into('<Q', b, footer + 32, 2**64-1)),
|
|
('version', lambda b: struct.pack_into('<I', b, footer + 8, 99)),
|
|
('truncated', lambda b: b.__delitem__(slice(footer-10, footer))),
|
|
):
|
|
bad = root / ('bad-' + name + suffix)
|
|
data = bytearray(valid)
|
|
if sys.platform == 'darwin':
|
|
# Zunächst Signatur entfernen; Finalisierung wird anschließend erneut angewendet.
|
|
bad.write_bytes(data)
|
|
must(run(['/usr/bin/codesign', '--remove-signature', bad], root))
|
|
data = bytearray(bad.read_bytes())
|
|
mutate(data)
|
|
if sys.platform == 'darwin':
|
|
at = 32
|
|
for _ in range(struct.unpack_from('<I', data, 16)[0]):
|
|
cmd, size = struct.unpack_from('<II', data, at)
|
|
if cmd == 0x19 and data[at+8:at+24].rstrip(b'\0') == b'__LINKEDIT':
|
|
start = struct.unpack_from('<Q', data, at+40)[0]
|
|
struct.pack_into('<Q', data, at+48, len(data)-start)
|
|
at += size
|
|
bad.write_bytes(data)
|
|
bad.chmod(0o755)
|
|
if sys.platform == 'darwin':
|
|
must(run(['/usr/bin/codesign', '--force', '--sign', '-', '--timestamp=none', bad], root))
|
|
result = run([bad], root, env=env)
|
|
assert result.returncode == 1 and b'Ladefehler' in result.stderr and not result.stdout, result
|
|
print('PASS corrupted', name, flush=True)
|
|
print('Native Abnahme bestanden:', args.target)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|