|
| 1 | +from __future__ import division |
| 2 | +from __future__ import print_function |
| 3 | + |
| 4 | +import glob |
| 5 | +import os |
| 6 | +import subprocess |
| 7 | +import sys |
| 8 | +from hashlib import sha256 |
| 9 | + |
| 10 | +import pydot_ng as pydot |
| 11 | + |
| 12 | +PY3 = not sys.version_info < (3, 0, 0) |
| 13 | + |
| 14 | +if PY3: |
| 15 | + NULL_SEP = b'' |
| 16 | + xrange = range |
| 17 | +else: |
| 18 | + NULL_SEP = '' |
| 19 | + bytes = str |
| 20 | + |
| 21 | +DOT_BINARY_PATH = pydot.find_graphviz()['dot'] |
| 22 | + |
| 23 | + |
| 24 | +TEST_DIR = os.path.dirname(__file__) |
| 25 | +REGRESSION_TESTS_DIR = os.path.join(TEST_DIR, "graphs") |
| 26 | +MY_REGRESSION_TESTS_DIR = os.path.join(TEST_DIR, "my_tests") |
| 27 | + |
| 28 | +TESTS_DIRS = ( |
| 29 | + ("regressions", REGRESSION_TESTS_DIR), |
| 30 | + ("my_regressions", MY_REGRESSION_TESTS_DIR), |
| 31 | +) |
| 32 | + |
| 33 | + |
| 34 | +def list_dots(path): |
| 35 | + searchpath = os.path.join(path, "*.dot") |
| 36 | + return [f for f in glob.glob(searchpath)] |
| 37 | + |
| 38 | + |
| 39 | +def pytest_generate_tests(metafunc): |
| 40 | + idlist = [] |
| 41 | + argvalues = [] |
| 42 | + for name, dir in TESTS_DIRS: |
| 43 | + for filepath in list_dots(dir): |
| 44 | + filename = os.path.basename(filepath) |
| 45 | + idlist.append("{}-{}".format(name, filename)) |
| 46 | + argvalues.append((filepath,)) |
| 47 | + metafunc.parametrize( |
| 48 | + argnames=["filepath"], |
| 49 | + argvalues=argvalues, |
| 50 | + ids=idlist, |
| 51 | + scope="function", |
| 52 | + ) |
| 53 | + |
| 54 | + |
| 55 | +def test_render_and_compare_dot_files(filepath): |
| 56 | + def _render_with_graphviz(filename): |
| 57 | + p = subprocess.Popen( |
| 58 | + (DOT_BINARY_PATH, '-Tjpe'), |
| 59 | + cwd=os.path.dirname(filename), |
| 60 | + stdin=open(filename, 'rt'), |
| 61 | + stderr=subprocess.PIPE, stdout=subprocess.PIPE) |
| 62 | + |
| 63 | + stdout = p.stdout |
| 64 | + |
| 65 | + stdout_output = list() |
| 66 | + while True: |
| 67 | + data = stdout.read() |
| 68 | + if not data: |
| 69 | + break |
| 70 | + stdout_output.append(data) |
| 71 | + stdout.close() |
| 72 | + |
| 73 | + if stdout_output: |
| 74 | + stdout_output = NULL_SEP.join(stdout_output) |
| 75 | + |
| 76 | + # this returns a status code we should check |
| 77 | + p.wait() |
| 78 | + |
| 79 | + return sha256(stdout_output).hexdigest() |
| 80 | + |
| 81 | + def _render_with_pydot(filename): |
| 82 | + g = pydot.graph_from_dot_file(filename) |
| 83 | + if not isinstance(g, list): |
| 84 | + g = [g] |
| 85 | + jpe_data = NULL_SEP.join([_g.create(format='jpe') for _g in g]) |
| 86 | + return sha256(jpe_data).hexdigest() |
| 87 | + parsed_data_hexdigest = _render_with_pydot(filepath) |
| 88 | + original_data_hexdigest = _render_with_graphviz(filepath) |
| 89 | + |
| 90 | + assert original_data_hexdigest == parsed_data_hexdigest |
0 commit comments