source: tests/pybin/tools.py @ a5121bf

aaron-thesisarm-ehcleanup-dtorsdeferred_resndemanglerenumforall-pointer-decayjacob/cs343-translationjenkins-sandboxnew-astnew-ast-unique-exprno_listpersistent-indexerpthread-emulationqualifiedEnum
Last change on this file since a5121bf was a5121bf, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

Tests can now be run from installed binaries or tree binaries

  • Property mode set to 100644
File size: 6.6 KB
Line 
1from __future__ import print_function
2
3import __main__
4import argparse
5import multiprocessing
6import os
7import re
8import signal
9import stat
10import sys
11import fileinput
12
13from pybin import settings
14from subprocess import Popen, PIPE, STDOUT
15
16################################################################################
17#               shell helpers
18################################################################################
19
20# helper functions to run terminal commands
21def sh(cmd, print2stdout = True, input = None):
22        # add input redirection if needed
23        if input and os.path.isfile(input):
24                cmd += " < %s" % input
25
26        # if this is a dry_run, only print the commands that would be ran
27        if settings.dry_run :
28                print("cmd: %s" % cmd)
29                return 0, None
30
31        # otherwise create a pipe and run the desired command
32        else :
33                proc = Popen(cmd, stdout=None if print2stdout else PIPE, stderr=STDOUT, shell=True)
34                out, err = proc.communicate()
35                return proc.returncode, out
36
37def is_ascii(fname):
38        if not os.path.isfile(fname):
39                return False
40
41        code, out = sh("file %s" % fname, print2stdout = False)
42        if code != 0:
43                return False
44
45        match = re.search(".*: (.*)", out)
46
47        if not match:
48                return False
49
50        return match.group(1).startswith("ASCII text")
51
52# Remove 1 or more files silently
53def rm( files ):
54        if isinstance( files, basestring ):
55                sh("rm -f %s > /dev/null 2>&1" % files )
56        else:
57                for file in files:
58                        sh("rm -f %s > /dev/null 2>&1" % file )
59
60# Create 1 or more directory
61def mkdir( files ):
62        if isinstance( files, basestring ):
63                sh("mkdir -p %s" % os.path.dirname(files) )
64        else:
65                for file in files:
66                        sh("mkdir -p %s" % os.path.dirname(file) )
67
68
69def chdir( dest = __main__.__file__ ):
70        abspath = os.path.abspath(dest)
71        dname = os.path.dirname(abspath)
72        os.chdir(dname)
73
74# diff two files
75def diff( lhs, rhs ):
76        # diff the output of the files
77        diff_cmd = ("diff --ignore-all-space "
78                                "--ignore-blank-lines "
79                                "--old-group-format='\t\tmissing lines :\n"
80                                "%%<' \\\n"
81                                "--new-group-format='\t\tnew lines :\n"
82                                "%%>' \\\n"
83                                "--unchanged-group-format='%%=' \\"
84                                "--changed-group-format='\t\texpected :\n"
85                                "%%<"
86                                "\t\tgot :\n"
87                                "%%>\n' \\\n"
88                                "--new-line-format='\t\t%%dn\t%%L' \\\n"
89                                "--old-line-format='\t\t%%dn\t%%L' \\\n"
90                                "--unchanged-line-format='' \\\n"
91                                "%s %s")
92
93        # fetch return code and error from the diff command
94        return sh(diff_cmd % (lhs, rhs), False)
95
96# call make
97def make(target, flags = '', redirects = '', error_file = None, silent = False):
98        test_param = """test="%s" """ % (error_file) if error_file else ''
99        cmd = ' '.join([
100                settings.make,
101                '-s' if silent else '',
102                test_param,
103                settings.debug.flags,
104                settings.install.flags,
105                flags,
106                target,
107                redirects
108        ])
109        return sh(cmd)
110
111def which(program):
112    import os
113    def is_exe(fpath):
114        return os.path.isfile(fpath) and os.access(fpath, os.X_OK)
115
116    fpath, fname = os.path.split(program)
117    if fpath:
118        if is_exe(program):
119            return program
120    else:
121        for path in os.environ["PATH"].split(os.pathsep):
122            exe_file = os.path.join(path, program)
123            if is_exe(exe_file):
124                return exe_file
125
126    return None
127################################################################################
128#               file handling
129################################################################################
130
131# helper function to replace patterns in a file
132def file_replace(fname, pat, s_after):
133        file = fileinput.FileInput(fname, inplace=True, backup='.bak')
134        for line in file:
135                print(line.replace(pat, s_after), end='')
136        file.close()
137
138# helper function to check if a files contains only a specific string
139def fileContainsOnly(file, text) :
140        with open(file) as f:
141                ff = f.read().strip()
142                result = ff == text.strip()
143
144                return result;
145
146# check whether or not a file is executable
147def fileIsExecutable(file) :
148        try :
149                fileinfo = os.stat(file)
150                return bool(fileinfo.st_mode & stat.S_IXUSR)
151        except Exception as inst:
152                print(type(inst))    # the exception instance
153                print(inst.args)     # arguments stored in .args
154                print(inst)
155                return False
156
157# transform path to canonical form
158def canonicalPath(path):
159        abspath = os.path.abspath(__main__.__file__)
160        dname = os.path.dirname(abspath)
161        return os.path.join(dname, os.path.normpath(path) )
162
163# compare path even if form is different
164def pathCmp(lhs, rhs):
165        return canonicalPath( lhs ) == canonicalPath( rhs )
166
167# walk all files in a path
168def pathWalk( op ):
169        def step(_, dirname, names):
170                for name in names:
171                        path = os.path.join(dirname, name)
172                        op( path )
173
174        # Start the walk
175        dname = settings.SRCDIR
176        os.path.walk(dname, step, '')
177
178################################################################################
179#               system
180################################################################################
181# count number of jobs to create
182def jobCount( options, tests ):
183        # check if the user already passed in a number of jobs for multi-threading
184        if not options.jobs:
185                make_flags = os.environ.get('MAKEFLAGS')
186                force = bool(make_flags)
187                make_jobs_fds = re.search("--jobserver-(auth|fds)=\s*([0-9]+),([0-9]+)", make_flags) if make_flags else None
188                if make_jobs_fds :
189                        tokens = os.read(int(make_jobs_fds.group(2)), 1024)
190                        options.jobs = len(tokens)
191                        os.write(int(make_jobs_fds.group(3)), tokens)
192                else :
193                        options.jobs = multiprocessing.cpu_count()
194        else :
195                force = True
196
197        # make sure we have a valid number of jobs that corresponds to user input
198        if options.jobs <= 0 :
199                print('ERROR: Invalid number of jobs', file=sys.stderr)
200                sys.exit(1)
201
202        return min( options.jobs, len(tests) ), force
203
204# setup a proper processor pool with correct signal handling
205def setupPool(jobs):
206        original_sigint_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
207        pool = multiprocessing.Pool(jobs)
208        signal.signal(signal.SIGINT, original_sigint_handler)
209
210        return pool
211
212# handle signals in scope
213class SignalHandling():
214        def __enter__(self):
215                # enable signal handling
216                signal.signal(signal.SIGINT, signal.SIG_DFL)
217
218        def __exit__(self, type, value, traceback):
219                # disable signal handling
220                signal.signal(signal.SIGINT, signal.SIG_IGN)
221
222################################################################################
223#               misc
224################################################################################
225
226# check if arguments is yes or no
227def yes_no(string):
228        if string == "yes" :
229                return True
230        if string == "no" :
231                return False
232        raise argparse.ArgumentTypeError(msg)
233        return False
234
235def fancy_print(text):
236        column = which('column')
237        if column:
238                cmd = "%s 2> /dev/null" % column
239                print(cmd)
240                proc = Popen(cmd, stdin=PIPE, stderr=None, shell=True)
241                proc.communicate(input=text)
242        else:
243                print(text)
Note: See TracBrowser for help on using the repository browser.