Changeset View
Changeset View
Standalone View
Standalone View
swh/graph/tests/test_cli.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 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 unittest | import unittest | ||||
| 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 swh.graph | |||||
| from swh.graph import cli | from swh.graph import cli | ||||
| def read_properties(properties_fname) -> Dict[str, str]: | def read_properties(properties_fname) -> Dict[str, str]: | ||||
| """read a Java .properties file""" | """read a Java .properties file""" | ||||
| properties = {} | properties = {} | ||||
| with open(properties_fname) as f: | with open(properties_fname) as f: | ||||
| for line in f: | for line in f: | ||||
| if line.startswith('#'): | if line.startswith('#'): | ||||
| continue | continue | ||||
| (key, value) = line.rstrip().split('=', maxsplit=1) | (key, value) = line.rstrip().split('=', maxsplit=1) | ||||
| properties[key] = value | properties[key] = value | ||||
| return properties | return properties | ||||
| class TestCompress(unittest.TestCase): | class TestCompress(unittest.TestCase): | ||||
| DATA_DIR = Path(swh.graph.__file__).parent.parent.parent \ | DATA_DIR = Path(__file__).parents[0] / 'dataset' | ||||
| / 'tests' / 'dataset' | |||||
| def setUp(self): | def setUp(self): | ||||
| self.runner = CliRunner() | self.runner = CliRunner() | ||||
| def test_pipeline(self): | def test_pipeline(self): | ||||
| """run full compression pipeline""" | """run full compression pipeline""" | ||||
| with TemporaryDirectory(suffix='.swh-graph-test') as tmpdir: | with TemporaryDirectory(suffix='.swh-graph-test') as tmpdir: | ||||
| result = self.runner.invoke( | result = self.runner.invoke( | ||||
| cli.compress, | cli.compress, | ||||
| ['--graph', self.DATA_DIR / 'example', '--outdir', tmpdir]) | ['--graph', self.DATA_DIR / 'example', '--outdir', tmpdir]) | ||||
| self.assertEqual(result.exit_code, 0) | self.assertEqual(result.exit_code, 0) | ||||
| properties = read_properties(Path(tmpdir) / 'example.properties') | properties = read_properties(Path(tmpdir) / 'example.properties') | ||||
| self.assertEqual(int(properties['nodes']), 21) | self.assertEqual(int(properties['nodes']), 21) | ||||
| self.assertEqual(int(properties['arcs']), 23) | self.assertEqual(int(properties['arcs']), 23) | ||||