pipelinetests/run.py at master · Statwolf/pipelinetests · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/usr/bin/env python3

from testsuite import params
from pipeline import exports
import traceback
import inspect

tests = params()

total = 0
failed = 0
passed = 0

failedList = []

def defaultCheck(data):
    return

def defaultRaises(error):
    if error == None:
        return

    raise error

for pipelineName in tests:
    print('* Running test suite for ' + pipelineName + ' pipeline\n')

    for info in tests[pipelineName]:
        total = total + 1

        if not 'name' in info:
            info['name'] = 'anonymous test'

        if not 'check' in info:
            info['check'] = defaultCheck

        if not 'raises' in info:
            info['raises'] = defaultRaises

        print('')
        print('---------------------------------------------------------')
        print('Running test: ' + info['name'])
        pipeline = exports[pipelineName](
            datasource=info['datasource'] if 'datasource' in info else None,
            log=info['log'] if 'log' in info else None,
            verbose=info['verbose'] if 'verbose' in info else False
        )

        try:
            if 'before' in info:
                info['before']()

            res = None
            error = None

            try:
                res = pipeline.execute(params=info['params'])
            except Exception as e:
                error = e

            if inspect.isclass(info['raises']) == False:
                info['raises'](error)
            else:
                if type(error) != info['raises']:
                    if error is None:
                        raise Exception('did\'t raises exception: {}'.format(info['raises']))

                    raise Exception('did\'t raises expected exception: but {} instead of {}'.format(type(error), info['raises']))

            info['check'](res)

            passed = passed + 1
        except Exception as e:
            failed = failed + 1
            failedList.append(pipelineName + ' - ' + info['name'])

            print('Test failed!')
            track = traceback.format_exc()
            print(track)
        finally:
            if 'after' in info:
                info['after']()

        print(' ')

    print(' ')

print('* Report')
print('Executed ' + str(total) + ' tests')
print(str(passed) + ' passed')
print(str(failed) + ' failed')
for name in failedList:
    print(' >> ' + name)