GSoC Module A : Week 2 (stacked on top of #920)

application/tests/harvester_test/git_repository_client_integration_test.py

@@ -0,0 +1,206 @@
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+import threading
+
+from application.utils.harvester.git_repository_client import (
+    GitRepositoryClient,
+)
+
+
+class IntegrationGitRepositoryClient(GitRepositoryClient):
+    def __init__(self, *args, repository_url: str, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._repository_url = repository_url
+
+    @property
+    def repository_url(self) -> str:
+        return self._repository_url
+
+
+def git(*args, cwd=None):
+    subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
+
+def git_output(*args, cwd=None):
+    return subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout.strip()
+
+
+class GitRepositoryClientIntegrationTests(unittest.TestCase):
+    def setUp(self):
+        self.tempdir = tempfile.TemporaryDirectory()
+        self.root = Path(self.tempdir.name)
+
+        self.remote = self.root / "remote.git"
+        self.work = self.root / "work"
+        self.cache = self.root / "cache"
+
+        git("init", "--bare", self.remote)
+
+        git("clone", self.remote, self.work)
+
+        git("config", "user.name", "Test User", cwd=self.work)
+        git("config", "user.email", "test@example.com", cwd=self.work)
+        git("checkout", "-b", "main", cwd=self.work)
+
+        (self.work / "test.txt").write_text("v1")
+
+        git("add", ".", cwd=self.work)
+        git("commit", "-m", "initial", cwd=self.work)
+        git("push", "origin", "main", cwd=self.work)
+
+    def tearDown(self):
+        self.tempdir.cleanup()
+
+    def create_client(self):
+        return IntegrationGitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+            local_path=self.cache,
+            repository_url=str(self.remote),
+        )
+
+    def test_fetch_updates_worktree_and_commit(self):
+        client = self.create_client()
+
+        client.clone()
+
+        sha1 = client.get_current_commit_sha()
+
+        self.assertEqual(
+            (client.get_local_path() / "test.txt").read_text(),
+            "v1",
+        )
+
+        (self.work / "test.txt").write_text("v2")
+
+        git("add", ".", cwd=self.work)
+        git("commit", "-m", "update", cwd=self.work)
+        git("push", "origin", "main", cwd=self.work)
+
+        expected_sha = git_output(
+            "rev-parse",
+            "HEAD",
+            cwd=self.work,
+        )
+
+        client.fetch()
+
+        self.assertEqual(
+            client.get_current_commit_sha(),
+            expected_sha,
+        )
+
+        self.assertNotEqual(
+            sha1,
+            expected_sha,
+        )
+
+        self.assertEqual(
+            (client.get_local_path() / "test.txt").read_text(),
+            "v2",
+        )
+
+    def test_verify_repository_integrity_rejects_fake_git_directory(self):
+        fake = self.root / "fake"
+
+        fake.mkdir()
+        (fake / ".git").mkdir()
+
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+            local_path=fake,
+        )
+
+        self.assertFalse(
+            client.verify_repository_integrity(),
+        )
+
+    def test_verify_repository_integrity_rejects_wrong_origin(self):
+        other_remote = self.root / "other.git"
+
+        git("init", "--bare", other_remote)
+
+        client = self.create_client()
+
+        client.clone()
+
+        git(
+            "remote",
+            "set-url",
+            "origin",
+            other_remote,
+            cwd=client.get_local_path(),
+        )
+
+        self.assertFalse(
+            client.verify_repository_integrity(),
+        )
+
+    def test_verify_repository_integrity_rejects_missing_branch(self):
+        client = IntegrationGitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+            branch="dev",
+            local_path=self.cache,
+            repository_url=str(self.remote),
+        )
+
+        git("clone", self.remote, self.cache)
+
+        self.assertFalse(
+            client.verify_repository_integrity(),
+        )
+
+    def test_sync_serializes_clone_operations(self):
+        client1 = self.create_client()
+        client2 = self.create_client()
+
+        exceptions = []
+
+        def run_sync(client):
+            try:
+                client.sync()
+            except Exception as exc:
+                exceptions.append(exc)
+
+        t1 = threading.Thread(target=run_sync, args=(client1,))
+        t2 = threading.Thread(target=run_sync, args=(client2,))
+
+        t1.start()
+        t2.start()
+
+        t1.join()
+        t2.join()
+
+        self.assertFalse(exceptions, f"Unexpected exceptions: {exceptions}")
+
+        self.assertTrue(client1.verify_repository_integrity())
+        self.assertTrue(client2.verify_repository_integrity())
+
+        self.assertTrue((self.cache / ".git").exists())
+
+        self.assertEqual(
+            client1.get_current_commit_sha(),
+            client2.get_current_commit_sha(),
+        )
+
+        self.assertEqual((self.cache / "test.txt").read_text(), "v1")
+
+
+if __name__ == "__main__":
+    unittest.main()

application/tests/harvester_test/git_repository_client_test.py

@@ -0,0 +1,156 @@
+import unittest
+from unittest.mock import patch
+
+import tempfile
+from pathlib import Path
+
+from application.utils.harvester.git_repository_client import (
+    GitRepositoryClient,
+)
+
+
+class GitRepositoryClientTests(unittest.TestCase):
+    def test_repository_url_generation(self):
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        self.assertEqual(
+            client.repository_url,
+            "https://github.com/OWASP/ASVS.git",
+        )
+
+    def test_local_repository_path(self):
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        self.assertEqual(
+            str(client.get_local_path()),
+            ".harvester_cache/owasp/asvs/main",
+        )
+
+    def test_repository_exists_locally_false(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            client = GitRepositoryClient(
+                owner="OWASP",
+                repository="ASVS",
+                local_path=Path(tmpdir) / "repo",
+            )
+
+            self.assertFalse(client.exists_locally())
+
+    def test_verify_repository_integrity_false(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            client = GitRepositoryClient(
+                owner="OWASP",
+                repository="ASVS",
+                local_path=Path(tmpdir) / "repo",
+            )
+
+            self.assertFalse(client.verify_repository_integrity())
+
+    def test_sync_clones_when_repository_missing(self):
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        with (
+            patch.object(
+                client,
+                "verify_repository_integrity",
+                return_value=False,
+            ),
+            patch.object(client, "clone") as mock_clone,
+        ):
+            client.sync()
+
+        mock_clone.assert_called_once()
+
+    def test_sync_fetches_when_repository_exists(self):
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        with (
+            patch.object(
+                client,
+                "verify_repository_integrity",
+                return_value=True,
+            ),
+            patch.object(client, "fetch") as mock_fetch,
+        ):
+            client.sync()
+
+        mock_fetch.assert_called_once()
+
+    @patch("application.utils.harvester.git_repository_client.subprocess.run")
+    def test_fetch_runs_git_command(self, mock_run):
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        client.fetch()
+
+        self.assertEqual(mock_run.call_count, 2)
+
+    @patch("application.utils.harvester.git_repository_client.subprocess.run")
+    def test_checkout_runs_git_command(self, mock_run):
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        client.checkout("main")
+
+        mock_run.assert_called_once_with(
+            [
+                "git",
+                "-C",
+                str(client.get_local_path()),
+                "checkout",
+                "main",
+            ],
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=300,
+        )
+
+    @patch("application.utils.harvester.git_repository_client.subprocess.run")
+    def test_get_current_commit_sha_runs_git_command(self, mock_run):
+        mock_run.return_value.stdout = "abc123\n"
+
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        sha = client.get_current_commit_sha()
+
+        self.assertEqual(sha, "abc123")
+        mock_run.assert_called_once()
+
+    @patch("application.utils.harvester.git_repository_client.subprocess.run")
+    def test_clone_runs_git_command(self, mock_run):
+        client = GitRepositoryClient(
+            owner="OWASP",
+            repository="ASVS",
+        )
+
+        with (
+            patch.object(client, "verify_repository_integrity", return_value=False),
+            patch.object(client, "is_valid_repository", return_value=True),
+        ):
+            client.clone()
+
+        mock_run.assert_called()
+
+
+if __name__ == "__main__":
+    unittest.main()