testph.py

"""
A simple test for ph; the philosophers problem solution - verify that
no adjacent philosophers are eating simultaneously.
"""

import subprocess
import signal
import sys
import re

def main(n):
    process = subprocess.Popen(['./ph', str(n)], stdout=subprocess.PIPE)
    try:
        _verify_process(process)
    except KeyboardInterrupt:
        print '    Goodbye'
    finally:
        process.send_signal(signal.SIGINT)
        process.wait()
        std_output, err_output = process.communicate()
        print std_output
        print 'Total run time:', _sum_runtime(std_output)

def _sum_runtime(output):
    total_time = 0
    for line in output.splitlines():
        ph_time = float(re.match('Philosopher \\(\\d+\\) used the CPU (\\d+\\.\\d+) sec.', line).groups()[0])
        total_time += ph_time
    return total_time

def _verify_process(process):
    currently_eating = set()
    while True:
        current_line = process.stdout.readline().strip()
        print current_line
        philosopher, action = re.match('Philosopher \\((\d+)\\) is (.*)', current_line).groups()
        philosopher = int(philosopher)
        if action == 'eating':
            currently_eating.add(philosopher)
            print '    Added %d.   Now eating: %r' % (philosopher, currently_eating)
        else:
            if philosopher in currently_eating:
                currently_eating.remove(philosopher)
                print '    Removed %d. Now eating: %r' % (philosopher, currently_eating)
        verify_eating(currently_eating)

def verify_eating(eating):
    for philosopher in eating:
        assert philosopher - 1 not in eating, "Philosophers %d and %d are eating!" % (philosopher, philosopher - 1)
        assert philosopher + 1 not in eating, "Philosophers %d and %d are eating!" % (philosopher, philosopher + 1)

if '__main__' == __name__:
    main(int(sys.argv[1]))