#!/usr/bin/env python

#

# Copyright 2008 the V8 project authors. All rights reserved.

# Redistribution and use in source and binary forms, with or without

# modification, are permitted provided that the following conditions are

# met:

#

# * Redistributions of source code must retain the above copyright

# notice, this list of conditions and the following disclaimer.

# * Redistributions in binary form must reproduce the above

# copyright notice, this list of conditions and the following

# disclaimer in the documentation and/or other materials provided

# with the distribution.

# * Neither the name of Google Inc. nor the names of its

# contributors may be used to endorse or promote products derived

# from this software without specific prior written permission.

#

# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS

# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT

# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR

# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT

# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import imp

import logging

import optparse

import os

import platform

import re

import signal

import subprocess

import sys

import tempfile

import time

import threading

import utils

import multiprocessing

import errno

from os.path import join, dirname, abspath, basename, isdir, exists

from datetime import datetime

from Queue import Queue, Empty

logger = logging.getLogger('testrunner')

skip_regex = re.compile(r'# SKIP\S*\s+(.*)', re.IGNORECASE)

VERBOSE = False

# ---------------------------------------------

# --- P r o g r e s s I n d i c a t o r s ---

# ---------------------------------------------

class ProgressIndicator(object):

def __init__(self, cases, flaky_tests_mode):

self.cases = cases

self.flaky_tests_mode = flaky_tests_mode

self.parallel_queue = Queue(len(cases))

self.sequential_queue = Queue(len(cases))

for case in cases:

if case.parallel:

self.parallel_queue.put_nowait(case)

else:

self.sequential_queue.put_nowait(case)

self.succeeded = 0

self.remaining = len(cases)

self.total = len(cases)

self.failed = [ ]

self.flaky_failed = [ ]

self.crashed = 0

self.flaky_crashed = 0

self.lock = threading.Lock()

self.shutdown_event = threading.Event()

def PrintFailureHeader(self, test):

if test.IsNegative():

negative_marker = '[negative] '

else:

negative_marker = ''

print "=== %(label)s %(negative)s===" % {

'label': test.GetLabel(),

'negative': negative_marker

}

print "Path: %s" % "/".join(test.path)

def Run(self, tasks):

self.Starting()

threads = []

# Spawn N-1 threads and then use this thread as the last one.

# That way -j1 avoids threading altogether which is a nice fallback

# in case of threading problems.

for i in xrange(tasks - 1):

thread = threading.Thread(target=self.RunSingle, args=[True, i + 1])

threads.append(thread)

thread.start()

try:

self.RunSingle(False, 0)

# Wait for the remaining threads

for thread in threads:

# Use a timeout so that signals (ctrl-c) will be processed.

thread.join(timeout=10000000)

except (KeyboardInterrupt, SystemExit), e:

self.shutdown_event.set()

except Exception, e:

# If there's an exception we schedule an interruption for any

# remaining threads.

self.shutdown_event.set()

# ...and then reraise the exception to bail out

raise

self.Done()

return not self.failed

def RunSingle(self, parallel, thread_id):

while not self.shutdown_event.is_set():

try:

test = self.parallel_queue.get_nowait()

except Empty:

if parallel:

return

try:

test = self.sequential_queue.get_nowait()

except Empty:

return

case = test.case

case.thread_id = thread_id

self.lock.acquire()

self.AboutToRun(case)

self.lock.release()

try:

start = datetime.now()

output = case.Run()

# SmartOS has a bug that causes unexpected ECONNREFUSED errors.

# See https://smartos.org/bugview/OS-2767

# If ECONNREFUSED on SmartOS, retry the test one time.

if (output.UnexpectedOutput() and

sys.platform == 'sunos5' and

'ECONNREFUSED' in output.output.stderr):

output = case.Run()

output.diagnostic.append('ECONNREFUSED received, test retried')

case.duration = (datetime.now() - start)

except IOError, e:

return

if self.shutdown_event.is_set():

return

self.lock.acquire()

if output.UnexpectedOutput():

if FLAKY in output.test.outcomes and self.flaky_tests_mode == DONTCARE:

self.flaky_failed.append(output)

if output.HasCrashed():

self.flaky_crashed += 1

else:

self.failed.append(output)

if output.HasCrashed():

self.crashed += 1

else:

self.succeeded += 1

self.remaining -= 1

self.HasRun(output)

self.lock.release()

def EscapeCommand(command):

parts = []

for part in command:

if ' ' in part:

# Escape spaces. We may need to escape more characters for this

# to work properly.

parts.append('"%s"' % part)

else:

parts.append(part)

return " ".join(parts)

class SimpleProgressIndicator(ProgressIndicator):

def Starting(self):

print 'Running %i tests' % len(self.cases)

def Done(self):

print

for failed in self.failed:

self.PrintFailureHeader(failed.test)

if failed.output.stderr:

print "--- stderr ---"

print failed.output.stderr.strip()

if failed.output.stdout:

print "--- stdout ---"

print failed.output.stdout.strip()

print "Command: %s" % EscapeCommand(failed.command)

if failed.HasCrashed():

print "--- CRASHED ---"

if failed.HasTimedOut():

print "--- TIMEOUT ---"

if len(self.failed) == 0:

print "==="

print "=== All tests succeeded"

print "==="

else:

print

print "==="

print "=== %i tests failed" % len(self.failed)

if self.crashed > 0:

print "=== %i tests CRASHED" % self.crashed

print "==="

class VerboseProgressIndicator(SimpleProgressIndicator):

def AboutToRun(self, case):

print 'Starting %s...' % case.GetLabel()

sys.stdout.flush()

def HasRun(self, output):

if output.UnexpectedOutput():

if output.HasCrashed():

outcome = 'CRASH'

else:

outcome = 'FAIL'

else:

outcome = 'pass'

print 'Done running %s: %s' % (output.test.GetLabel(), outcome)

class DotsProgressIndicator(SimpleProgressIndicator):

def AboutToRun(self, case):

pass

def HasRun(self, output):

total = self.succeeded + len(self.failed)

if (total > 1) and (total % 50 == 1):

sys.stdout.write('\n')

if output.UnexpectedOutput():

if output.HasCrashed():

sys.stdout.write('C')

sys.stdout.flush()

elif output.HasTimedOut():

sys.stdout.write('T')

sys.stdout.flush()

else:

sys.stdout.write('F')

sys.stdout.flush()

else:

sys.stdout.write('.')

sys.stdout.flush()

class TapProgressIndicator(SimpleProgressIndicator):

def _printDiagnostic(self, messages):

for l in messages.splitlines():

logger.info('# ' + l)

def Starting(self):

logger.info('1..%i' % len(self.cases))

self._done = 0

def AboutToRun(self, case):

pass

def HasRun(self, output):

self._done += 1

command = basename(output.command[-1])

if output.UnexpectedOutput():

status_line = 'not ok %i %s' % (self._done, command)

if FLAKY in output.test.outcomes and self.flaky_tests_mode == DONTCARE:

status_line = status_line + ' # TODO : Fix flaky test'

logger.info(status_line)

self._printDiagnostic("\n".join(output.diagnostic))

if output.HasTimedOut():

self._printDiagnostic('TIMEOUT')

self._printDiagnostic(output.output.stderr)

self._printDiagnostic(output.output.stdout)

else:

skip = skip_regex.search(output.output.stdout)

if skip:

logger.info(

'ok %i %s # skip %s' % (self._done, command, skip.group(1)))

else:

status_line = 'ok %i %s' % (self._done, command)

if FLAKY in output.test.outcomes:

status_line = status_line + ' # TODO : Fix flaky test'

logger.info(status_line)

self._printDiagnostic("\n".join(output.diagnostic))

duration = output.test.duration

# total_seconds() was added in 2.7

total_seconds = (duration.microseconds +

(duration.seconds + duration.days * 24 * 3600) * 10**6) / 10**6

logger.info(' ---')

logger.info(' duration_ms: %d.%d' % (total_seconds, duration.microseconds / 1000))

logger.info(' ...')

def Done(self):

pass

class CompactProgressIndicator(ProgressIndicator):

def __init__(self, cases, flaky_tests_mode, templates):

super(CompactProgressIndicator, self).__init__(cases, flaky_tests_mode)

self.templates = templates

self.last_status_length = 0

self.start_time = time.time()

def Starting(self):

pass

def Done(self):

self.PrintProgress('Done')

def AboutToRun(self, case):

self.PrintProgress(case.GetLabel())

def HasRun(self, output):

if output.UnexpectedOutput():

self.ClearLine(self.last_status_length)

self.PrintFailureHeader(output.test)

stdout = output.output.stdout.strip()

if len(stdout):

print self.templates['stdout'] % stdout

stderr = output.output.stderr.strip()

if len(stderr):

print self.templates['stderr'] % stderr

print "Command: %s" % EscapeCommand(output.command)

if output.HasCrashed():

print "--- CRASHED ---"

if output.HasTimedOut():

print "--- TIMEOUT ---"

def Truncate(self, str, length):

if length and (len(str) > (length - 3)):

return str[:(length-3)] + "..."

else:

return str

def PrintProgress(self, name):

self.ClearLine(self.last_status_length)

elapsed = time.time() - self.start_time

status = self.templates['status_line'] % {

'passed': self.succeeded,

'remaining': (((self.total - self.remaining) * 100) // self.total),

'failed': len(self.failed),

'test': name,

'mins': int(elapsed) / 60,

'secs': int(elapsed) % 60

}

status = self.Truncate(status, 78)

self.last_status_length = len(status)

print status,

sys.stdout.flush()

class ColorProgressIndicator(CompactProgressIndicator):

def __init__(self, cases, flaky_tests_mode):

templates = {

'status_line': "[%(mins)02i:%(secs)02i|\033[34m%%%(remaining) 4d\033[0m|\033[32m+%(passed) 4d\033[0m|\033[31m-%(failed) 4d\033[0m]: %(test)s",

'stdout': "\033[1m%s\033[0m",

'stderr': "\033[31m%s\033[0m",

}

super(ColorProgressIndicator, self).__init__(cases, flaky_tests_mode, templates)

def ClearLine(self, last_line_length):

print "\033[1K\r",

class MonochromeProgressIndicator(CompactProgressIndicator):

def __init__(self, cases, flaky_tests_mode):

templates = {

'status_line': "[%(mins)02i:%(secs)02i|%%%(remaining) 4d|+%(passed) 4d|-%(failed) 4d]: %(test)s",

'stdout': '%s',

'stderr': '%s',

'clear': lambda last_line_length: ("\r" + (" " * last_line_length) + "\r"),

'max_length': 78

}

super(MonochromeProgressIndicator, self).__init__(cases, flaky_tests_mode, templates)

def ClearLine(self, last_line_length):

print ("\r" + (" " * last_line_length) + "\r"),

PROGRESS_INDICATORS = {

'verbose': VerboseProgressIndicator,

'dots': DotsProgressIndicator,

'color': ColorProgressIndicator,

'tap': TapProgressIndicator,

'mono': MonochromeProgressIndicator

}

# -------------------------

# --- F r a m e w o r k ---

# -------------------------

class CommandOutput(object):

def __init__(self, exit_code, timed_out, stdout, stderr):

self.exit_code = exit_code

self.timed_out = timed_out

self.stdout = stdout

self.stderr = stderr

self.failed = None

class TestCase(object):

def __init__(self, context, path, arch, mode):

self.path = path

self.context = context

self.duration = None

self.arch = arch

self.mode = mode

self.parallel = False

self.thread_id = 0

def IsNegative(self):

return self.context.expect_fail

def CompareTime(self, other):

return cmp(other.duration, self.duration)

def DidFail(self, output):

if output.failed is None:

output.failed = self.IsFailureOutput(output)

return output.failed

def IsFailureOutput(self, output):

return output.exit_code != 0

def GetSource(self):

return "(no source available)"

def RunCommand(self, command, env):

full_command = self.context.processor(command)

output = Execute(full_command,

self.context,

self.context.GetTimeout(self.mode),

env)

self.Cleanup()

return TestOutput(self,

full_command,

output,

self.context.store_unexpected_output)

def BeforeRun(self):

pass

def AfterRun(self, result):

pass

def Run(self):

self.BeforeRun()

try:

result = self.RunCommand(self.GetCommand(), {

"TEST_THREAD_ID": "%d" % self.thread_id

})

finally:

# Tests can leave the tty in non-blocking mode. If the test runner

# tries to print to stdout/stderr after that and the tty buffer is

# full, it'll die with a EAGAIN OSError. Ergo, put the tty back in

# blocking mode before proceeding.

if sys.platform != 'win32':

from fcntl import fcntl, F_GETFL, F_SETFL

from os import O_NONBLOCK

for fd in 0,1,2: fcntl(fd, F_SETFL, ~O_NONBLOCK & fcntl(fd, F_GETFL))

self.AfterRun(result)

return result

def Cleanup(self):

return

class TestOutput(object):

def __init__(self, test, command, output, store_unexpected_output):

self.test = test

self.command = command

self.output = output

self.store_unexpected_output = store_unexpected_output

self.diagnostic = []

def UnexpectedOutput(self):

if self.HasCrashed():

outcome = CRASH

elif self.HasTimedOut():

outcome = TIMEOUT

elif self.HasFailed():

outcome = FAIL

else:

outcome = PASS

return not outcome in self.test.outcomes

def HasPreciousOutput(self):

return self.UnexpectedOutput() and self.store_unexpected_output

def HasCrashed(self):

if utils.IsWindows():

return 0x80000000 & self.output.exit_code and not (0x3FFFFF00 & self.output.exit_code)

else:

# Timed out tests will have exit_code -signal.SIGTERM.

if self.output.timed_out:

return False

return self.output.exit_code < 0 and \

self.output.exit_code != -signal.SIGABRT

def HasTimedOut(self):

return self.output.timed_out;

def HasFailed(self):

execution_failed = self.test.DidFail(self.output)

if self.test.IsNegative():

return not execution_failed

else:

return execution_failed

def KillProcessWithID(pid):

if utils.IsWindows():

os.popen('taskkill /T /F /PID %d' % pid)

else:

os.kill(pid, signal.SIGTERM)

MAX_SLEEP_TIME = 0.1

INITIAL_SLEEP_TIME = 0.0001

SLEEP_TIME_FACTOR = 1.25

SEM_INVALID_VALUE = -1

SEM_NOGPFAULTERRORBOX = 0x0002 # Microsoft Platform SDK WinBase.h

def Win32SetErrorMode(mode):

prev_error_mode = SEM_INVALID_VALUE

try:

import ctypes

prev_error_mode = ctypes.windll.kernel32.SetErrorMode(mode);

except ImportError:

pass

return prev_error_mode

def RunProcess(context, timeout, args, **rest):

if context.verbose: print "#", " ".join(args)

popen_args = args

prev_error_mode = SEM_INVALID_VALUE;

if utils.IsWindows():

if context.suppress_dialogs:

# Try to change the error mode to avoid dialogs on fatal errors. Don't

# touch any existing error mode flags by merging the existing error mode.

# See http://blogs.msdn.com/oldnewthing/archive/2004/07/27/198410.aspx.

error_mode = SEM_NOGPFAULTERRORBOX;

prev_error_mode = Win32SetErrorMode(error_mode);

Win32SetErrorMode(error_mode | prev_error_mode);

process = subprocess.Popen(

shell = utils.IsWindows(),

args = popen_args,

**rest

)

if utils.IsWindows() and context.suppress_dialogs and prev_error_mode != SEM_INVALID_VALUE:

Win32SetErrorMode(prev_error_mode)

# Compute the end time - if the process crosses this limit we

# consider it timed out.

if timeout is None: end_time = None

else: end_time = time.time() + timeout

timed_out = False

# Repeatedly check the exit code from the process in a

# loop and keep track of whether or not it times out.

exit_code = None

sleep_time = INITIAL_SLEEP_TIME

while exit_code is None:

if (not end_time is None) and (time.time() >= end_time):

# Kill the process and wait for it to exit.

KillProcessWithID(process.pid)

exit_code = process.wait()

timed_out = True

else:

exit_code = process.poll()

time.sleep(sleep_time)

sleep_time = sleep_time * SLEEP_TIME_FACTOR

if sleep_time > MAX_SLEEP_TIME:

sleep_time = MAX_SLEEP_TIME

return (process, exit_code, timed_out)

def PrintError(str):

sys.stderr.write(str)

sys.stderr.write('\n')

def CheckedUnlink(name):

while True:

try:

os.unlink(name)

except OSError, e:

# On Windows unlink() fails if another process (typically a virus scanner

# or the indexing service) has the file open. Those processes keep a

# file open for a short time only, so yield and try again; it'll succeed.

if sys.platform == 'win32' and e.errno == errno.EACCES:

time.sleep(0)

continue

PrintError("os.unlink() " + str(e))

break

def Execute(args, context, timeout=None, env={}):

(fd_out, outname) = tempfile.mkstemp()

(fd_err, errname) = tempfile.mkstemp()

# Extend environment

env_copy = os.environ.copy()

for key, value in env.iteritems():

env_copy[key] = value

(process, exit_code, timed_out) = RunProcess(

context,

timeout,

args = args,

stdout = fd_out,

stderr = fd_err,

env = env_copy

)

os.close(fd_out)

os.close(fd_err)

output = file(outname).read()

errors = file(errname).read()

CheckedUnlink(outname)

CheckedUnlink(errname)

return CommandOutput(exit_code, timed_out, output, errors)

def ExecuteNoCapture(args, context, timeout=None):

(process, exit_code, timed_out) = RunProcess(

context,

timeout,

args = args,

)

return CommandOutput(exit_code, False, "", "")

def CarCdr(path):

if len(path) == 0:

return (None, [ ])

else:

return (path[0], path[1:])

class TestConfiguration(object):

def __init__(self, context, root):

self.context = context

self.root = root

def Contains(self, path, file):

if len(path) > len(file):

return False

for i in xrange(len(path)):

if not path[i].match(file[i]):

return False

return True

def GetTestStatus(self, sections, defs):

pass

class TestSuite(object):

def __init__(self, name):

self.name = name

def GetName(self):

return self.name

# Use this to run several variants of the tests, e.g.:

# VARIANT_FLAGS = [[], ['--always_compact', '--noflush_code']]

VARIANT_FLAGS = [[]]

class TestRepository(TestSuite):

def __init__(self, path):

normalized_path = abspath(path)

super(TestRepository, self).__init__(basename(normalized_path))

self.path = normalized_path

self.is_loaded = False

self.config = None

def GetConfiguration(self, context):

if self.is_loaded:

return self.config

self.is_loaded = True

file = None

try:

(file, pathname, description) = imp.find_module('testcfg', [ self.path ])

module = imp.load_module('testcfg', file, pathname, description)

self.config = module.GetConfiguration(context, self.path)

if hasattr(self.config, 'additional_flags'):

self.config.additional_flags += context.node_args

else:

self.config.additional_flags = context.node_args

finally:

if file:

file.close()

return self.config

def GetBuildRequirements(self, path, context):

return self.GetConfiguration(context).GetBuildRequirements()

def AddTestsToList(self, result, current_path, path, context, arch, mode):

for v in VARIANT_FLAGS:

tests = self.GetConfiguration(context).ListTests(current_path, path,

arch, mode)

for t in tests: t.variant_flags = v

result += tests

def GetTestStatus(self, context, sections, defs):

self.GetConfiguration(context).GetTestStatus(sections, defs)

class LiteralTestSuite(TestSuite):

def __init__(self, tests):

super(LiteralTestSuite, self).__init__('root')

self.tests = tests

def GetBuildRequirements(self, path, context):

(name, rest) = CarCdr(path)

result = [ ]

for test in self.tests:

if not name or name.match(test.GetName()):

result += test.GetBuildRequirements(rest, context)

return result

def ListTests(self, current_path, path, context, arch, mode):

(name, rest) = CarCdr(path)

result = [ ]

for test in self.tests:

test_name = test.GetName()

if not name or name.match(test_name):

full_path = current_path + [test_name]

test.AddTestsToList(result, full_path, path, context, arch, mode)

result.sort(cmp=lambda a, b: cmp(a.GetName(), b.GetName()))

return result

def GetTestStatus(self, context, sections, defs):

for test in self.tests:

test.GetTestStatus(context, sections, defs)

SUFFIX = {

'debug' : '_g',

'release' : '' }

FLAGS = {

'debug' : ['--enable-slow-asserts', '--debug-code', '--verify-heap'],

'release' : []}

TIMEOUT_SCALEFACTOR = {

'armv6' : { 'debug' : 12, 'release' : 3 }, # The ARM buildbots are slow.

'arm' : { 'debug' : 8, 'release' : 2 },

'ia32' : { 'debug' : 4, 'release' : 1 },

'ppc' : { 'debug' : 4, 'release' : 1 },

's390' : { 'debug' : 4, 'release' : 1 } }

class Context(object):

def __init__(self, workspace, buildspace, verbose, vm, args, expect_fail,

timeout, processor, suppress_dialogs, store_unexpected_output):

self.workspace = workspace

self.buildspace = buildspace

self.verbose = verbose

self.vm_root = vm

self.node_args = args

self.expect_fail = expect_fail

self.timeout = timeout

self.processor = processor

self.suppress_dialogs = suppress_dialogs

self.store_unexpected_output = store_unexpected_output

def GetVm(self, arch, mode):

if arch == 'none':

name = 'out/Debug/node' if mode == 'debug' else 'out/Release/node'

else:

name = 'out/%s.%s/node' % (arch, mode)

# Currently GYP does not support output_dir for MSVS.

# http://code.google.com/p/gyp/issues/detail?id=40

# It will put the builds into Release/node.exe or Debug/node.exe

if utils.IsWindows():

out_dir = os.path.join(dirname(__file__), "..", "out")

if not exists(out_dir):

if mode == 'debug':

name = os.path.abspath('Debug/node.exe')

else:

name = os.path.abspath('Release/node.exe')

else:

name = os.path.abspath(name + '.exe')

return name

def GetVmFlags(self, testcase, mode):

return testcase.variant_flags + FLAGS[mode]

def GetTimeout(self, mode):

return self.timeout * TIMEOUT_SCALEFACTOR[ARCH_GUESS or 'ia32'][mode]

def RunTestCases(cases_to_run, progress, tasks, flaky_tests_mode):

progress = PROGRESS_INDICATORS[progress](cases_to_run, flaky_tests_mode)

return progress.Run(tasks)

def BuildRequirements(context, requirements, mode, scons_flags):

command_line = (['scons', '-Y', context.workspace, 'mode=' + ",".join(mode)]

+ requirements

+ scons_flags)

output = ExecuteNoCapture(command_line, context)

return output.exit_code == 0

# -------------------------------------------

# --- T e s t C o n f i g u r a t i o n ---

# -------------------------------------------

SKIP = 'skip'

FAIL = 'fail'

PASS = 'pass'

OKAY = 'okay'

TIMEOUT = 'timeout'

CRASH = 'crash'

SLOW = 'slow'

FLAKY = 'flaky'

DONTCARE = 'dontcare'

class Expression(object):

pass

class Constant(Expression):

def __init__(self, value):

self.value = value

def Evaluate(self, env, defs):

return self.value

class Variable(Expression):

def __init__(self, name):

self.name = name

def GetOutcomes(self, env, defs):

if self.name in env: return ListSet([env[self.name]])

else: return Nothing()

class Outcome(Expression):

def __init__(self, name):

self.name = name

def GetOutcomes(self, env, defs):

if self.name in defs:

return defs[self.name].GetOutcomes(env, defs)

else:

return ListSet([self.name])

class Set(object):

pass

class ListSet(Set):

def __init__(self, elms):

self.elms = elms

def __str__(self):

return "ListSet%s" % str(self.elms)

def Intersect(self, that):

if not isinstance(that, ListSet):

return that.Intersect(self)

return ListSet([ x for x in self.elms if x in that.elms ])

def Union(self, that):

if not isinstance(that, ListSet):

return that.Union(self)

return ListSet(self.elms + [ x for x in that.elms if x not in self.elms ])

def IsEmpty(self):

return len(self.elms) == 0

class Everything(Set):

def Intersect(self, that):

return that

def Union(self, that):

return self

def IsEmpty(self):

return False

class Nothing(Set):

def Intersect(self, that):

return self

def Union(self, that):

return that

def IsEmpty(self):

return True

class Operation(Expression):

def __init__(self, left, op, right):

self.left = left

self.op = op

self.right = right

def Evaluate(self, env, defs):

if self.op == '||' or self.op == ',':

return self.left.Evaluate(env, defs) or self.right.Evaluate(env, defs)

elif self.op == 'if':

return False

elif self.op == '==':

inter = self.left.GetOutcomes(env, defs).Intersect(self.right.GetOutcomes(env, defs))

return not inter.IsEmpty()

else:

assert self.op == '&&'

return self.left.Evaluate(env, defs) and self.right.Evaluate(env, defs)

def GetOutcomes(self, env, defs):

if self.op == '||' or self.op == ',':

return self.left.GetOutcomes(env, defs).Union(self.right.GetOutcomes(env, defs))

elif self.op == 'if':

if self.right.Evaluate(env, defs): return self.left.GetOutcomes(env, defs)

else: return Nothing()

else:

assert self.op == '&&'

return self.left.GetOutcomes(env, defs).Intersect(self.right.GetOutcomes(env, defs))

def IsAlpha(str):

for char in str:

if not (char.isalpha() or char.isdigit() or char == '_'):

return False

return True

class Tokenizer(object):

"""A simple string tokenizer that chops expressions into variables,

parens and operators"""

def __init__(self, expr):

self.index = 0

self.expr = expr

self.length = len(expr)

self.tokens = None

def Current(self, length = 1):

if not self.HasMore(length): return ""

return self.expr[self.index:self.index+length]

def HasMore(self, length = 1):

return self.index < self.length + (length - 1)

def Advance(self, count = 1):

self.index = self.index + count

def AddToken(self, token):

self.tokens.append(token)

def SkipSpaces(self):

while self.HasMore() and self.Current().isspace():

self.Advance()

def Tokenize(self):

self.tokens = [ ]