test_conformance/run_conformance.py

#! /usr/bin/python

#/******************************************************************
#//
#//  OpenCL Conformance Tests
#//
#//  Copyright:  (c) 2008-2009 by Apple Inc. All Rights Reserved.
#//
#******************************************************************/

import os, re, sys, subprocess, time, commands, tempfile, math, string

DEBUG = 0

log_file_name = "opencl_conformance_results_" + time.strftime("%Y-%m-%d_%H-%M", time.localtime())+ ".log"
process_pid = 0

# The amount of time between printing a "." (if no output from test) or ":" (if output)
#  to the screen while the tests are running.
seconds_between_status_updates = 60*60*24*7  # effectively never

# Help info
def write_help_info() :
 print("run_conformance.py test_list [CL_DEVICE_TYPE(s) to test] [partial-test-names, ...] [log=path/to/log/file/]")
 print(" test_list - the .csv file containing the test names and commands to run the tests.")
 print(" [partial-test-names, ...] - optional partial strings to select a subset of the tests to run.")
 print(" [CL_DEVICE_TYPE(s) to test] - list of CL device types to test, default is CL_DEVICE_TYPE_DEFAULT.")
 print(" [log=path/to/log/file/] - provide a path for the test log file, default is in the current directory.")
 print("   (Note: spaces are not allowed in the log file path.")


# Get the time formatted nicely
def get_time() :
 return time.strftime("%d-%b %H:%M:%S", time.localtime())

# Write text to the screen and the log file
def write_screen_log(text) :
 global log_file
 print(text)
 log_file.write(text+"\n")

# Load the tests from a csv formated file of the form name,command
def get_tests(filename, devices_to_test):
 tests = []
 if (os.path.exists(filename) == False):
  print("FAILED: test_list \"" + filename + "\" does not exist.")
  print("")
  write_help_info()
  sys.exit(-1)
 file = open(filename, 'r')
 for line in file.readlines():
  comment = re.search("^#.*", line)
  if (comment):
   continue
  device_specific_match = re.search("^\s*(.+?)\s*,\s*(.+?)\s*,\s*(.+?)\s*$", line)
  if (device_specific_match):
   if (device_specific_match.group(1) in devices_to_test):
    test_path = string.replace(device_specific_match.group(3), '/', os.sep)
    test_name = string.replace(device_specific_match.group(2), '/', os.sep)
    tests.append((test_name, test_path))
   else:
    print("Skipping " + device_specific_match.group(2) + " because " + device_specific_match.group(1) + " is not in the list of devices to test.")
   continue
  match = re.search("^\s*(.+?)\s*,\s*(.+?)\s*$", line)
  if (match):
   test_path = string.replace(match.group(2), '/', os.sep)
   test_name = string.replace(match.group(1), '/', os.sep)
   tests.append((test_name, test_path))
 return tests


def run_test_checking_output(current_directory, test_dir, log_file):
  global process_pid, seconds_between_status_updates
  failures_this_run = 0
  start_time = time.time()
  # Create a temporary file for capturing the output from the test
  (output_fd, output_name) = tempfile.mkstemp()
  if ( not os.path.exists(output_name)) :
    write_screen_log("\n           ==> ERROR: could not create temporary file %s ." % output_name)
    os.close(output_fd)
    return -1
  # Execute the test
  program_to_run = test_dir_without_args = test_dir.split(None, 1)[0]
  if ( os.sep == '\\' ) : program_to_run += ".exe"
  if (os.path.exists(current_directory + os.sep + program_to_run)) :
    os.chdir(os.path.dirname(current_directory+os.sep+test_dir_without_args) )
    try:
      if (DEBUG): p = subprocess.Popen("", stderr=subprocess.STDOUT, stdout=subprocess.PIPE, shell=True)
      else :  p = subprocess.Popen(current_directory + os.sep + test_dir, stderr=output_fd, stdout=output_fd, shell=True)
    except OSError:
      write_screen_log("\n           ==> ERROR: failed to execute test. Failing test. : " + str(OSError))
      os.close(output_fd)
      return -1
  else:
    write_screen_log("\n           ==> ERROR: test file (" + current_directory + os.sep + program_to_run +") does not exist.  Failing test.")
    os.close(output_fd)
    return -1
  # Set the global pid so we can kill it if this is aborted
  process_pid = p.pid
  # Read one character at a time from the temporary output file while the process is running.
  # When we get an end-of-line, look for errors and write the results to the log file.
  # This allows us to process the file as it is being produced.
  # Keep track of the state for reading
  # Whether we are done, if we have more to read, and where in the file we last read
  done = False
  more_to_read = True
  pointer = 0
  pointer_at_last_user_update = 0
  output_this_run = False
  try:
    read_output = open(output_name, 'r')
  except IOError:
    write_screen_log("\n           ==> ERROR: could not open output file from test.")
    os.close(output_fd)
    return -1
  line = ""
  while (not done or more_to_read):
    os.fsync(output_fd)
    # Determine if we should display some output
    elapsed_time = (time.time() - start_time)
    if (elapsed_time > seconds_between_status_updates):
      start_time = time.time()
      # If we've received output from the test since the last update, display a #
      if (pointer != pointer_at_last_user_update):
        sys.stdout.write(":")
      else:
        sys.stdout.write(".")
      pointer_at_last_user_update = pointer
      sys.stdout.flush()
    # Check if we're done
    p.poll()
    if (not done and p.returncode != None):
      if (p.returncode < 0):
        if (not output_this_run):
         print ""
         output_this_run = True
        write_screen_log("           ==> ERROR: test killed/crashed: " + str(p.returncode)+ ".")
      done = True
    # Try reading
    try:
      read_output.seek(pointer)
      char_read = read_output.read(1)
    except IOError:
      time.sleep(1)
      continue
    # If we got a full line then process it
    if (char_read == "\n"):
      # Look for failures and report them as such
      match = re.search(".*(FAILED|ERROR).*", line)
      if (match):
        if (not output_this_run):
         print ""
         output_this_run = True
        print("           ==> " + line.replace('\n',''))
      match = re.search(".*FAILED.*", line)
      if (match):
        failures_this_run = failures_this_run + 1
      match = re.search(".*(PASSED).*", line)
      if (match):
       if (not output_this_run):
        print ""
        output_this_run = True
       print("               " + line.replace('\n',''))
      # Write it to the log
      log_file.write("     " + line +"\n")
      log_file.flush()
      line = ""
      pointer = pointer + 1
    # If we are at the end of the file, then re-open it to get new data
    elif (char_read == ""):
      more_to_read = False
      read_output.close()
      time.sleep(1)
      try:
        os.fsync(output_fd)
        read_output = open(output_name, 'r')
        # See if there is more to read. This happens if the process ends and we have data left.
        read_output.seek(pointer)
        if (read_output.read(1) != ""):
          more_to_read = True
      except IOError:
        write_screen_log("\n           ==> ERROR: could not reopen output file from test.")
        return -1
        done = True
    else:
      line = line + char_read
      pointer = pointer + 1
  # Now we are done, so write out any remaining data in the file:
  # This should only happen if the process exited with an error.
  os.fsync(output_fd)
  while (read_output.read(1) != ""):
    log_file.write(read_output.read(1))
  # Return the total number of failures
  if (p.returncode == 0 and failures_this_run > 0):
   write_screen_log("\n           ==> ERROR: Test returned 0, but number of FAILED lines reported is " + str(failures_this_run) +".")
   return failures_this_run
  return p.returncode


def run_tests(tests) :
  global curent_directory
  global process_pid
  # Run the tests
  failures = 0
  previous_test = None
  test_number = 1
  for test in tests:
   # Print the name of the test we're running and the time
   (test_name, test_dir) = test
   if (test_dir != previous_test):
    print("==========   " + test_dir)
    log_file.write("========================================================================================\n")
    log_file.write("========================================================================================\n")
    log_file.write("(" + get_time() + ")     Running Tests: " + test_dir +"\n")
    log_file.write("========================================================================================\n")
    log_file.write("========================================================================================\n")
    previous_test = test_dir
   print("("+get_time()+")     BEGIN  " + test_name.ljust(40) +": "),
   log_file.write("     ----------------------------------------------------------------------------------------\n")
   log_file.write("     (" + get_time() + ")     Running Sub Test: " + test_name + "\n")
   log_file.write("     ----------------------------------------------------------------------------------------\n")
   log_file.flush()
   sys.stdout.flush()

   # Run the test
   result = 0
   start_time = time.time()
   try:
    process_pid = 0
    result = run_test_checking_output(current_directory, test_dir, log_file)
   except KeyboardInterrupt:
    # Catch an interrupt from the user
    write_screen_log("\nFAILED: Execution interrupted.  Killing test process, but not aborting full test run.")
    os.kill(process_pid, 9)
    answer = raw_input("Abort all tests? (y/n)")
    if (answer.find("y") != -1):
     write_screen_log("\nUser chose to abort all tests.")
     log_file.close()
     sys.exit(-1)
    else:
     write_screen_log("\nUser chose to continue with other tests. Reporting this test as failed.")
     result = 1
   run_time = (time.time() - start_time)

   # Move print the finish status
   if (result == 0):
    print("("+get_time()+")     PASSED " + test_name.ljust(40) +": (" + str(int(run_time)).rjust(3) + "s, test " + str(test_number).rjust(3) + os.sep + str(len(tests)) +")"),
   else:
    print("("+get_time()+")     FAILED " + test_name.ljust(40) +": (" + str(int(run_time)).rjust(3) + "s, test " + str(test_number).rjust(3) + os.sep + str(len(tests)) +")"),

   test_number = test_number + 1
   log_file.write("     ----------------------------------------------------------------------------------------\n")
   log_file.flush()

   print("")
   if (result != 0):
    log_file.write("  *******************************************************************************************\n")
    log_file.write("  *  ("+get_time()+")     Test " + test_name + " ==> FAILED: " + str(result)+"\n")
    log_file.write("  *******************************************************************************************\n")
    failures = failures + 1
   else:
    log_file.write("     ("+get_time()+")     Test " + test_name +" passed in " + str(run_time) + "s\n")

   log_file.write("     ----------------------------------------------------------------------------------------\n")
   log_file.write("\n")
  return failures





# ########################
# Begin OpenCL conformance run script
# ########################

if (len(sys.argv) < 2):
 write_help_info()
 sys.exit(-1)


current_directory = os.getcwd()
# Open the log file
for arg in sys.argv:
 match = re.search("log=(\S+)", arg)
 if (match):
  log_file_name = match.group(1).rstrip('/') + os.sep + log_file_name
try:
 log_file = open(log_file_name, "w")
except IOError:
 print "Could not open log file " + log_file_name

# Determine which devices to test
device_types = ["CL_DEVICE_TYPE_DEFAULT", "CL_DEVICE_TYPE_CPU", "CL_DEVICE_TYPE_GPU", "CL_DEVICE_TYPE_ACCELERATOR", "CL_DEVICE_TYPE_ALL"]
devices_to_test = []
for device in device_types:
 if device in sys.argv[2:]:
  devices_to_test.append(device)
if (len(devices_to_test) == 0):
 devices_to_test = ["CL_DEVICE_TYPE_DEFAULT"]
write_screen_log("Testing on: " + str(devices_to_test))

# Get the tests
tests = get_tests(sys.argv[1], devices_to_test)

# If tests are specified on the command line then run just those ones
tests_to_use = []
num_of_patterns_to_match = 0
for arg in sys.argv[2:]:
 if arg in device_types:
  continue
 if re.search("log=(\S+)", arg):
  continue
 num_of_patterns_to_match = num_of_patterns_to_match + 1
 found_it = False
 for test in tests:
  (test_name, test_dir) = test
  if (test_name.find(arg) != -1 or test_dir.find(arg) != -1):
   found_it = True
   if (test not in tests_to_use):
    tests_to_use.append(test)
 if (found_it == False):
  print("Failed to find a test matching " + arg)
if (len(tests_to_use) == 0):
 if (num_of_patterns_to_match > 0):
  print("FAILED: Failed to find any tests matching the given command-line options.")
  print("")
  write_help_info()
  sys.exit(-1)
else:
 tests = tests_to_use[:]

write_screen_log("Test execution arguments: " + str(sys.argv))
write_screen_log("Logging to file " + log_file_name +".")
write_screen_log("Loaded tests from " + sys.argv[1] + ", total of " + str(len(tests)) + " tests selected to run:")
for (test_name, test_command) in tests:
 write_screen_log(test_name.ljust(50) + " (" + test_command +")")

# Run the tests
total_failures = 0
for device_to_test in devices_to_test:
 os.environ['CL_DEVICE_TYPE'] = device_to_test
 write_screen_log("========================================================================================")
 write_screen_log("========================================================================================")
 write_screen_log(("Setting CL_DEVICE_TYPE to " + device_to_test).center(90))
 write_screen_log("========================================================================================")
 write_screen_log("========================================================================================")
 failures = run_tests(tests)
 write_screen_log("========================================================================================")
 if (failures == 0):
  write_screen_log(">> TEST on " + device_to_test + " PASSED")
 else:
  write_screen_log(">> TEST on " + device_to_test + " FAILED (" + str(failures) + " FAILURES)")
 write_screen_log("========================================================================================")
 total_failures = total_failures + failures

write_screen_log("("+get_time()+") Testing complete.  " + str(total_failures) + " failures for " + str(len(tests)) + " tests.")
log_file.close()