Migrate subsection of vcstools' API for compatibility with bloom

.github/workflows/ci.yml

@@ -41,12 +41,22 @@ jobs:
           brew install subversion mercurial
         if: matrix.os == 'macos-latest'
 
+      - name: Install dependencies (macOS)
+        run: |
+          brew install subversion mercurial breezy
+        if: matrix.os == 'macos-latest'
+
       - name: Install dependencies (Ubuntu)
         run: |
           sudo apt-get update
-          sudo apt-get install -y --no-install-recommends subversion mercurial
+          sudo apt-get install -y --no-install-recommends subversion mercurial brz
         if: startsWith(matrix.os, 'ubuntu')
 
+      - name: Install breezy pip (Ubuntu 24.04)
+        run: |
+          python -m pip install --upgrade breezy  # Adds breezy to the PYTHONPATH
+        if: startsWith(matrix.os, 'ubuntu-24.04')
+
       - name: Test with pytest
         run: |
           pip install --upgrade .[test]

test/__init__.py

@@ -140,7 +140,7 @@ def tearDownClass(cls):
 
 
 class StagedReposFile2(unittest.TestCase):
-    """Fixture for testing subversion and mercurial clients."""
+    """Fixture for testing subversion, mercurial, and bazaar clients."""
 
     _svn = which('svn')
     _svnadmin = which('svnadmin')
@@ -150,6 +150,11 @@ class StagedReposFile2(unittest.TestCase):
         'HGUSER': 'vcs2l',
         'EMAIL': 'vcs2l@example.com',
     }
+    _bzr = which('bzr')
+    _bzr_env = {
+        **os.environ,
+        'BZR_EMAIL': 'vcs2l <vcs2l@example.com>',
+    }
     _commit_date = '2005-01-01T00:00:00-06:00'
 
     temp_dir = None
@@ -163,6 +168,8 @@ def setUpClass(cls):
             raise unittest.SkipTest('`svnadmin` was not found')
         if not cls._hg:
             raise unittest.SkipTest('`hg` was not found')
+        if not cls._bzr:
+            raise unittest.SkipTest('`bzr` was not found')
         try:
             # check if the svn executable is usable (on macOS)
             # and not only exists to state that the program is not installed
@@ -221,6 +228,29 @@ def setUpClass(cls):
                 env=cls._hg_env,
             )
 
+        # Create the staged bazaar repository
+        bzrrepo_path = os.path.join(cls.temp_dir.name, 'bzrrepo')
+        os.mkdir(bzrrepo_path)
+        shutil.copy(
+            os.path.join(os.path.dirname(os.path.dirname(__file__)), 'LICENSE'),
+            bzrrepo_path,
+        )
+        for command in (
+            ('init', '.'),
+            ('add', 'LICENSE'),
+            ('commit', '-m', 'Initial commit'),
+        ):
+            subprocess.check_call(
+                [
+                    cls._bzr,
+                    *command,
+                ],
+                cwd=bzrrepo_path,
+                env=cls._bzr_env,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+
         # Populate the staged.repos file
         repos = {
             'hg/branch': {

test/test_bzr.py

@@ -0,0 +1,112 @@
+"""Tests for BzrClient."""
+
+import os
+import tarfile
+import unittest
+from tempfile import TemporaryDirectory
+
+from vcs2l.clients.bzr import BzrClient
+
+from . import StagedReposFile2, to_file_url
+
+
+class TestBzrCheckout(StagedReposFile2):
+    """Test BzrClient.checkout using the staged bzr repository."""
+
+    def test_default_branch(self):
+        """Checkout without a version branches the entire repository."""
+        with TemporaryDirectory(suffix='.bzr_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            client = BzrClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'bzrrepo'))
+
+            result = client.checkout(url)
+            self.assertTrue(result)
+            self.assertTrue(os.path.isdir(os.path.join(dest, '.bzr')))
+            self.assertTrue(os.path.isfile(os.path.join(dest, 'LICENSE')))
+
+    def test_specific_hash(self):
+        """Checkout at revision 1."""
+        with TemporaryDirectory(suffix='.bzr_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            client = BzrClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'bzrrepo'))
+
+            result = client.checkout(url, version='1')
+            self.assertTrue(result)
+            self.assertTrue(os.path.isdir(os.path.join(dest, '.bzr')))
+
+    def test_nonempty_dir(self):
+        """Checkout into a non-empty directory should raise RuntimeError."""
+        with TemporaryDirectory(suffix='.bzr_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            os.makedirs(dest)
+            with open(os.path.join(dest, 'blocker.txt'), 'w', encoding='utf-8') as f:
+                f.write('occupied')
+
+            client = BzrClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'bzrrepo'))
+
+            with self.assertRaises(RuntimeError):
+                client.checkout(url)
+
+
+class TestBzrExportRepository(StagedReposFile2):
+    """Test BzrClient.export_repository using the staged bzr repository."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls._branch_dir = TemporaryDirectory(suffix='.bzr_export')
+        cls._branch_path = os.path.join(cls._branch_dir.name, 'repo')
+        client = BzrClient(cls._branch_path)
+        url = to_file_url(os.path.join(cls.temp_dir.name, 'bzrrepo'))
+        assert client.checkout(url), 'Failed to branch staged bzr repo'
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._branch_dir.cleanup()
+        super().tearDownClass()
+
+    def test_creates_tarball(self):
+        """export_repository should create a .tar.gz archive."""
+        with TemporaryDirectory(suffix='.bzr_export_out') as tmp:
+            basepath = os.path.join(tmp, 'export')
+            client = BzrClient(self._branch_path)
+
+            result = client.export_repository('1', basepath)
+            self.assertTrue(result)
+            self.assertTrue(os.path.isfile(basepath + '.tar.gz'))
+
+            with tarfile.open(basepath + '.tar.gz', 'r:gz') as tar:
+                names = tar.getnames()
+                self.assertTrue(len(names) > 0)
+
+    def test_contains_license(self):
+        """Exported archive should contain LICENSE."""
+        with TemporaryDirectory(suffix='.bzr_export_out') as tmp:
+            basepath = os.path.join(tmp, 'export_license')
+            client = BzrClient(self._branch_path)
+
+            result = client.export_repository('1', basepath)
+            self.assertTrue(result)
+
+            with tarfile.open(basepath + '.tar.gz', 'r:gz') as tar:
+                names = tar.getnames()
+                self.assertTrue(
+                    any('LICENSE' in n for n in names),
+                    'LICENSE not found in archive: %s' % names,
+                )
+
+    def test_invalid_version(self):
+        """export_repository with a bad revision should return False."""
+        with TemporaryDirectory(suffix='.bzr_export_out') as tmp:
+            basepath = os.path.join(tmp, 'export_bad')
+            client = BzrClient(self._branch_path)
+
+            result = client.export_repository('9999', basepath)
+            self.assertFalse(result)
+
+
+if __name__ == '__main__':
+    unittest.main()

test/test_git.py

@@ -0,0 +1,143 @@
+"""Tests for Git Client."""
+
+import os
+import tarfile
+import unittest
+from tempfile import TemporaryDirectory
+
+from vcs2l.clients.git import GitClient
+
+from . import StagedReposFile, to_file_url
+
+
+class TestGitCheckout(StagedReposFile):
+    """Test GitClient.checkout using the staged git repository."""
+
+    def test_default_branch(self):
+        """Checkout without a version gets the default branch."""
+        with TemporaryDirectory(suffix='.git_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            client = GitClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'gitrepo'))
+
+            result = client.checkout(url)
+            self.assertTrue(result)
+            self.assertTrue(os.path.isdir(os.path.join(dest, '.git')))
+
+    def test_specific_branch(self):
+        """Checkout the main branch by name."""
+        with TemporaryDirectory(suffix='.git_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            client = GitClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'gitrepo'))
+
+            result = client.checkout(url, version='main')
+            self.assertTrue(result)
+            self.assertTrue(os.path.isdir(os.path.join(dest, '.git')))
+
+    def test_specific_hash(self):
+        """Checkout a specific commit hash."""
+        with TemporaryDirectory(suffix='.git_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            client = GitClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'gitrepo'))
+
+            result = client.checkout(url, version='0.1.26')
+            self.assertTrue(result)
+
+    def test_nonempty_dir(self):
+        """Checkout into a non-empty directory should return False."""
+        with TemporaryDirectory(suffix='.git_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            os.makedirs(dest)
+            # Place a file so the directory is non-empty
+            with open(os.path.join(dest, 'blocker.txt'), 'w', encoding='utf-8') as f:
+                f.write('occupied')
+
+            client = GitClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'gitrepo'))
+
+            result = client.checkout(url)
+            self.assertFalse(result)
+
+    def test_invalid_version(self):
+        """Checkout with a non-existent version should return False."""
+        with TemporaryDirectory(suffix='.git_checkout') as tmp:
+            dest = os.path.join(tmp, 'repo')
+            client = GitClient(dest)
+            url = to_file_url(os.path.join(self.temp_dir.name, 'gitrepo'))
+
+            result = client.checkout(url, version='nonexistent-branch-xyz')
+            self.assertFalse(result)
+
+
+class TestGitExportRepository(StagedReposFile):
+    """Test GitClient.export_repository using the staged git repository."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        # Clone the staged repo so we have a working git directory
+        cls._clone_dir = TemporaryDirectory(suffix='.git_export')
+        cls._clone_path = os.path.join(cls._clone_dir.name, 'repo')
+        client = GitClient(cls._clone_path)
+        url = to_file_url(os.path.join(cls.temp_dir.name, 'gitrepo'))
+        assert client.checkout(url, version='0.1.27'), 'Failed to clone staged repo'
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._clone_dir.cleanup()
+        super().tearDownClass()
+
+    def test_creates_tarball(self):
+        """export_repository should create a .tar.gz archive."""
+        with TemporaryDirectory(suffix='.git_export_out') as tmp:
+            basepath = os.path.join(tmp, 'export')
+            client = GitClient(self._clone_path)
+
+            result = client.export_repository('HEAD', basepath)
+            self.assertTrue(result)
+            self.assertTrue(os.path.isfile(basepath + '.tar.gz'))
+
+            with tarfile.open(basepath + '.tar.gz', 'r:gz') as tar:
+                names = tar.getnames()
+                self.assertTrue(len(names) > 0)
+
+    def test_at_tag(self):
+        """export_repository at a specific tag should succeed."""
+        with TemporaryDirectory(suffix='.git_export_out') as tmp:
+            basepath = os.path.join(tmp, 'export_tag')
+            client = GitClient(self._clone_path)
+
+            result = client.export_repository('0.1.27', basepath)
+            self.assertTrue(result)
+            self.assertTrue(os.path.isfile(basepath + '.tar.gz'))
+
+    def test_invalid_version(self):
+        """export_repository with a bad ref should return False."""
+        with TemporaryDirectory(suffix='.git_export_out') as tmp:
+            basepath = os.path.join(tmp, 'export_bad')
+            client = GitClient(self._clone_path)
+
+            result = client.export_repository('nonexistent-ref-xyz', basepath)
+            self.assertFalse(result)
+
+    def test_contains_license(self):
+        """Exported archive at a tag after LICENSE merge should contain LICENSE."""
+        with TemporaryDirectory(suffix='.git_export_out') as tmp:
+            basepath = os.path.join(tmp, 'export_license')
+            client = GitClient(self._clone_path)
+
+            result = client.export_repository('0.1.27', basepath)
+            self.assertTrue(result)
+
+            with tarfile.open(basepath + '.tar.gz', 'r:gz') as tar:
+                names = tar.getnames()
+                self.assertTrue(
+                    any('LICENSE' in n for n in names),
+                    f'LICENSE not found in archive: {names}',
+                )
+
+
+if __name__ == '__main__':
+    unittest.main()