Differential D8919 Diff 32176 swh/graph/tests/test_cli.py

Changeset View

Standalone View

swh/graph/tests/test_cli.py

# Copyright (C) 2019 The Software Heritage developers		# Copyright (C) 2019-2022 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

		import os
from pathlib import Path		from pathlib import Path
from tempfile import TemporaryDirectory		from tempfile import TemporaryDirectory
from typing import Dict		from typing import Dict

from click.testing import CliRunner		from click.testing import CliRunner
		import pytest
import yaml		import yaml

from swh.graph.cli import graph_cli_group		from swh.graph.cli import graph_cli_group

DATA_DIR = Path(__file__).parents[0] / "dataset"		DATA_DIR = Path(__file__).parents[0] / "dataset"


def read_properties(properties_fname) -> Dict[str, str]:		def read_properties(properties_fname) -> Dict[str, str]:
Show All 32 Lines	with TemporaryDirectory(suffix=".swh-graph-test") as tmpdir:
"example",		"example",
],		],
)		)
assert result.exit_code == 0, result		assert result.exit_code == 0, result
properties = read_properties(Path(tmpdir) / "example.properties")		properties = read_properties(Path(tmpdir) / "example.properties")

assert int(properties["nodes"]) == 24		assert int(properties["nodes"]) == 24
assert int(properties["arcs"]) == 28		assert int(properties["arcs"]) == 28


		@pytest.mark.parametrize("exit_code", [0, 1])
		def test_luigi(mocker, tmpdir, exit_code):
		"""calls Luigi with the given configuration"""
		# bare bone configuration, to allow testing the compression pipeline
		# with minimum RAM requirements on trivial graphs
		runner = CliRunner()

		subprocess_run = mocker.patch("subprocess.run")
		subprocess_run.return_value.returncode = exit_code

		with TemporaryDirectory(suffix=".swh-graph-test") as tmpdir:
		result = runner.invoke(
		graph_cli_group,
		[
		"luigi",
		"--base-directory",
		f"{tmpdir}/base_dir",
		"--dataset-name",
		"2022-12-07",
		"--",
		"foo",
		"bar",
		"--baz",
		"qux",
		],
		catch_exceptions=False,
		)
		assert result.exit_code == exit_code, result

		luigi_config_path = subprocess_run.mock_calls[0][2]["env"]["LUIGI_CONFIG_PATH"]
		subprocess_run.assert_called_once_with(
		[
		"luigi",
		"--module",
		"swh.dataset.luigi",
		"--module",
		"swh.graph.luigi",
		"foo",
		"bar",
		"--baz",
		"qux",
		],
		env={"LUIGI_CONFIG_PATH": luigi_config_path, **os.environ},
		)