FossilRepo

fossilrepo / tests / test_bundle_cli.py
Source Blame History 172 lines
c588255… ragelink 1 """Tests for the fossilrepo-ctl bundle export/import commands."""
c588255… ragelink 2
c588255… ragelink 3 from unittest.mock import MagicMock, patch
c588255… ragelink 4
c588255… ragelink 5 import pytest
c588255… ragelink 6 from click.testing import CliRunner
c588255… ragelink 7
c588255… ragelink 8 from ctl.main import cli
c588255… ragelink 9
c588255… ragelink 10
c588255… ragelink 11 @pytest.fixture
c588255… ragelink 12 def runner():
c588255… ragelink 13 return CliRunner()
c588255… ragelink 14
c588255… ragelink 15
c588255… ragelink 16 @pytest.mark.django_db
c588255… ragelink 17 class TestBundleExport:
c588255… ragelink 18 def test_export_missing_project(self, runner):
c588255… ragelink 19 """Export with a non-existent project slug prints an error."""
c588255… ragelink 20 result = runner.invoke(cli, ["bundle", "export", "nonexistent-project", "/tmp/out.bundle"])
c588255… ragelink 21 assert result.exit_code == 0
c588255… ragelink 22 assert "No repository found" in result.output
c588255… ragelink 23
c588255… ragelink 24 def test_export_repo_not_on_disk(self, runner, sample_project):
c588255… ragelink 25 """Export when the .fossil file does not exist on disk."""
313537c… ragelink 26 from fossil.models import FossilRepository
313537c… ragelink 27
313537c… ragelink 28 with patch.object(type(FossilRepository.objects.get(project=sample_project)), "exists_on_disk", new_callable=lambda: property(lambda self: False)):
313537c… ragelink 29 result = runner.invoke(cli, ["bundle", "export", sample_project.slug, "/tmp/out.bundle"])
c588255… ragelink 30 assert result.exit_code == 0
c588255… ragelink 31 assert "not found on disk" in result.output
c588255… ragelink 32
c588255… ragelink 33 def test_export_fossil_not_available(self, runner, sample_project):
c588255… ragelink 34 """Export when fossil binary is not found."""
c588255… ragelink 35 from fossil.models import FossilRepository
c588255… ragelink 36
c588255… ragelink 37 repo = FossilRepository.objects.get(project=sample_project)
c588255… ragelink 38
c588255… ragelink 39 with (
c588255… ragelink 40 patch.object(type(repo), "exists_on_disk", new_callable=lambda: property(lambda self: True)),
c588255… ragelink 41 patch("fossil.cli.FossilCLI.is_available", return_value=False),
c588255… ragelink 42 ):
c588255… ragelink 43 result = runner.invoke(cli, ["bundle", "export", sample_project.slug, "/tmp/out.bundle"])
c588255… ragelink 44 assert result.exit_code == 0
c588255… ragelink 45 assert "Fossil binary not found" in result.output
c588255… ragelink 46
c588255… ragelink 47 def test_export_success(self, runner, sample_project, tmp_path):
c588255… ragelink 48 """Export succeeds when fossil binary is available and returns 0."""
c588255… ragelink 49 from fossil.models import FossilRepository
c588255… ragelink 50
c588255… ragelink 51 repo = FossilRepository.objects.get(project=sample_project)
c588255… ragelink 52 output_path = tmp_path / "test.bundle"
c588255… ragelink 53
c588255… ragelink 54 mock_run = MagicMock()
c588255… ragelink 55 mock_run.returncode = 0
c588255… ragelink 56 mock_run.stdout = ""
c588255… ragelink 57 mock_run.stderr = ""
c588255… ragelink 58
c588255… ragelink 59 with (
c588255… ragelink 60 patch.object(type(repo), "exists_on_disk", new_callable=lambda: property(lambda self: True)),
c588255… ragelink 61 patch("fossil.cli.FossilCLI.is_available", return_value=True),
c588255… ragelink 62 patch("subprocess.run", return_value=mock_run),
c588255… ragelink 63 ):
c588255… ragelink 64 # Create a fake output file so size calculation works
c588255… ragelink 65 output_path.write_bytes(b"x" * 1024)
c588255… ragelink 66 result = runner.invoke(cli, ["bundle", "export", sample_project.slug, str(output_path)])
c588255… ragelink 67 assert result.exit_code == 0
c588255… ragelink 68 assert "Success" in result.output
c588255… ragelink 69
c588255… ragelink 70 def test_export_failure(self, runner, sample_project, tmp_path):
c588255… ragelink 71 """Export reports failure when fossil returns non-zero."""
c588255… ragelink 72 from fossil.models import FossilRepository
c588255… ragelink 73
c588255… ragelink 74 repo = FossilRepository.objects.get(project=sample_project)
c588255… ragelink 75 output_path = tmp_path / "test.bundle"
c588255… ragelink 76
c588255… ragelink 77 mock_run = MagicMock()
c588255… ragelink 78 mock_run.returncode = 1
c588255… ragelink 79 mock_run.stdout = ""
c588255… ragelink 80 mock_run.stderr = "bundle export failed"
c588255… ragelink 81
c588255… ragelink 82 with (
c588255… ragelink 83 patch.object(type(repo), "exists_on_disk", new_callable=lambda: property(lambda self: True)),
c588255… ragelink 84 patch("fossil.cli.FossilCLI.is_available", return_value=True),
c588255… ragelink 85 patch("subprocess.run", return_value=mock_run),
c588255… ragelink 86 ):
c588255… ragelink 87 result = runner.invoke(cli, ["bundle", "export", sample_project.slug, str(output_path)])
c588255… ragelink 88 assert result.exit_code == 0
c588255… ragelink 89 assert "Failed" in result.output
c588255… ragelink 90
c588255… ragelink 91
c588255… ragelink 92 @pytest.mark.django_db
c588255… ragelink 93 class TestBundleImport:
c588255… ragelink 94 def test_import_missing_project(self, runner):
c588255… ragelink 95 """Import with a non-existent project slug prints an error."""
c588255… ragelink 96 result = runner.invoke(cli, ["bundle", "import", "nonexistent-project", "/tmp/in.bundle"])
c588255… ragelink 97 assert result.exit_code == 0
c588255… ragelink 98 assert "No repository found" in result.output
c588255… ragelink 99
c588255… ragelink 100 def test_import_bundle_file_not_found(self, runner, sample_project):
c588255… ragelink 101 """Import when the bundle file does not exist."""
c588255… ragelink 102 from fossil.models import FossilRepository
c588255… ragelink 103
c588255… ragelink 104 repo = FossilRepository.objects.get(project=sample_project)
c588255… ragelink 105
c588255… ragelink 106 with patch.object(type(repo), "exists_on_disk", new_callable=lambda: property(lambda self: True)):
c588255… ragelink 107 result = runner.invoke(cli, ["bundle", "import", sample_project.slug, "/tmp/definitely-not-a-file.bundle"])
c588255… ragelink 108 assert result.exit_code == 0
c588255… ragelink 109 assert "not found" in result.output.lower()
c588255… ragelink 110
c588255… ragelink 111 def test_import_success(self, runner, sample_project, tmp_path):
c588255… ragelink 112 """Import succeeds when fossil binary is available and returns 0."""
c588255… ragelink 113 from fossil.models import FossilRepository
c588255… ragelink 114
c588255… ragelink 115 repo = FossilRepository.objects.get(project=sample_project)
c588255… ragelink 116 bundle_file = tmp_path / "test.bundle"
c588255… ragelink 117 bundle_file.write_bytes(b"fake-bundle")
c588255… ragelink 118
c588255… ragelink 119 mock_run = MagicMock()
c588255… ragelink 120 mock_run.returncode = 0
c588255… ragelink 121 mock_run.stdout = "imported 42 artifacts"
c588255… ragelink 122 mock_run.stderr = ""
c588255… ragelink 123
c588255… ragelink 124 with (
c588255… ragelink 125 patch.object(type(repo), "exists_on_disk", new_callable=lambda: property(lambda self: True)),
c588255… ragelink 126 patch("fossil.cli.FossilCLI.is_available", return_value=True),
c588255… ragelink 127 patch("subprocess.run", return_value=mock_run),
c588255… ragelink 128 ):
c588255… ragelink 129 result = runner.invoke(cli, ["bundle", "import", sample_project.slug, str(bundle_file)])
c588255… ragelink 130 assert result.exit_code == 0
c588255… ragelink 131 assert "Success" in result.output
c588255… ragelink 132
c588255… ragelink 133 def test_import_failure(self, runner, sample_project, tmp_path):
c588255… ragelink 134 """Import reports failure when fossil returns non-zero."""
c588255… ragelink 135 from fossil.models import FossilRepository
c588255… ragelink 136
c588255… ragelink 137 repo = FossilRepository.objects.get(project=sample_project)
c588255… ragelink 138 bundle_file = tmp_path / "test.bundle"
c588255… ragelink 139 bundle_file.write_bytes(b"fake-bundle")
c588255… ragelink 140
c588255… ragelink 141 mock_run = MagicMock()
c588255… ragelink 142 mock_run.returncode = 1
c588255… ragelink 143 mock_run.stdout = ""
c588255… ragelink 144 mock_run.stderr = "invalid bundle format"
c588255… ragelink 145
c588255… ragelink 146 with (
c588255… ragelink 147 patch.object(type(repo), "exists_on_disk", new_callable=lambda: property(lambda self: True)),
c588255… ragelink 148 patch("fossil.cli.FossilCLI.is_available", return_value=True),
c588255… ragelink 149 patch("subprocess.run", return_value=mock_run),
c588255… ragelink 150 ):
c588255… ragelink 151 result = runner.invoke(cli, ["bundle", "import", sample_project.slug, str(bundle_file)])
c588255… ragelink 152 assert result.exit_code == 0
c588255… ragelink 153 assert "Failed" in result.output
c588255… ragelink 154
c588255… ragelink 155
c588255… ragelink 156 class TestBundleCLIGroup:
c588255… ragelink 157 def test_bundle_help(self, runner):
c588255… ragelink 158 """Bundle group shows help text."""
c588255… ragelink 159 result = runner.invoke(cli, ["bundle", "--help"])
c588255… ragelink 160 assert result.exit_code == 0
c588255… ragelink 161 assert "export" in result.output.lower()
c588255… ragelink 162 assert "import" in result.output.lower()
c588255… ragelink 163
c588255… ragelink 164 def test_export_help(self, runner):
c588255… ragelink 165 result = runner.invoke(cli, ["bundle", "export", "--help"])
c588255… ragelink 166 assert result.exit_code == 0
c588255… ragelink 167 assert "PROJECT_SLUG" in result.output
c588255… ragelink 168
c588255… ragelink 169 def test_import_help(self, runner):
c588255… ragelink 170 result = runner.invoke(cli, ["bundle", "import", "--help"])
c588255… ragelink 171 assert result.exit_code == 0
c588255… ragelink 172 assert "PROJECT_SLUG" in result.output

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button