2
0

test_dumb.py 9.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264
  1. # test_dumb.py -- Tests for dumb HTTP git repositories
  2. # Copyright (C) 2025 Dulwich contributors
  3. #
  4. # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
  5. # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
  6. # General Public License as public by the Free Software Foundation; version 2.0
  7. # or (at your option) any later version. You can redistribute it and/or
  8. # modify it under the terms of either of these two licenses.
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. #
  16. # You should have received a copy of the licenses; if not, see
  17. # <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
  18. # and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
  19. # License, Version 2.0.
  20. #
  21. """Tests for dumb HTTP git repositories."""
  22. import zlib
  23. from unittest import TestCase
  24. from unittest.mock import Mock
  25. from dulwich.dumb import DumbHTTPObjectStore, DumbRemoteRepo
  26. from dulwich.errors import NotGitRepository
  27. from dulwich.objects import Blob, Commit, Tag, Tree, hex_to_sha, sha_to_hex
  28. class MockResponse:
  29. def __init__(self, status=200, content=b"", headers=None):
  30. self.status = status
  31. self.content = content
  32. self.headers = headers or {}
  33. self.closed = False
  34. def close(self):
  35. self.closed = True
  36. class DumbHTTPObjectStoreTests(TestCase):
  37. """Tests for DumbHTTPObjectStore."""
  38. def setUp(self):
  39. self.base_url = "https://example.com/repo.git/"
  40. self.responses = {}
  41. self.store = DumbHTTPObjectStore(self.base_url, self._mock_http_request)
  42. def _mock_http_request(self, url, headers):
  43. """Mock HTTP request function."""
  44. if url in self.responses:
  45. resp_data = self.responses[url]
  46. resp = MockResponse(
  47. resp_data.get("status", 200), resp_data.get("content", b"")
  48. )
  49. return resp, lambda size: resp.content
  50. else:
  51. resp = MockResponse(404)
  52. return resp, lambda size: b""
  53. def _add_response(self, path, content, status=200):
  54. """Add a mock response for a given path."""
  55. url = self.base_url + path
  56. self.responses[url] = {"status": status, "content": content}
  57. def _make_object(self, obj):
  58. """Create compressed git object data."""
  59. type_name = {
  60. Blob.type_num: b"blob",
  61. Tree.type_num: b"tree",
  62. Commit.type_num: b"commit",
  63. Tag.type_num: b"tag",
  64. }[obj.type_num]
  65. content = obj.as_raw_string()
  66. header = type_name + b" " + str(len(content)).encode() + b"\x00"
  67. return zlib.compress(header + content)
  68. def test_fetch_loose_object_blob(self):
  69. # Create a blob object
  70. blob = Blob()
  71. blob.data = b"Hello, world!"
  72. sha = blob.sha().digest()
  73. hex_sha = sha_to_hex(sha)
  74. # Add mock response
  75. path = f"objects/{hex_sha[:2]}/{hex_sha[2:]}"
  76. self._add_response(path, self._make_object(blob))
  77. # Fetch the object
  78. type_num, content = self.store._fetch_loose_object(sha)
  79. self.assertEqual(Blob.type_num, type_num)
  80. self.assertEqual(b"Hello, world!", content)
  81. def test_fetch_loose_object_not_found(self):
  82. sha = b"1" * 20
  83. self.assertRaises(KeyError, self.store._fetch_loose_object, sha)
  84. def test_fetch_loose_object_invalid_format(self):
  85. sha = b"1" * 20
  86. hex_sha = sha_to_hex(sha)
  87. path = f"objects/{hex_sha[:2]}/{hex_sha[2:]}"
  88. # Add invalid compressed data
  89. self._add_response(path, b"invalid data")
  90. self.assertRaises(Exception, self.store._fetch_loose_object, sha)
  91. def test_load_packs_empty(self):
  92. # No packs file
  93. self.store._load_packs()
  94. self.assertEqual([], self.store._packs)
  95. def test_load_packs_with_entries(self):
  96. packs_content = b"""P pack-1234567890abcdef1234567890abcdef12345678.pack
  97. P pack-abcdef1234567890abcdef1234567890abcdef12.pack
  98. """
  99. self._add_response("objects/info/packs", packs_content)
  100. self.store._load_packs()
  101. self.assertEqual(2, len(self.store._packs))
  102. self.assertEqual(
  103. "pack-1234567890abcdef1234567890abcdef12345678", self.store._packs[0][0]
  104. )
  105. self.assertEqual(
  106. "pack-abcdef1234567890abcdef1234567890abcdef12", self.store._packs[1][0]
  107. )
  108. def test_get_raw_from_cache(self):
  109. sha = b"1" * 20
  110. self.store._cached_objects[sha] = (Blob.type_num, b"cached content")
  111. type_num, content = self.store.get_raw(sha)
  112. self.assertEqual(Blob.type_num, type_num)
  113. self.assertEqual(b"cached content", content)
  114. def test_contains_loose(self):
  115. # Create a blob object
  116. blob = Blob()
  117. blob.data = b"Test blob"
  118. sha = blob.sha().digest()
  119. hex_sha = sha_to_hex(sha)
  120. # Add mock response
  121. path = f"objects/{hex_sha[:2]}/{hex_sha[2:]}"
  122. self._add_response(path, self._make_object(blob))
  123. self.assertTrue(self.store.contains_loose(sha))
  124. self.assertFalse(self.store.contains_loose(b"0" * 20))
  125. def test_add_object_not_implemented(self):
  126. blob = Blob()
  127. blob.data = b"test"
  128. self.assertRaises(NotImplementedError, self.store.add_object, blob)
  129. def test_add_objects_not_implemented(self):
  130. self.assertRaises(NotImplementedError, self.store.add_objects, [])
  131. class DumbRemoteRepoTests(TestCase):
  132. """Tests for DumbRemoteRepo."""
  133. def setUp(self):
  134. self.base_url = "https://example.com/repo.git/"
  135. self.responses = {}
  136. self.repo = DumbRemoteRepo(self.base_url, self._mock_http_request)
  137. def _mock_http_request(self, url, headers):
  138. """Mock HTTP request function."""
  139. if url in self.responses:
  140. resp_data = self.responses[url]
  141. resp = MockResponse(
  142. resp_data.get("status", 200), resp_data.get("content", b"")
  143. )
  144. return resp, lambda size: resp.content[:size] if size else resp.content
  145. else:
  146. resp = MockResponse(404)
  147. return resp, lambda size: b""
  148. def _add_response(self, path, content, status=200):
  149. """Add a mock response for a given path."""
  150. url = self.base_url + path
  151. self.responses[url] = {"status": status, "content": content}
  152. def test_get_refs(self):
  153. refs_content = b"""0123456789abcdef0123456789abcdef01234567\trefs/heads/master
  154. abcdef0123456789abcdef0123456789abcdef01\trefs/heads/develop
  155. fedcba9876543210fedcba9876543210fedcba98\trefs/tags/v1.0
  156. """
  157. self._add_response("info/refs", refs_content)
  158. refs = self.repo.get_refs()
  159. self.assertEqual(3, len(refs))
  160. self.assertEqual(
  161. hex_to_sha(b"0123456789abcdef0123456789abcdef01234567"),
  162. refs[b"refs/heads/master"],
  163. )
  164. self.assertEqual(
  165. hex_to_sha(b"abcdef0123456789abcdef0123456789abcdef01"),
  166. refs[b"refs/heads/develop"],
  167. )
  168. self.assertEqual(
  169. hex_to_sha(b"fedcba9876543210fedcba9876543210fedcba98"),
  170. refs[b"refs/tags/v1.0"],
  171. )
  172. def test_get_refs_not_found(self):
  173. self.assertRaises(NotGitRepository, self.repo.get_refs)
  174. def test_get_peeled(self):
  175. refs_content = b"0123456789abcdef0123456789abcdef01234567\trefs/heads/master\n"
  176. self._add_response("info/refs", refs_content)
  177. # For dumb HTTP, peeled just returns the ref value
  178. peeled = self.repo.get_peeled(b"refs/heads/master")
  179. self.assertEqual(
  180. hex_to_sha(b"0123456789abcdef0123456789abcdef01234567"), peeled
  181. )
  182. def test_fetch_pack_data_no_wants(self):
  183. refs_content = b"0123456789abcdef0123456789abcdef01234567\trefs/heads/master\n"
  184. self._add_response("info/refs", refs_content)
  185. graph_walker = Mock()
  186. def determine_wants(refs):
  187. return []
  188. result = list(self.repo.fetch_pack_data(graph_walker, determine_wants))
  189. self.assertEqual([], result)
  190. def test_fetch_pack_data_with_blob(self):
  191. # Set up refs
  192. refs_content = b"0123456789abcdef0123456789abcdef01234567\trefs/heads/master\n"
  193. self._add_response("info/refs", refs_content)
  194. # Create a simple blob object
  195. blob = Blob()
  196. blob.data = b"Test content"
  197. blob_sha = blob.sha().digest()
  198. # Add blob response
  199. self.repo._object_store._cached_objects[blob_sha] = (
  200. Blob.type_num,
  201. blob.as_raw_string(),
  202. )
  203. # Mock graph walker
  204. graph_walker = Mock()
  205. graph_walker.ack.return_value = [] # No existing objects
  206. def determine_wants(refs):
  207. return [blob_sha]
  208. result = list(self.repo.fetch_pack_data(graph_walker, determine_wants))
  209. self.assertEqual(1, len(result))
  210. self.assertEqual(Blob.type_num, result[0].pack_type_num)
  211. self.assertEqual(blob.as_raw_string(), result[0].obj)
  212. def test_object_store_property(self):
  213. self.assertIsInstance(self.repo.object_store, DumbHTTPObjectStore)
  214. self.assertEqual(self.base_url, self.repo.object_store.base_url)