1#!/usr/bin/python3
2
3import subprocess
4import argparse
5import difflib
6import filecmp
7import fnmatch
8import json
9import sys
10import re
11import os
12
13fmtr_class = argparse.ArgumentDefaultsHelpFormatter
14parser = argparse.ArgumentParser(prog = 'nasm-t.py',
15                                 formatter_class=fmtr_class)
16
17parser.add_argument('-d', '--directory',
18                    dest = 'dir', default = './travis/test',
19                    help = 'Directory with tests')
20
21parser.add_argument('--nasm',
22                    dest = 'nasm', default = './nasm',
23                    help = 'Nasm executable to use')
24
25parser.add_argument('--hexdump',
26                    dest = 'hexdump', default = '/usr/bin/hexdump',
27                    help = 'Hexdump executable to use')
28
29sp = parser.add_subparsers(dest = 'cmd')
30for cmd in ['run']:
31    spp = sp.add_parser(cmd, help = 'Run test cases')
32    spp.add_argument('-t', '--test',
33                     dest = 'test',
34                     help = 'Run the selected test only',
35                     required = False)
36
37for cmd in ['list']:
38    spp = sp.add_parser(cmd, help = 'List test cases')
39
40for cmd in ['update']:
41    spp = sp.add_parser(cmd, help = 'Update test cases with new compiler')
42    spp.add_argument('-t', '--test',
43                     dest = 'test',
44                     help = 'Update the selected test only',
45                     required = False)
46
47args = parser.parse_args()
48
49if args.cmd == None:
50    parser.print_help()
51    sys.exit(1)
52
53def read_stdfile(path):
54    with open(path, "rb") as f:
55        data = f.read().decode("utf-8").strip("\n")
56        f.close()
57        return data
58
59#
60# Check if descriptor has mandatory fields
61def is_valid_desc(desc):
62    if desc == None:
63        return False
64    if 'description' not in desc:
65        return false
66    return True
67
68#
69# Expand ref/id in descriptors array
70def expand_templates(desc_array):
71    desc_ids = { }
72    for d in desc_array:
73        if 'id' in d:
74            desc_ids[d['id']] = d
75    for i, d in enumerate(desc_array):
76        if 'ref' in d and d['ref'] in desc_ids:
77            ref = desc_ids[d['ref']]
78            own = d.copy()
79            desc_array[i] = ref.copy()
80            for k, v in own.items():
81                desc_array[i][k] = v
82            del desc_array[i]['id']
83    return desc_array
84
85def prepare_desc(desc, basedir, name, path):
86    if not is_valid_desc(desc):
87        return False
88    #
89    # Put private fields
90    desc['_base-dir'] = basedir
91    desc['_json-file'] = name
92    desc['_json-path'] = path
93    desc['_test-name'] = basedir + os.sep + name[:-5]
94    #
95    # If no target provided never update
96    if 'target' not in desc:
97        desc['target'] = []
98        desc['update'] = 'false'
99    #
100    # Which code to expect when nasm finishes
101    desc['_wait'] = 0
102    if 'error' in desc:
103        if desc['error'] == 'expected':
104            desc['_wait'] = 1
105    #
106    # Walk over targets and generate match templates
107    # if were not provided yet
108    for d in desc['target']:
109        if 'output' in d and not 'match' in d:
110            d['match'] = d['output'] + ".t"
111    return True
112
113def read_json(path):
114    desc = None
115    try:
116        with open(path, "rb") as f:
117            try:
118                desc = json.loads(f.read().decode("utf-8").strip("\n"))
119            except:
120                desc = None
121            finally:
122                f.close()
123    except:
124        pass
125    return desc
126
127def read_desc(basedir, name):
128    path = basedir + os.sep + name
129    desc = read_json(path)
130    desc_array = []
131    if type(desc) == dict:
132        if prepare_desc(desc, basedir, name, path) == True:
133            desc_array += [desc]
134    elif type(desc) == list:
135        expand_templates(desc)
136        for de in desc:
137            if prepare_desc(de, basedir, name, path) == True:
138                desc_array += [de]
139    return desc_array
140
141def collect_test_desc_from_file(path):
142    if not fnmatch.fnmatch(path, '*.json'):
143        path += '.json'
144    basedir = os.path.dirname(path)
145    filename = os.path.basename(path)
146    return read_desc(basedir, filename)
147
148def collect_test_desc_from_dir(basedir):
149    desc_array = []
150    if os.path.isdir(basedir):
151        for filename in os.listdir(basedir):
152            if os.path.isdir(basedir + os.sep + filename):
153                desc_array += collect_test_desc_from_dir(basedir + os.sep + filename)
154            elif fnmatch.fnmatch(filename, '*.json'):
155                desc = read_desc(basedir, filename)
156                if desc == None:
157                    continue
158                desc_array += desc
159    return desc_array
160
161if args.cmd == 'list':
162    fmt_entry = '%-32s %s'
163    desc_array = collect_test_desc_from_dir(args.dir)
164    print(fmt_entry % ('Name', 'Description'))
165    for desc in desc_array:
166        print(fmt_entry % (desc['_test-name'], desc['description']))
167
168def test_abort(test, message):
169    print("\t%s: %s" % (test, message))
170    print("=== Test %s ABORT ===" % (test))
171    sys.exit(1)
172    return False
173
174def test_fail(test, message):
175    print("\t%s: %s" % (test, message))
176    print("=== Test %s FAIL ===" % (test))
177    return False
178
179def test_skip(test, message):
180    print("\t%s: %s" % (test, message))
181    print("=== Test %s SKIP ===" % (test))
182    return True
183
184def test_over(test):
185    print("=== Test %s ERROR OVER ===" % (test))
186    return True
187
188def test_pass(test):
189    print("=== Test %s PASS ===" % (test))
190    return True
191
192def test_updated(test):
193    print("=== Test %s UPDATED ===" % (test))
194    return True
195
196def run_hexdump(path):
197    p = subprocess.Popen([args.hexdump, "-C", path],
198                         stdout = subprocess.PIPE,
199                         close_fds = True)
200    if p.wait() == 0:
201        return p
202    return None
203
204def show_std(stdname, data):
205    print("\t--- %s" % (stdname))
206    for i in data.split("\n"):
207        print("\t%s" % i)
208    print("\t---")
209
210def cmp_std(test, data_name, data, match):
211    match_data = read_stdfile(match)
212    if match_data == None:
213        return test_fail(test, "Can't read " + match)
214    if data != match_data:
215        print("\t--- %s" % (data_name))
216        for i in data.split("\n"):
217            print("\t%s" % i)
218        print("\t--- %s" % (match))
219        for i in match_data.split("\n"):
220            print("\t%s" % i)
221
222        diff = difflib.unified_diff(data.split("\n"), match_data.split("\n"),
223                                    fromfile = data_name, tofile = match)
224        for i in diff:
225            print("\t%s" % i.strip("\n"))
226        print("\t---")
227        return False
228    return True
229
230def show_diff(test, patha, pathb):
231    pa = run_hexdump(patha)
232    pb = run_hexdump(pathb)
233    if pa == None or pb == None:
234        return test_fail(test, "Can't create dumps")
235    sa = pa.stdout.read().decode("utf-8").strip("\n")
236    sb = pb.stdout.read().decode("utf-8").strip("\n")
237    print("\t--- hexdump %s" % (patha))
238    for i in sa.split("\n"):
239        print("\t%s" % i)
240    print("\t--- hexdump %s" % (pathb))
241    for i in sb.split("\n"):
242        print("\t%s" % i)
243    pa.stdout.close()
244    pb.stdout.close()
245
246    diff = difflib.unified_diff(sa.split("\n"), sb.split("\n"),
247                                fromfile = patha, tofile = pathb)
248    for i in diff:
249        print("\t%s" % i.strip("\n"))
250    print("\t---")
251    return True
252
253def prepare_run_opts(desc):
254    opts = []
255
256    if 'format' in desc:
257        opts += ['-f', desc['format']]
258    if 'option' in desc:
259        opts += desc['option'].split(" ")
260    for t in desc['target']:
261        if 'output' in t:
262            if 'option' in t:
263                opts += t['option'].split(" ") + [desc['_base-dir'] + os.sep + t['output']]
264            else:
265                opts += ['-o', desc['_base-dir'] + os.sep + t['output']]
266        if 'stdout' in t or 'stderr' in t:
267            if 'option' in t:
268                opts += t['option'].split(" ")
269    if 'source' in desc:
270        opts += [desc['_base-dir'] + os.sep + desc['source']]
271    return opts
272
273def exec_nasm(desc):
274    print("\tProcessing %s" % (desc['_test-name']))
275    opts = [args.nasm] + prepare_run_opts(desc)
276
277    print("\tExecuting %s" % (" ".join(opts)))
278    pnasm = subprocess.Popen(opts,
279                             stdout = subprocess.PIPE,
280                             stderr = subprocess.PIPE,
281                             close_fds = True)
282    if pnasm == None:
283        test_fail(desc['_test-name'], "Unable to execute test")
284        return None
285    wait_rc = pnasm.wait();
286
287    stdout = pnasm.stdout.read().decode("utf-8").strip("\n")
288    stderr = pnasm.stderr.read().decode("utf-8").strip("\n")
289    pnasm.stdout.close()
290    pnasm.stderr.close()
291
292    if desc['_wait'] != wait_rc:
293        if stdout != "":
294            show_std("stdout", stdout)
295        if stderr != "":
296            show_std("stderr", stderr)
297        test_fail(desc['_test-name'],
298                  "Unexpected ret code: " + str(wait_rc))
299        return None, None, None
300    return pnasm, stdout, stderr
301
302def test_run(desc):
303    print("=== Running %s ===" % (desc['_test-name']))
304
305    pnasm, stdout, stderr = exec_nasm(desc)
306    if pnasm == None:
307        return False
308
309    for t in desc['target']:
310        if 'output' in t:
311            output = desc['_base-dir'] + os.sep + t['output']
312            match = desc['_base-dir'] + os.sep + t['match']
313            if desc['_wait'] == 1:
314                continue
315            print("\tComparing %s %s" % (output, match))
316            if filecmp.cmp(match, output) == False:
317                show_diff(desc['_test-name'], match, output)
318                return test_fail(desc['_test-name'], match + " and " + output + " files are different")
319        elif 'stdout' in t:
320            print("\tComparing stdout")
321            match = desc['_base-dir'] + os.sep + t['stdout']
322            if cmp_std(desc['_test-name'], 'stdout', stdout, match) == False:
323                return test_fail(desc['_test-name'], "Stdout mismatch")
324            else:
325                stdout = ""
326        elif 'stderr' in t:
327            print("\tComparing stderr")
328            match = desc['_base-dir'] + os.sep + t['stderr']
329            if cmp_std(desc['_test-name'], 'stderr', stderr, match) == False:
330                return test_fail(desc['_test-name'], "Stderr mismatch")
331            else:
332                stderr = ""
333
334    if stdout != "":
335        show_std("stdout", stdout)
336        return test_fail(desc['_test-name'], "Stdout is not empty")
337
338    if stderr != "":
339        show_std("stderr", stderr)
340        return test_fail(desc['_test-name'], "Stderr is not empty")
341
342    return test_pass(desc['_test-name'])
343
344#
345# Compile sources and generate new targets
346def test_update(desc):
347    print("=== Updating %s ===" % (desc['_test-name']))
348
349    if 'update' in desc and desc['update'] == 'false':
350        return test_skip(desc['_test-name'], "No output provided")
351
352    pnasm, stdout, stderr = exec_nasm(desc)
353    if pnasm == None:
354        return False
355
356    for t in desc['target']:
357        if 'output' in t:
358            output = desc['_base-dir'] + os.sep + t['output']
359            match = desc['_base-dir'] + os.sep + t['match']
360            print("\tMoving %s to %s" % (output, match))
361            os.rename(output, match)
362        if 'stdout' in t:
363            match = desc['_base-dir'] + os.sep + t['stdout']
364            print("\tMoving %s to %s" % ('stdout', match))
365            with open(match, "wb") as f:
366                f.write(stdout)
367                f.close()
368        if 'stderr' in t:
369            match = desc['_base-dir'] + os.sep + t['stderr']
370            print("\tMoving %s to %s" % ('stderr', match))
371            with open(match, "wb") as f:
372                f.write(stderr)
373                f.close()
374
375    return test_updated(desc['_test-name'])
376
377if args.cmd == 'run':
378    desc_array = []
379    if args.test == None:
380        desc_array = collect_test_desc_from_dir(args.dir)
381    else:
382        desc_array = collect_test_desc_from_file(args.test)
383        if len(desc_array) == 0:
384            test_abort(args.test, "Can't obtain test descriptors")
385
386    for desc in desc_array:
387        if test_run(desc) == False:
388            if 'error' in desc and desc['error'] == 'over':
389                test_over(desc['_test-name'])
390            else:
391                test_abort(desc['_test-name'], "Error detected")
392
393if args.cmd == 'update':
394    desc_array = []
395    if args.test == None:
396        desc_array = collect_test_desc_from_dir(args.dir)
397    else:
398        desc_array = collect_test_desc_from_file(args.test)
399        if len(desc_array) == 0:
400            test_abort(args.test, "Can't obtain a test descriptors")
401
402    for desc in desc_array:
403        if test_update(desc) == False:
404            if 'error' in desc and desc['error'] == 'over':
405                test_over(desc['_test-name'])
406            else:
407                test_abort(desc['_test-name'], "Error detected")
408