build/bin/run_inputs.py

Go to the documentation of this file.

#!/usr/bin/python3
 
import sys
import subprocess
from subprocess import Popen, PIPE, STDOUT
import os
import re
 
cyclus_path =  "/root/.local/include/cyclus/../../bin/cyclus"
input_path = "/cycamore/../input"
    
def main_body(flag):
    files, catalogs, catalognames = get_files(input_path)
    copy_catalogs(catalogs,cyclus_path.strip("cyclus"))
    summ = Summary()
    for name in files :
        file_to_test = TestFile(cyclus_path, name, flag) 
        file_to_test.run()
        summ.add_to_summary(file_to_test)
    clean_catalogs(cyclus_path.strip("cyclus"),catalognames)
    summ.print_summary()
 
def main():
    """This function finds input files, runs them, and prints a summary"""
    flag = check_inputs()
    main_body(flag)
 
def check_inputs():
    """This function checks the input arguments"""
    if len(sys.argv) > 2: 
        print_usage()
        sys.exit(1) 
    elif len(sys.argv) == 2:
        if re.search("-v*",sys.argv[1]):
            flag = sys.argv[1]
        else : 
            print_usage()
            sys.exit(1)
    elif len(sys.argv) == 1 :
        flag = "-v0"
    return flag
 
def print_usage() :
    """This prints the proper way to treat the command line interface"""
    print(""" Usage: python3 run_inputs.py\n 
            Allowed Options : \n 
            -v arg                    output log verbosity. \n 
                                
            Can be text: \n 
                                
              LEV_ERROR (least verbose, default), LEV_WARN,\n 
                                
              LEV_INFO1 (through 5), and LEV_DEBUG1 (through 5).\n 
                                
            Or an integer:\n
                                
              0 (LEV_ERROR equiv) through 11 (LEV_DEBUG5 equiv)\n""")
 
def get_files(path):
    """This function walks the 'path' tree and finds input files"""
    catalogs=[]
    catalognames=[]
    inputs=[]
    for root, dirs, files in os.walk(path, followlinks=True):
        if '.git' in dirs:
            dirs.remove('.git')
        for name in files: 
            if re.search("\.xml",name) :
                if re.search("recipebook",name) or \
                        re.search("facilitycatalog",name):
                    catalogs.append(os.path.join(root,name))
                    catalognames.append(name)
                else : 
                    inputs.append(os.path.join(root, name))
            else :
                files.remove(name)
    print("The catalogs to be moved are:")
    print(catalogs)
    print("The files to be tested are:")
    print(inputs)
    return inputs, catalogs, catalognames
 
def copy_catalogs(catalogs,cyclus_path) :
    """Copies files in the catalogs list to the cyclus executable directory""" 
    for cat in catalogs : 
        p = Popen("cp "+cat+" "+cyclus_path,
                shell=True, stdout=PIPE, stderr=STDOUT)
 
def clean_catalogs(cyclus_path, catalogs) : 
    """Removes previously copied catalogs from executable directory."""
    for cat in catalogs :
        p = Popen("rm "+ os.path.join(cyclus_path,cat),
                shell=True, stdout=PIPE, stderr=STDOUT)
 
class Summary():
    """An object to hold the results of all the tests"""
    def __init__(self):
        self.passed = []
        self.failed = []
 
    def add_to_summary(self, test_file) :
        """Adds a test file to this summary object"""
        if test_file.passed : 
            self.passed.append( test_file.infile )
        else :
            self.failed.append( test_file.infile )
 
    def print_summary(self) :
        """Prints the summary"""
        print("Input files passed = " + str(len(self.passed)))
        print("Input files failed = " + str(len(self.failed)))
        print("Failed input files : ") 
        for test in self.failed : 
            print(test)
 
class TestFile():
    """An object representing the inputxml file to test"""
    def __init__(self, cyclus_path, file_path, flag):
        self.infile = file_path 
        self.cyclus_path = cyclus_path 
        self.passed=True
        self.flag = " "+flag+" "
  
    def run(self):
        """Runs all of the input file tests"""
        output = self.get_outputget_output()
        if self.no_errorsno_errors(output) :
            self.passed = True
        else :
            self.passed = False
 
    def get_output(self):
        """Returns the output from running the FileTest"""
        try :
            p = Popen(self.cyclus_path+" "+ self.infile + self.flag,
                    shell=True, stdout=PIPE, stderr=STDOUT)
            io_tuple = p.communicate()
            output = io_tuple[0]
        except subprocess.CalledProcessError as e:
            print(e)
        return str(output)
        
    def no_errors(self, output):
        """returns true if there were no errors or segfaults running this TestFile"""
        to_ret = True
        print("Input file " + self.infile) 
        if re.search("No such file or directory",output) :
            print("Cyclus executable not found in path.")
        elif re.search("ERROR",output) or re.search("Segmentation fault",output):
            to_ret = False
            print(" resulted in errors: ")
            print(output)
        else :
            print(" passed. ")
        return to_ret 
 
if __name__ == '__main__' : main()