Changeset View
Changeset View
Standalone View
Standalone View
swh/model/tests/test_cli.py
Show All 16 Lines | |||||
@pytest.mark.fs | @pytest.mark.fs | ||||
class TestIdentify(DataMixin, unittest.TestCase): | class TestIdentify(DataMixin, unittest.TestCase): | ||||
def setUp(self): | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
self.runner = CliRunner() | self.runner = CliRunner() | ||||
def assertPidOK(self, result, pid): | def assertSWHID(self, result, swhid): | ||||
self.assertEqual(result.exit_code, 0) | self.assertEqual(result.exit_code, 0) | ||||
self.assertEqual(result.output.split()[0], pid) | self.assertEqual(result.output.split()[0], swhid) | ||||
def test_no_args(self): | def test_no_args(self): | ||||
result = self.runner.invoke(cli.identify) | result = self.runner.invoke(cli.identify) | ||||
self.assertNotEqual(result.exit_code, 0) | self.assertNotEqual(result.exit_code, 0) | ||||
def test_content_id(self): | def test_content_id(self): | ||||
"""identify file content""" | """identify file content""" | ||||
self.make_contents(self.tmpdir_name) | self.make_contents(self.tmpdir_name) | ||||
for filename, content in self.contents.items(): | for filename, content in self.contents.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
result = self.runner.invoke(cli.identify, ["--type", "content", path]) | result = self.runner.invoke(cli.identify, ["--type", "content", path]) | ||||
self.assertPidOK(result, "swh:1:cnt:" + hash_to_hex(content["sha1_git"])) | self.assertSWHID(result, "swh:1:cnt:" + hash_to_hex(content["sha1_git"])) | ||||
def test_content_id_from_stdin(self): | def test_content_id_from_stdin(self): | ||||
"""identify file content""" | """identify file content""" | ||||
self.make_contents(self.tmpdir_name) | self.make_contents(self.tmpdir_name) | ||||
for _, content in self.contents.items(): | for _, content in self.contents.items(): | ||||
result = self.runner.invoke(cli.identify, ["-"], input=content["data"]) | result = self.runner.invoke(cli.identify, ["-"], input=content["data"]) | ||||
self.assertPidOK(result, "swh:1:cnt:" + hash_to_hex(content["sha1_git"])) | self.assertSWHID(result, "swh:1:cnt:" + hash_to_hex(content["sha1_git"])) | ||||
def test_directory_id(self): | def test_directory_id(self): | ||||
"""identify an entire directory""" | """identify an entire directory""" | ||||
self.make_from_tarball(self.tmpdir_name) | self.make_from_tarball(self.tmpdir_name) | ||||
path = os.path.join(self.tmpdir_name, b"sample-folder") | path = os.path.join(self.tmpdir_name, b"sample-folder") | ||||
result = self.runner.invoke(cli.identify, ["--type", "directory", path]) | result = self.runner.invoke(cli.identify, ["--type", "directory", path]) | ||||
self.assertPidOK(result, "swh:1:dir:e8b0f1466af8608c8a3fb9879db172b887e80759") | self.assertSWHID(result, "swh:1:dir:e8b0f1466af8608c8a3fb9879db172b887e80759") | ||||
def test_snapshot_id(self): | def test_snapshot_id(self): | ||||
"""identify a snapshot""" | """identify a snapshot""" | ||||
tarball = os.path.join( | tarball = os.path.join( | ||||
os.path.dirname(__file__), "data", "repos", "sample-repo.tgz" | os.path.dirname(__file__), "data", "repos", "sample-repo.tgz" | ||||
) | ) | ||||
with tempfile.TemporaryDirectory(prefix="swh.model.cli") as d: | with tempfile.TemporaryDirectory(prefix="swh.model.cli") as d: | ||||
with tarfile.open(tarball, "r:gz") as t: | with tarfile.open(tarball, "r:gz") as t: | ||||
t.extractall(d) | t.extractall(d) | ||||
repo_dir = os.path.join(d, "sample-repo") | repo_dir = os.path.join(d, "sample-repo") | ||||
result = self.runner.invoke( | result = self.runner.invoke( | ||||
cli.identify, ["--type", "snapshot", repo_dir] | cli.identify, ["--type", "snapshot", repo_dir] | ||||
) | ) | ||||
self.assertPidOK( | self.assertSWHID( | ||||
result, "swh:1:snp:abc888898124270905a0ef3c67e872ce08e7e0c1" | result, "swh:1:snp:abc888898124270905a0ef3c67e872ce08e7e0c1" | ||||
) | ) | ||||
def test_origin_id(self): | def test_origin_id(self): | ||||
"""identify an origin URL""" | """identify an origin URL""" | ||||
url = "https://github.com/torvalds/linux" | url = "https://github.com/torvalds/linux" | ||||
result = self.runner.invoke(cli.identify, ["--type", "origin", url]) | result = self.runner.invoke(cli.identify, ["--type", "origin", url]) | ||||
self.assertPidOK(result, "swh:1:ori:b63a575fe3faab7692c9f38fb09d4bb45651bb0f") | self.assertSWHID(result, "swh:1:ori:b63a575fe3faab7692c9f38fb09d4bb45651bb0f") | ||||
def test_symlink(self): | def test_symlink(self): | ||||
"""identify symlink --- both itself and target""" | """identify symlink --- both itself and target""" | ||||
regular = os.path.join(self.tmpdir_name, b"foo.txt") | regular = os.path.join(self.tmpdir_name, b"foo.txt") | ||||
link = os.path.join(self.tmpdir_name, b"bar.txt") | link = os.path.join(self.tmpdir_name, b"bar.txt") | ||||
open(regular, "w").write("foo\n") | open(regular, "w").write("foo\n") | ||||
os.symlink(os.path.basename(regular), link) | os.symlink(os.path.basename(regular), link) | ||||
result = self.runner.invoke(cli.identify, [link]) | result = self.runner.invoke(cli.identify, [link]) | ||||
self.assertPidOK(result, "swh:1:cnt:257cc5642cb1a054f08cc83f2d943e56fd3ebe99") | self.assertSWHID(result, "swh:1:cnt:257cc5642cb1a054f08cc83f2d943e56fd3ebe99") | ||||
result = self.runner.invoke(cli.identify, ["--no-dereference", link]) | result = self.runner.invoke(cli.identify, ["--no-dereference", link]) | ||||
self.assertPidOK(result, "swh:1:cnt:996f1789ff67c0e3f69ef5933a55d54c5d0e9954") | self.assertSWHID(result, "swh:1:cnt:996f1789ff67c0e3f69ef5933a55d54c5d0e9954") | ||||
def test_show_filename(self): | def test_show_filename(self): | ||||
"""filename is shown by default""" | """filename is shown by default""" | ||||
self.make_contents(self.tmpdir_name) | self.make_contents(self.tmpdir_name) | ||||
for filename, content in self.contents.items(): | for filename, content in self.contents.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
result = self.runner.invoke(cli.identify, ["--type", "content", path]) | result = self.runner.invoke(cli.identify, ["--type", "content", path]) | ||||
self.assertEqual(result.exit_code, 0) | self.assertEqual(result.exit_code, 0) | ||||
self.assertEqual( | self.assertEqual( | ||||
result.output.rstrip(), | result.output.rstrip(), | ||||
"swh:1:cnt:%s\t%s" % (hash_to_hex(content["sha1_git"]), path.decode()), | "swh:1:cnt:%s\t%s" % (hash_to_hex(content["sha1_git"]), path.decode()), | ||||
) | ) | ||||
def test_hide_filename(self): | def test_hide_filename(self): | ||||
"""filename is hidden upon request""" | """filename is hidden upon request""" | ||||
self.make_contents(self.tmpdir_name) | self.make_contents(self.tmpdir_name) | ||||
for filename, content in self.contents.items(): | for filename, content in self.contents.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
result = self.runner.invoke( | result = self.runner.invoke( | ||||
cli.identify, ["--type", "content", "--no-filename", path] | cli.identify, ["--type", "content", "--no-filename", path] | ||||
) | ) | ||||
self.assertPidOK(result, "swh:1:cnt:" + hash_to_hex(content["sha1_git"])) | self.assertSWHID(result, "swh:1:cnt:" + hash_to_hex(content["sha1_git"])) | ||||
def test_auto_content(self): | def test_auto_content(self): | ||||
"""automatic object type detection: content""" | """automatic object type detection: content""" | ||||
with tempfile.NamedTemporaryFile(prefix="swh.model.cli") as f: | with tempfile.NamedTemporaryFile(prefix="swh.model.cli") as f: | ||||
result = self.runner.invoke(cli.identify, [f.name]) | result = self.runner.invoke(cli.identify, [f.name]) | ||||
self.assertEqual(result.exit_code, 0) | self.assertEqual(result.exit_code, 0) | ||||
self.assertRegex(result.output, r"^swh:\d+:cnt:") | self.assertRegex(result.output, r"^swh:\d+:cnt:") | ||||
Show All 29 Lines |