Example of unit testing Fusion-C's printf() with OpenMSX

autoexec.bat

rem put this file in dsk directory along with printf.com
cls
printf

env.py

#!/bin/env python3

OPENMSX_PATH = "/usr/bin/openmsx"

lib.py

#!/bin/env python3
import io
import os
import sys
import subprocess
import select

from subprocess import PIPE, TimeoutExpired
from env import OPENMSX_PATH


class OutputError(IOError): pass


def has_data(stream, timeout=0.1):
    'Check if there are data available.'
    result = select.select([stream], [], [], timeout)[0]
    return bool(result)


def expect(expected_data, from_=sys.stdin, timeout=0.1, debug=False):
    'Expect message from OpenMSX.'
    s = io.StringIO()
    if debug:
        print('* expecting "%s"...' % expected_data.rstrip(), end='')
        sys.stdout.flush()
    while has_data(from_, timeout=timeout):
        tmp = from_.readline()
        print(tmp, file=s, end='')
        if s.tell() >= len(expected_data):
            if s.getvalue().startswith(expected_data):
                if debug: print(' success!')
                return
    if debug: print(' fail!')
    raise OutputError()


def comm(ostream, istream, *args, debug=False):
    'Send message to openmsx and return answer.'
    s = io.StringIO()
    print(*args, file=s, end='')
    # OpenMSX XML parser doesn't need closing tags for openmsx-control.
    input = '<openmsx-control><command>%s</command>\n' % s.getvalue()
    if debug: print('* writing:', input.replace('\n', ';'))
    size = ostream.write(input)
    if size > 0:
        s.truncate(0)
        while has_data(istream):
            #tmp = istream.read(1)
            tmp = istream.readline()
            if not len(tmp): break
            print(tmp, file=s, end='')
        if debug: print('* reading:', s.getvalue())
        return s.getvalue()
    raise OutputError()


def show(proc, renderer='sdl', debug=False):
    data = comm(proc.stdin, proc.stdout, 'set renderer %s' % renderer, debug=debug)

printf.c

/* Test if Fusion-C print is working. */
#include <stdio.h>

/* send byte to debug device, single byte, decimal mode. */
#define HASH            #
#define h(x)            HASH x
#define sendb(x)        __asm               \
                        push af;            \
                        ld a, h(0x12);      \
                        out (h(0x2e)), a;   \
                        ld a, h(x);         \
                        out (h(0x2f)), a;   \
                        pop af;             \
                        __endasm
 

int main() {
    printf("Hello world!\r\n");
    sendb(0); /* trigger OpenMSX watchpoint */
    return 0;
}

test_printf.py

#!/bin/env python3
import sys
from lib import *


debug = False


with subprocess.Popen([OPENMSX_PATH, '-machine', 'msx2', '-ext', 'msxdos2', '-ext', 'debugdevice', '-control', 'stdio'], text=True, bufsize=1, stdin=PIPE, stdout=PIPE, stderr=PIPE) as proc:
    try:
        expect('<openmsx-output>', from_=proc.stdout, timeout=10, debug=debug)
        #show(proc, debug=False)
        data = comm(proc.stdin, proc.stdout, 'diska dsk', debug=debug)
        data = comm(proc.stdin, proc.stdout, 'set speed 1000', debug=debug) # 10 times faster
        # Set breakpoint when program runs
        data = comm(proc.stdin, proc.stdout, 'debug set_watchpoint write_io 0x2f [expr true] {puts stderr [get_screen]}', debug=debug)
        data = comm(proc.stdin, proc.stdout, 'set power on', debug=debug)

        # Compare program output with expected result.
        expect('Hello world!', from_=proc.stderr, timeout=10, debug=debug)
        data = comm(proc.stdin, proc.stdout, 'quit', debug=debug)
    except OutputError:
        data = comm(proc.stdin, proc.stdout, 'quit', debug=debug)
        sys.exit(1)

sys.exit(0)