swh/model/tests/test_cli.py

@@ -4,26 +4,28 @@
 # See top-level LICENSE file for more information
 
 import os
+import sys
 import tarfile
 import tempfile
 import unittest
+import unittest.mock
 
 from click.testing import CliRunner
 import pytest
 
 from swh.model import cli
 from swh.model.hashutil import hash_to_hex
+from swh.model.tests.swh_model_data import SAMPLE_FOLDER_SWHIDS
 from swh.model.tests.test_from_disk import DataMixin
 
 
-@pytest.mark.fs
 class TestIdentify(DataMixin, unittest.TestCase):
     def setUp(self):
         super().setUp()
         self.runner = CliRunner()
 
     def assertSWHID(self, result, swhid):
-        self.assertEqual(result.exit_code, 0)
+        self.assertEqual(result.exit_code, 0, result.output)
         self.assertEqual(result.output.split()[0], swhid)
 
     def test_no_args(self):
@@ -52,6 +54,7 @@ class TestIdentify(DataMixin, unittest.TestCase):
         result = self.runner.invoke(cli.identify, ["--type", "directory", path])
         self.assertSWHID(result, "swh:1:dir:e8b0f1466af8608c8a3fb9879db172b887e80759")
 
+    @pytest.mark.requires_optional_deps
     def test_snapshot_id(self):
         """identify a snapshot"""
         tarball = os.path.join(
@@ -68,6 +71,20 @@ class TestIdentify(DataMixin, unittest.TestCase):
                     result, "swh:1:snp:abc888898124270905a0ef3c67e872ce08e7e0c1"
                 )
 
+    def test_snapshot_without_dulwich(self):
+        """checks swh-identify returns a 'nice' message instead of a traceback
+        when dulwich is not installed"""
+        with unittest.mock.patch.dict(sys.modules, {"dulwich": None}):
+            with tempfile.TemporaryDirectory(prefix="swh.model.cli") as d:
+                result = self.runner.invoke(
+                    cli.identify,
+                    ["--type", "snapshot", d],
+                    catch_exceptions=False,
+                )
+
+        assert result.exit_code == 1
+        assert "'swh.model[cli]'" in result.output
+
     def test_origin_id(self):
         """identify an origin URL"""
         url = "https://github.com/torvalds/linux"
@@ -78,7 +95,8 @@ class TestIdentify(DataMixin, unittest.TestCase):
         """identify symlink --- both itself and target"""
         regular = os.path.join(self.tmpdir_name, b"foo.txt")
         link = os.path.join(self.tmpdir_name, b"bar.txt")
-        open(regular, "w").write("foo\n")
+        with open(regular, "w") as f:
+            f.write("foo\n")
         os.symlink(os.path.basename(regular), link)
 
         result = self.runner.invoke(cli.identify, [link])
@@ -127,7 +145,7 @@ class TestIdentify(DataMixin, unittest.TestCase):
     def test_auto_origin(self):
         """automatic object type detection: origin"""
         result = self.runner.invoke(cli.identify, ["https://github.com/torvalds/linux"])
-        self.assertEqual(result.exit_code, 0)
+        self.assertEqual(result.exit_code, 0, result.output)
         self.assertRegex(result.output, r"^swh:\d+:ori:")
 
     def test_verify_content(self):
@@ -139,10 +157,57 @@ class TestIdentify(DataMixin, unittest.TestCase):
             # match
             path = os.path.join(self.tmpdir_name, filename)
             result = self.runner.invoke(cli.identify, ["--verify", expected_id, path])
-            self.assertEqual(result.exit_code, 0)
+            self.assertEqual(result.exit_code, 0, result.output)
 
             # mismatch
             with open(path, "a") as f:
                 f.write("trailing garbage to make verification fail")
             result = self.runner.invoke(cli.identify, ["--verify", expected_id, path])
             self.assertEqual(result.exit_code, 1)
+
+    def test_exclude(self):
+        """exclude patterns"""
+        self.make_from_tarball(self.tmpdir_name)
+        path = os.path.join(self.tmpdir_name, b"sample-folder")
+
+        excluded_dir = os.path.join(path, b"excluded_dir\x96")
+        os.mkdir(excluded_dir)
+        with open(os.path.join(excluded_dir, b"some_file"), "w") as f:
+            f.write("content")
+
+        result = self.runner.invoke(
+            cli.identify, ["--type", "directory", "--exclude", "excluded_*", path]
+        )
+
+        self.assertSWHID(result, "swh:1:dir:e8b0f1466af8608c8a3fb9879db172b887e80759")
+
+    def test_recursive_directory(self):
+        self.make_from_tarball(self.tmpdir_name)
+        path = os.path.join(self.tmpdir_name, b"sample-folder")
+        result = self.runner.invoke(cli.identify, ["--recursive", path])
+        self.assertEqual(result.exit_code, 0, result.output)
+
+        result = result.output.split()
+        result_swhids = []
+        # get all SWHID from the result
+        for i in range(0, len(result)):
+            if i % 2 == 0:
+                result_swhids.append(result[i])
+
+        assert len(result_swhids) == len(SAMPLE_FOLDER_SWHIDS)
+        for swhid in SAMPLE_FOLDER_SWHIDS:
+            assert swhid in result_swhids
+
+    def test_recursive_directory_no_filename(self):
+        self.make_from_tarball(self.tmpdir_name)
+        path = os.path.join(self.tmpdir_name, b"sample-folder")
+        result = self.runner.invoke(
+            cli.identify, ["--recursive", "--no-filename", path]
+        )
+        self.assertEqual(result.exit_code, 0, result.output)
+
+        result_swhids = result.output.split()
+
+        assert len(result_swhids) == len(SAMPLE_FOLDER_SWHIDS)
+        for swhid in SAMPLE_FOLDER_SWHIDS:
+            assert swhid in result_swhids

swh/model/tests/test_discovery.py

+# Copyright (C) 2023 The Software Heritage developers
+# See the AUTHORS file at the top-level directory of this distribution
+# License: GNU General Public License version 3, or any later version
+# See top-level LICENSE file for more information
+
+from dataclasses import dataclass
+from typing import Iterable, List
+
+from swh.model import discovery, model
+from swh.model.hashutil import hash_to_bytes
+from swh.model.model import Sha1Git
+from swh.model.tests.test_identifiers import directory_example
+
+pytest_plugins = ["aiohttp.pytest_plugin"]
+
+UNKNOWN_HASH = hash_to_bytes("17140cb6109f1e3296dc52e2b2cd29bcb40e86be")
+KNOWN_CONTENT_HASH = hash_to_bytes("e8e4106de42e2d5d5efab6a9422b9a8677c993c8")
+KNOWN_DIRECTORY_HASH = hash_to_bytes("d7ed3d2c31d608823be58b1cbe57605310615231")
+KNOWN_DIRECTORY_HASH_2 = hash_to_bytes("c76724e9a0be4b60f4bf0cb48b261df8eda94b1d")
+
+
+@dataclass
+class FakeArchive:
+    contents: List[model.Content]
+    skipped_contents: List[model.SkippedContent]
+    directories: List[model.Directory]
+
+    def content_missing(self, contents: List[Sha1Git]) -> Iterable[Sha1Git]:
+        return []
+
+    def skipped_content_missing(
+        self, skipped_contents: List[Sha1Git]
+    ) -> Iterable[Sha1Git]:
+        """List skipped content missing from the archive by sha1"""
+        return []
+
+    def directory_missing(self, directories: List[Sha1Git]) -> Iterable[Sha1Git]:
+        """List directories missing from the archive by sha1"""
+        return []
+
+
+def test_filter_known_objects(monkeypatch):
+    # Test with smaller sample sizes to actually trigger the random sampling
+    monkeypatch.setattr(discovery, "SAMPLE_SIZE", 1)
+
+    base_directory = model.Directory.from_dict(directory_example)
+
+    # Hardcoding another hash is enough since it's all that's being checked
+    directory_data = directory_example.copy()
+    directory_data["id"] = KNOWN_DIRECTORY_HASH_2
+    other_directory = model.Directory.from_dict(directory_data)
+    archive = FakeArchive(
+        contents=[model.Content.from_data(b"blabla")],
+        skipped_contents=[model.SkippedContent.from_data(b"blabla2", reason="reason")],
+        directories=[
+            base_directory,
+            other_directory,
+        ],
+    )
+    assert archive.contents[0].sha1_git == KNOWN_CONTENT_HASH
+    assert archive.directories[0].id == KNOWN_DIRECTORY_HASH
+    assert archive.directories[1].id == KNOWN_DIRECTORY_HASH_2
+    (contents, skipped_contents, directories) = discovery.filter_known_objects(archive)
+    assert len(contents) == 0
+    assert len(skipped_contents) == 0
+    assert len(directories) == 0