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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
# -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Run examples as unit tests.
This module executes examples as unit tests, thus ensuring they at least
can be executed with current taskflow. For examples with deterministic
output, the output can be put to file with same name and '.out.txt'
extension; then it will be checked that output did not change.
When this module is used as main module, output for all examples are
generated. Please note that this will break tests as output for most
examples is indeterministic.
"""
import os
import re
import subprocess
import sys
import taskflow.test
ROOT_DIR = os.path.abspath(
os.path.dirname(
os.path.dirname(
os.path.dirname(__file__))))
def root_path(*args):
return os.path.join(ROOT_DIR, *args)
def run_example(name):
path = root_path('taskflow', 'examples', '%s.py' % name)
obj = subprocess.Popen([sys.executable, path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output = obj.communicate()
stdout = output[0].decode()
stderr = output[1].decode()
rc = obj.wait()
if rc != 0:
raise RuntimeError('Example %s failed, return code=%s\n'
'<<<Begin captured STDOUT>>>\n%s'
'<<<End captured STDOUT>>>\n'
'<<<Begin captured STDERR>>>\n%s'
'<<<End captured STDERR>>>'
% (name, rc, stdout, stderr))
return stdout
def expected_output_path(name):
return root_path('taskflow', 'examples', '%s.out.txt' % name)
def list_examples():
examples_dir = root_path('taskflow', 'examples')
for filename in os.listdir(examples_dir):
name, ext = os.path.splitext(filename)
if ext == ".py" and 'utils' not in name.lower():
yield name
class ExamplesTestCase(taskflow.test.TestCase):
maxDiff = None # sky's the limit
uuid_re = re.compile('XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX'
.replace('X', '[0-9a-f]'))
@classmethod
def update(cls):
def add_test_method(name, method_name):
def test_example(self):
self._check_example(name)
test_example.__name__ = method_name
setattr(cls, method_name, test_example)
for name in list_examples():
add_test_method(name, 'test_%s' % name)
def _check_example(self, name):
output = run_example(name)
eop = expected_output_path(name)
if os.path.isfile(eop):
with open(eop) as f:
expected_output = f.read()
# NOTE(imelnikov): on each run new uuid is generated, so we just
# replace them with some constant string
output = self.uuid_re.sub('<SOME UUID>', output)
expected_output = self.uuid_re.sub('<SOME UUID>', expected_output)
self.assertEqual(output, expected_output)
ExamplesTestCase.update()
def make_output_files():
"""Generate output files for all examples."""
for name in list_examples():
output = run_example(name)
with open(expected_output_path(name), 'w') as f:
f.write(output)
if __name__ == '__main__':
make_output_files()
|
