6507626273

Committed 13 Oct 2023 11:25AM UTC coverage: 42.006% (-53.7%) from 95.72%

Build # 6507626273

Build Type

Pull #433

github

Committed by

odscjames

Commit Message

New "Geo" optional dependencies

https://github.com/OpenDataServices/flatten-tool/issues/424

Pull Request Pull Request #433: New "Geo" optional dependencies

Run Details

38 of 38 new or added lines in 6 files covered. (100.0%)

1466 of 3490 relevant lines covered (42.01%)

4.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.57

/flattentool/tests/test_docs.py

import os
import shlex
import subprocess
import sys
import uuid
from os.path import join

import pytest

examples_in_docs_data = []
examples_in_docs_data_geo = []


def _get_examples_in_docs_data():
    global examples_in_docs_data, examples_in_docs_data_geo
    examples_in_docs_data = []
    examples_in_docs_data_geo = []
    for root, dirs, files in os.walk("examples"):
        for filename in files:
            if root.startswith("examples/help/") and sys.version_info[:2] != (3, 8):
                # Different versions of python produce differently formatted help output.
                # We only test in one version, Python 3.8.
                # (Same as we lint code with, so dev's can have one virtual env)
                continue
            if "cmd.txt" in filename:
                if root.startswith("examples/wkt"):
                    examples_in_docs_data_geo.append((root, filename))
                else:
                    examples_in_docs_data.append((root, filename))


_get_examples_in_docs_data()


def test_examples_receipt():
    with open("examples/receipt/source-map/expected.json", "rb") as fp:
        expected = fp.read()
        for expected_filename in [
            "normalised/expected.json",
            "combine-table-into-cafe/expected.json",
            "combine-table-into-cafe-2/expected.json",
        ]:
            with open("examples/receipt/" + expected_filename, "rb") as fp2:
                assert (
                    fp2.read() == expected
                ), "Files differ: examples/receipt/source-map/expected.json, examples/receipt/{}".format(
                    expected_filename
                )


@pytest.mark.parametrize("root, filename", examples_in_docs_data)
def test_example_in_doc(root, filename):
    _test_example_in_doc_worker(root, filename)


@pytest.mark.parametrize("root, filename", examples_in_docs_data_geo)
@pytest.mark.geo
def test_example_in_doc_geo(root, filename):
    _test_example_in_doc_worker(root, filename)


@pytest.mark.parametrize("root, filename", examples_in_docs_data)
def _test_example_in_doc_worker(root, filename):
    if os.path.exists(join(root, "actual")) and os.path.isdir(join(root, "actual")):
        os.rename(join(root, "actual"), join(root, "actual." + str(uuid.uuid4())))
    os.mkdir(join(root, "actual"))
    expected_return_code = 0
    expected_stdout = b""
    if os.path.exists(join(root, "expected_return_code.txt")):
        with open(join(root, "expected_return_code.txt"), "rb") as fp:
            expected_return_code = int(fp.read().strip())
    with open(join(root, filename), "rb") as fp:
        cmds = str(fp.read(), "utf8").strip().split("\n")
        actual_stdout = b""
        actual_stderr = b""
        for cmd in cmds:
            assert cmd.startswith("$ flatten-tool ") or cmd.startswith(
                "$ cat "
            ), "Expected commands to start with '$ flatten-tool'. This doesn't: {}".format(
                cmd
            )
            # Since we are defining all the commands ourselves, this is reasonably safe
            cmd_parts = shlex.split(cmd[len("$ ") :])
            # Include coverage output in the results
            if cmd_parts[0] == "flatten-tool":
                cmd_parts = [
                    "coverage",
                    "run",
                    "--source",
                    "flattentool",
                    "--parallel-mode",
                ] + cmd_parts
            process = subprocess.Popen(
                cmd_parts, stdout=subprocess.PIPE, stderr=subprocess.PIPE
            )
            (cmd_actual_stdout, cmd_actual_stderr) = process.communicate()
            process.wait()
            assert process.returncode == expected_return_code, cmd
            actual_stdout += cmd_actual_stdout or b""
            actual_stderr += cmd_actual_stderr or b""
    if os.path.exists(join(root, "expected")) and os.path.isdir(join(root, "expected")):
        # Create case
        assert len(os.listdir(join(root, "expected"))) == len(
            os.listdir(join(root, "actual"))
        ), "Different number of files. {}".format(cmds)
        for expected_filename in os.listdir(join(root, "expected")):
            assert os.path.exists(
                join(root, "actual", expected_filename)
            ), "File {} was not generated {}".format(expected_filename, cmds)
            with open(join(root, "expected", expected_filename), "rb") as fp_expected:
                with open(join(root, "actual", expected_filename), "rb") as fp_actual:
                    assert _strip(fp_actual.read()) == _strip(
                        fp_expected.read()
                    ), "File {} has unexpected content. {}".format(
                        expected_filename, cmds
                    )
        expected_stdout = b""
    # Flatten case
    if os.path.exists(join(root, "expected.txt")):
        with open(join(root, "expected.txt"), "rb") as fstdout:
            expected_stdout = fstdout.read()
    elif os.path.exists(join(root, "expected.json")):
        with open(join(root, "expected.json"), "rb") as fstdout:
            expected_stdout = fstdout.read()
    elif os.path.exists(join(root, "expected.xml")):
        with open(join(root, "expected.xml"), "rb") as fstdout:
            expected_stdout = fstdout.read()
    if "help" in root:
        # Ignore whitespace differences for help messages
        assert b" ".join(actual_stdout.split()) == b" ".join(expected_stdout.split())
    else:
        assert _strip(actual_stdout) == _strip(
            expected_stdout
        ), "Different stdout: {}".format(cmds)
    expected_stderr = b""
    if os.path.exists(join(root, "expected_stderr_partial.txt")):
        with open(join(root, "expected_stderr_partial.txt"), "rb") as fstderr:
            data = fstderr.read()
        assert data in actual_stderr
    if os.path.exists(join(root, "expected_stderr.json")):
        with open(join(root, "expected_stderr.json"), "rb") as fstderr:
            data = fstderr.read()
            expected_stderr_lines = str(data, "utf8").split("\n")
            for line in expected_stderr_lines:
                if line:
                    expected_stderr += (line + "\n").encode("utf8")
                else:
                    expected_stderr += b"\n"
        assert _simplify_warnings(_strip(actual_stderr)) == _simplify_warnings(
            _strip(expected_stderr)
        ), "Different stderr: {}".format(cmds)


def test_expected_number_of_examples_in_docs_data():
    expected = 67
    # See _get_examples_in_docs_data()
    if sys.version_info[:2] != (3, 8):
        expected -= 3
        # number of help tests
    assert len(examples_in_docs_data) + len(examples_in_docs_data_geo) == expected


def _simplify_warnings(lines):
    return "\n".join([_simplify_line(line) for line in lines.split("\n")])


def _simplify_line(line):
    if "DataErrorWarning: " in line:
        return line[line.find("DataErrorWarning: ") :]
    return line


# Older versions of Python have an extra whitespace at the end compared to newer ones
# https://bugs.python.org/issue16333
def _strip(output):
    # Don't worry about any extra blank lines at the end either
    outstr = str(output, "utf8").rstrip("\n")
    return "\n".join(line.rstrip(" ") for line in outstr.split("\n"))


# Useful for a coverage check - see developer docs for how to run the check
if __name__ == "__main__":
    test_examples_receipt()
    for root, filename in examples_in_docs_data:
        test_example_in_doc(root, filename)

1	import os	10✔
2	import shlex	10✔
3	import subprocess	10✔
4	import sys	10✔
5	import uuid	10✔
6	from os.path import join	10✔
7
8	import pytest	10✔
9
10	examples_in_docs_data = []	10✔
11	examples_in_docs_data_geo = []	10✔
12
13
14	def _get_examples_in_docs_data():	10✔
15	global examples_in_docs_data, examples_in_docs_data_geo
16	examples_in_docs_data = []	10✔
17	examples_in_docs_data_geo = []	10✔
18	for root, dirs, files in os.walk("examples"):	10✔
19	for filename in files:	10✔
20	if root.startswith("examples/help/") and sys.version_info[:2] != (3, 8):	10✔
21	# Different versions of python produce differently formatted help output.
22	# We only test in one version, Python 3.8.
23	# (Same as we lint code with, so dev's can have one virtual env)
24	continue	8✔
25	if "cmd.txt" in filename:	10✔
26	if root.startswith("examples/wkt"):	10✔
27	examples_in_docs_data_geo.append((root, filename))	10✔
28	else:
29	examples_in_docs_data.append((root, filename))	10✔
30
31
32	_get_examples_in_docs_data()	10✔
33
34
35	def test_examples_receipt():	10✔
36	with open("examples/receipt/source-map/expected.json", "rb") as fp:	×
37	expected = fp.read()	×
38	for expected_filename in [	×
39	"normalised/expected.json",
40	"combine-table-into-cafe/expected.json",
41	"combine-table-into-cafe-2/expected.json",
42	]:
43	with open("examples/receipt/" + expected_filename, "rb") as fp2:	×
44	assert (	×
45	fp2.read() == expected
46	), "Files differ: examples/receipt/source-map/expected.json, examples/receipt/{}".format(
47	expected_filename
48	)
49
50
51	@pytest.mark.parametrize("root, filename", examples_in_docs_data)	10✔
52	def test_example_in_doc(root, filename):	8✔
53	_test_example_in_doc_worker(root, filename)	×
54
55
56	@pytest.mark.parametrize("root, filename", examples_in_docs_data_geo)	10✔
57	@pytest.mark.geo	10✔
58	def test_example_in_doc_geo(root, filename):	8✔
59	_test_example_in_doc_worker(root, filename)	10✔
60
61
62	@pytest.mark.parametrize("root, filename", examples_in_docs_data)	10✔
63	def _test_example_in_doc_worker(root, filename):	8✔
64	if os.path.exists(join(root, "actual")) and os.path.isdir(join(root, "actual")):	10✔
65	os.rename(join(root, "actual"), join(root, "actual." + str(uuid.uuid4())))	×
66	os.mkdir(join(root, "actual"))	10✔
67	expected_return_code = 0	10✔
68	expected_stdout = b""	10✔
69	if os.path.exists(join(root, "expected_return_code.txt")):	10✔
70	with open(join(root, "expected_return_code.txt"), "rb") as fp:	×
71	expected_return_code = int(fp.read().strip())	×
72	with open(join(root, filename), "rb") as fp:	10✔
73	cmds = str(fp.read(), "utf8").strip().split("\n")	10✔
74	actual_stdout = b""	10✔
75	actual_stderr = b""	10✔
76	for cmd in cmds:	10✔
77	assert cmd.startswith("$ flatten-tool ") or cmd.startswith(	10✔
78	"$ cat "
79	), "Expected commands to start with '$ flatten-tool'. This doesn't: {}".format(
80	cmd
81	)
82	# Since we are defining all the commands ourselves, this is reasonably safe
83	cmd_parts = shlex.split(cmd[len("$ ") :])	10✔
84	# Include coverage output in the results
85	if cmd_parts[0] == "flatten-tool":	10✔
86	cmd_parts = [	10✔
87	"coverage",
88	"run",
89	"--source",
90	"flattentool",
91	"--parallel-mode",
92	] + cmd_parts
93	process = subprocess.Popen(	10✔
94	cmd_parts, stdout=subprocess.PIPE, stderr=subprocess.PIPE
95	)
96	(cmd_actual_stdout, cmd_actual_stderr) = process.communicate()	10✔
97	process.wait()	10✔
98	assert process.returncode == expected_return_code, cmd	10✔
99	actual_stdout += cmd_actual_stdout or b""	10✔
100	actual_stderr += cmd_actual_stderr or b""	10✔
101	if os.path.exists(join(root, "expected")) and os.path.isdir(join(root, "expected")):	10✔
102	# Create case
103	assert len(os.listdir(join(root, "expected"))) == len(	10✔
104	os.listdir(join(root, "actual"))
105	), "Different number of files. {}".format(cmds)
106	for expected_filename in os.listdir(join(root, "expected")):	10✔
107	assert os.path.exists(	10✔
108	join(root, "actual", expected_filename)
109	), "File {} was not generated {}".format(expected_filename, cmds)
110	with open(join(root, "expected", expected_filename), "rb") as fp_expected:	10✔
111	with open(join(root, "actual", expected_filename), "rb") as fp_actual:	10✔
112	assert _strip(fp_actual.read()) == _strip(	10✔
113	fp_expected.read()
114	), "File {} has unexpected content. {}".format(
115	expected_filename, cmds
116	)
117	expected_stdout = b""	10✔
118	# Flatten case
119	if os.path.exists(join(root, "expected.txt")):	10✔
120	with open(join(root, "expected.txt"), "rb") as fstdout:	×
121	expected_stdout = fstdout.read()	×
122	elif os.path.exists(join(root, "expected.json")):	10✔
123	with open(join(root, "expected.json"), "rb") as fstdout:	10✔
124	expected_stdout = fstdout.read()	10✔
125	elif os.path.exists(join(root, "expected.xml")):	10✔
126	with open(join(root, "expected.xml"), "rb") as fstdout:	×
127	expected_stdout = fstdout.read()	×
128	if "help" in root:	10✔
129	# Ignore whitespace differences for help messages
130	assert b" ".join(actual_stdout.split()) == b" ".join(expected_stdout.split())	×
131	else:
132	assert _strip(actual_stdout) == _strip(	10✔
133	expected_stdout
134	), "Different stdout: {}".format(cmds)
135	expected_stderr = b""	10✔
136	if os.path.exists(join(root, "expected_stderr_partial.txt")):	10✔
137	with open(join(root, "expected_stderr_partial.txt"), "rb") as fstderr:	×
138	data = fstderr.read()	×
139	assert data in actual_stderr	×
140	if os.path.exists(join(root, "expected_stderr.json")):	10✔
141	with open(join(root, "expected_stderr.json"), "rb") as fstderr:	×
142	data = fstderr.read()	×
143	expected_stderr_lines = str(data, "utf8").split("\n")	×
144	for line in expected_stderr_lines:	×
145	if line:	×
146	expected_stderr += (line + "\n").encode("utf8")	×
147	else:
148	expected_stderr += b"\n"	×
149	assert _simplify_warnings(_strip(actual_stderr)) == _simplify_warnings(	×
150	_strip(expected_stderr)
151	), "Different stderr: {}".format(cmds)
152
153
154	def test_expected_number_of_examples_in_docs_data():	10✔
155	expected = 67	×
156	# See _get_examples_in_docs_data()
157	if sys.version_info[:2] != (3, 8):	×
158	expected -= 3	×
159	# number of help tests
160	assert len(examples_in_docs_data) + len(examples_in_docs_data_geo) == expected	×
161
162
163	def _simplify_warnings(lines):	10✔
164	return "\n".join([_simplify_line(line) for line in lines.split("\n")])	×
165
166
167	def _simplify_line(line):	10✔
168	if "DataErrorWarning: " in line:	×
169	return line[line.find("DataErrorWarning: ") :]	×
170	return line	×
171
172
173	# Older versions of Python have an extra whitespace at the end compared to newer ones
174	# https://bugs.python.org/issue16333
175	def _strip(output):	10✔
176	# Don't worry about any extra blank lines at the end either
177	outstr = str(output, "utf8").rstrip("\n")	10✔
178	return "\n".join(line.rstrip(" ") for line in outstr.split("\n"))	10✔
179
180
181	# Useful for a coverage check - see developer docs for how to run the check
182	if __name__ == "__main__":	10✔
183	test_examples_receipt()	×
184	for root, filename in examples_in_docs_data:	×
185	test_example_in_doc(root, filename)	×

OpenDataServices / flatten-tool / 6507626273

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous