1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
|
# test_bundle.py -- test bundle compatibility with CGit
# Copyright (C) 2025 Jelmer Vernooij <jelmer@jelmer.uk>
#
# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
# General Public License as published by the Free Software Foundation; version 2.0
# or (at your option) any later version. You can redistribute it and/or
# modify it under the terms of either of these two licenses.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# You should have received a copy of the licenses; if not, see
# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
# License, Version 2.0.
#
"""Tests for bundle compatibility with CGit."""
import os
import tempfile
from dulwich.bundle import create_bundle_from_repo, read_bundle, write_bundle
from dulwich.objects import Commit, Tree
from dulwich.repo import Repo
from .utils import CompatTestCase, rmtree_ro, run_git_or_fail
class CompatBundleTestCase(CompatTestCase):
def setUp(self) -> None:
super().setUp()
self.test_dir = tempfile.mkdtemp()
self.addCleanup(rmtree_ro, self.test_dir)
self.repo_path = os.path.join(self.test_dir, "repo")
self.repo = Repo.init(self.repo_path, mkdir=True)
self.addCleanup(self.repo.close)
def test_create_bundle_git_compat(self) -> None:
"""Test creating a bundle that git can read."""
# Create a commit
commit = Commit()
commit.committer = commit.author = b"Test User <test@example.com>"
commit.commit_time = commit.author_time = 1234567890
commit.commit_timezone = commit.author_timezone = 0
commit.message = b"Test commit"
tree = Tree()
self.repo.object_store.add_object(tree)
commit.tree = tree.id
self.repo.object_store.add_object(commit)
# Update ref
self.repo.refs[b"refs/heads/master"] = commit.id
# Create bundle using dulwich
bundle_path = os.path.join(self.test_dir, "test.bundle")
# Use create_bundle_from_repo helper
bundle = create_bundle_from_repo(self.repo)
self.addCleanup(bundle.close)
with open(bundle_path, "wb") as f:
write_bundle(f, bundle)
# Verify git can read the bundle (must run from a repo directory)
output = run_git_or_fail(["bundle", "verify", bundle_path], cwd=self.repo_path)
self.assertIn(b"The bundle contains", output)
self.assertIn(b"refs/heads/master", output)
def test_read_git_bundle(self) -> None:
"""Test reading a bundle created by git."""
# Create a commit using git
run_git_or_fail(["config", "user.name", "Test User"], cwd=self.repo_path)
run_git_or_fail(
["config", "user.email", "test@example.com"], cwd=self.repo_path
)
# Create a file and commit
test_file = os.path.join(self.repo_path, "test.txt")
with open(test_file, "w") as f:
f.write("test content\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Test commit"], cwd=self.repo_path)
# Create bundle using git
bundle_path = os.path.join(self.test_dir, "git.bundle")
run_git_or_fail(["bundle", "create", bundle_path, "HEAD"], cwd=self.repo_path)
# Read bundle using dulwich
with open(bundle_path, "rb") as f:
bundle = read_bundle(f)
self.addCleanup(bundle.close)
# Verify bundle contents
self.assertEqual(2, bundle.version)
self.assertIn(b"HEAD", bundle.references)
self.assertEqual({}, bundle.capabilities)
self.assertEqual([], bundle.prerequisites)
def test_read_git_bundle_multiple_refs(self) -> None:
"""Test reading a bundle with multiple references created by git."""
# Create commits and branches using git
run_git_or_fail(["config", "user.name", "Test User"], cwd=self.repo_path)
run_git_or_fail(
["config", "user.email", "test@example.com"], cwd=self.repo_path
)
# Create initial commit
test_file = os.path.join(self.repo_path, "test.txt")
with open(test_file, "w") as f:
f.write("initial content\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Initial commit"], cwd=self.repo_path)
# Create feature branch
run_git_or_fail(["checkout", "-b", "feature"], cwd=self.repo_path)
with open(test_file, "w") as f:
f.write("feature content\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Feature commit"], cwd=self.repo_path)
# Create another branch
run_git_or_fail(["checkout", "-b", "develop", "master"], cwd=self.repo_path)
dev_file = os.path.join(self.repo_path, "dev.txt")
with open(dev_file, "w") as f:
f.write("dev content\n")
run_git_or_fail(["add", "dev.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Dev commit"], cwd=self.repo_path)
# Create bundle with all branches
bundle_path = os.path.join(self.test_dir, "multi_ref.bundle")
run_git_or_fail(["bundle", "create", bundle_path, "--all"], cwd=self.repo_path)
# Read bundle using dulwich
with open(bundle_path, "rb") as f:
bundle = read_bundle(f)
self.addCleanup(bundle.close)
# Verify bundle contains all refs
self.assertIn(b"refs/heads/master", bundle.references)
self.assertIn(b"refs/heads/feature", bundle.references)
self.assertIn(b"refs/heads/develop", bundle.references)
def test_read_git_bundle_with_prerequisites(self) -> None:
"""Test reading a bundle with prerequisites created by git."""
# Create initial commits using git
run_git_or_fail(["config", "user.name", "Test User"], cwd=self.repo_path)
run_git_or_fail(
["config", "user.email", "test@example.com"], cwd=self.repo_path
)
# Create base commits
test_file = os.path.join(self.repo_path, "test.txt")
with open(test_file, "w") as f:
f.write("content 1\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Commit 1"], cwd=self.repo_path)
with open(test_file, "a") as f:
f.write("content 2\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Commit 2"], cwd=self.repo_path)
# Get the first commit hash to use as base
first_commit = run_git_or_fail(
["rev-parse", "HEAD~1"], cwd=self.repo_path
).strip()
# Create more commits
with open(test_file, "a") as f:
f.write("content 3\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Commit 3"], cwd=self.repo_path)
# Create bundle with prerequisites (only commits after first_commit)
bundle_path = os.path.join(self.test_dir, "prereq.bundle")
run_git_or_fail(
["bundle", "create", bundle_path, f"{first_commit.decode()}..HEAD"],
cwd=self.repo_path,
)
# Read bundle using dulwich
with open(bundle_path, "rb") as f:
bundle = read_bundle(f)
self.addCleanup(bundle.close)
# Verify bundle has prerequisites
self.assertGreater(len(bundle.prerequisites), 0)
# The prerequisite should be the first commit
prereq_ids = [p[0] for p in bundle.prerequisites]
self.assertIn(first_commit, prereq_ids)
def test_read_git_bundle_complex_pack(self) -> None:
"""Test reading a bundle with complex pack data (multiple objects) created by git."""
# Create a more complex repository structure
run_git_or_fail(["config", "user.name", "Test User"], cwd=self.repo_path)
run_git_or_fail(
["config", "user.email", "test@example.com"], cwd=self.repo_path
)
# Create multiple files in subdirectories
os.makedirs(os.path.join(self.repo_path, "src", "main"), exist_ok=True)
os.makedirs(os.path.join(self.repo_path, "tests"), exist_ok=True)
# Add various file types
files = [
("README.md", "# Test Project\n\nThis is a test."),
("src/main/app.py", "def main():\n print('Hello')\n"),
("src/main/utils.py", "def helper():\n return 42\n"),
("tests/test_app.py", "def test_main():\n pass\n"),
(".gitignore", "*.pyc\n__pycache__/\n"),
]
for filepath, content in files:
full_path = os.path.join(self.repo_path, filepath)
with open(full_path, "w") as f:
f.write(content)
run_git_or_fail(["add", filepath], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Initial complex commit"], cwd=self.repo_path)
# Make additional changes
with open(os.path.join(self.repo_path, "src/main/app.py"), "a") as f:
f.write("\nif __name__ == '__main__':\n main()\n")
run_git_or_fail(["add", "src/main/app.py"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Update app.py"], cwd=self.repo_path)
# Create bundle
bundle_path = os.path.join(self.test_dir, "complex.bundle")
run_git_or_fail(["bundle", "create", bundle_path, "HEAD"], cwd=self.repo_path)
# Read bundle using dulwich
with open(bundle_path, "rb") as f:
bundle = read_bundle(f)
self.addCleanup(bundle.close)
# Verify bundle contents
self.assertEqual(2, bundle.version)
self.assertIn(b"HEAD", bundle.references)
# Verify pack data exists
self.assertIsNotNone(bundle.pack_data)
self.assertGreater(len(bundle.pack_data), 0)
def test_clone_from_git_bundle(self) -> None:
"""Test cloning from a bundle created by git."""
# Create a repository with some history
run_git_or_fail(["config", "user.name", "Test User"], cwd=self.repo_path)
run_git_or_fail(
["config", "user.email", "test@example.com"], cwd=self.repo_path
)
# Create commits
test_file = os.path.join(self.repo_path, "test.txt")
for i in range(3):
with open(test_file, "a") as f:
f.write(f"Line {i}\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", f"Commit {i}"], cwd=self.repo_path)
# Create bundle
bundle_path = os.path.join(self.test_dir, "clone.bundle")
run_git_or_fail(["bundle", "create", bundle_path, "HEAD"], cwd=self.repo_path)
# Read bundle using dulwich
with open(bundle_path, "rb") as f:
bundle = read_bundle(f)
self.addCleanup(bundle.close)
# Verify bundle was read correctly
self.assertEqual(2, bundle.version)
self.assertIn(b"HEAD", bundle.references)
self.assertEqual([], bundle.prerequisites)
# Verify pack data exists
self.assertIsNotNone(bundle.pack_data)
self.assertGreater(len(bundle.pack_data), 0)
# Use git to verify the bundle can be used for cloning
clone_path = os.path.join(self.test_dir, "cloned_repo")
run_git_or_fail(["clone", bundle_path, clone_path])
# Verify the cloned repository exists and has content
self.assertTrue(os.path.exists(clone_path))
self.assertTrue(os.path.exists(os.path.join(clone_path, "test.txt")))
def test_unbundle_git_bundle(self) -> None:
"""Test unbundling a bundle created by git using dulwich CLI."""
# Create a repository with commits using git
run_git_or_fail(["config", "user.name", "Test User"], cwd=self.repo_path)
run_git_or_fail(
["config", "user.email", "test@example.com"], cwd=self.repo_path
)
# Create commits
test_file = os.path.join(self.repo_path, "test.txt")
with open(test_file, "w") as f:
f.write("content 1\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Commit 1"], cwd=self.repo_path)
with open(test_file, "a") as f:
f.write("content 2\n")
run_git_or_fail(["add", "test.txt"], cwd=self.repo_path)
run_git_or_fail(["commit", "-m", "Commit 2"], cwd=self.repo_path)
# Get commit SHA for verification
head_sha = run_git_or_fail(["rev-parse", "HEAD"], cwd=self.repo_path).strip()
# Create bundle using git
bundle_path = os.path.join(self.test_dir, "unbundle_test.bundle")
run_git_or_fail(["bundle", "create", bundle_path, "master"], cwd=self.repo_path)
# Create a new empty repository to unbundle into
unbundle_repo_path = os.path.join(self.test_dir, "unbundle_repo")
unbundle_repo = Repo.init(unbundle_repo_path, mkdir=True)
self.addCleanup(unbundle_repo.close)
# Read the bundle and store objects using dulwich
with open(bundle_path, "rb") as f:
bundle = read_bundle(f)
self.addCleanup(bundle.close)
# Use the bundle's store_objects method to unbundle
bundle.store_objects(unbundle_repo.object_store)
# Verify objects are now in the repository
# Check that the HEAD commit exists
self.assertIn(head_sha, unbundle_repo.object_store)
# Verify we can retrieve the commit
commit = unbundle_repo.object_store[head_sha]
self.assertEqual(b"Commit 2\n", commit.message)
|
