test_cloud_gcs.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177
  1. #!/usr/bin/python3
  2. #
  3. # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
  4. # General Public License as public by the Free Software Foundation; version 2.0
  5. # or (at your option) any later version. You can redistribute it and/or
  6. # modify it under the terms of either of these two licenses.
  7. #
  8. # Unless required by applicable law or agreed to in writing, software
  9. # distributed under the License is distributed on an "AS IS" BASIS,
  10. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11. # See the License for the specific language governing permissions and
  12. # limitations under the License.
  13. #
  14. # You should have received a copy of the licenses; if not, see
  15. # <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
  16. # and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
  17. # License, Version 2.0.
  18. """Tests for dulwich.cloud.gcs."""
  19. import unittest
  20. from unittest.mock import MagicMock, patch
  21. from dulwich.cloud.gcs import GcsObjectStore
  22. class GcsObjectStoreTests(unittest.TestCase):
  23. """Tests for the GcsObjectStore class."""
  24. def setUp(self):
  25. self.mock_bucket = MagicMock()
  26. self.store = GcsObjectStore(self.mock_bucket, subpath="git")
  27. def test_init(self):
  28. """Test initialization and repr."""
  29. self.assertEqual(self.mock_bucket, self.store.bucket)
  30. self.assertEqual("git", self.store.subpath)
  31. self.assertIn("GcsObjectStore", repr(self.store))
  32. self.assertIn("git", repr(self.store))
  33. def test_remove_pack(self):
  34. """Test _remove_pack method."""
  35. self.store._remove_pack("pack-1234")
  36. self.mock_bucket.delete_blobs.assert_called_once()
  37. args = self.mock_bucket.delete_blobs.call_args[0][0]
  38. self.assertEqual(
  39. sorted(args), sorted(["git/pack-1234.pack", "git/pack-1234.idx"])
  40. )
  41. def test_iter_pack_names(self):
  42. """Test _iter_pack_names method."""
  43. # Create mock blobs with the expected attributes
  44. mock_blob1 = MagicMock()
  45. mock_blob1.name = "git/pack-1234.pack"
  46. mock_blob2 = MagicMock()
  47. mock_blob2.name = "git/pack-1234.idx"
  48. mock_blob3 = MagicMock()
  49. mock_blob3.name = "git/pack-5678.pack"
  50. # Only pack-1234 has both .pack and .idx files
  51. self.mock_bucket.list_blobs.return_value = [mock_blob1, mock_blob2, mock_blob3]
  52. pack_names = list(self.store._iter_pack_names())
  53. self.assertEqual(["pack-1234"], pack_names)
  54. # Verify that list_blobs was called with the correct prefix
  55. self.mock_bucket.list_blobs.assert_called_once_with(prefix="git")
  56. def test_load_pack_data(self):
  57. """Test _load_pack_data method."""
  58. # Create a mock blob that will simulate downloading a pack file
  59. mock_blob = MagicMock()
  60. self.mock_bucket.blob.return_value = mock_blob
  61. # We need to patch PackData to avoid actual pack format validation
  62. with patch("dulwich.cloud.gcs.PackData") as mock_pack_data_cls:
  63. mock_pack_data = MagicMock()
  64. mock_pack_data_cls.return_value = mock_pack_data
  65. # Mock the download_to_file to actually write some data
  66. def mock_download_to_file(file_obj):
  67. file_obj.write(b"not-a-real-pack-file")
  68. mock_blob.download_to_file.side_effect = mock_download_to_file
  69. # Call the method under test
  70. result = self.store._load_pack_data("pack-1234")
  71. # Check results
  72. self.assertEqual(mock_pack_data, result)
  73. self.mock_bucket.blob.assert_called_once_with("git/pack-1234.pack")
  74. mock_blob.download_to_file.assert_called_once()
  75. # Verify PackData was called with the correct parameters
  76. mock_pack_data_cls.assert_called_once()
  77. args, _ = mock_pack_data_cls.call_args
  78. self.assertEqual("pack-1234.pack", args[0])
  79. def test_load_pack_index(self):
  80. """Test _load_pack_index method."""
  81. # We need to patch load_pack_index_file since we don't want to test its internals
  82. with patch("dulwich.cloud.gcs.load_pack_index_file") as mock_load:
  83. # Create a mock blob that will simulate downloading an index file
  84. mock_blob = MagicMock()
  85. self.mock_bucket.blob.return_value = mock_blob
  86. # Mock the download_to_file to actually write some data
  87. def mock_download_to_file(file_obj):
  88. file_obj.write(b"index-file-content")
  89. mock_blob.download_to_file.side_effect = mock_download_to_file
  90. # Setup the return value for the mocked load_pack_index_file
  91. mock_index = MagicMock()
  92. mock_load.return_value = mock_index
  93. # Call the method under test
  94. result = self.store._load_pack_index("pack-1234")
  95. # Check results
  96. self.assertEqual(mock_index, result)
  97. self.mock_bucket.blob.assert_called_once_with("git/pack-1234.idx")
  98. mock_blob.download_to_file.assert_called_once()
  99. mock_load.assert_called_once()
  100. # Verify the correct filename is passed
  101. self.assertEqual("pack-1234.idx", mock_load.call_args[0][0])
  102. def test_get_pack(self):
  103. """Test _get_pack method."""
  104. with patch("dulwich.cloud.gcs.Pack") as mock_pack_cls:
  105. mock_pack = MagicMock()
  106. mock_pack_cls.from_lazy_objects.return_value = mock_pack
  107. result = self.store._get_pack("pack-1234")
  108. self.assertEqual(mock_pack, result)
  109. mock_pack_cls.from_lazy_objects.assert_called_once()
  110. # Get the lazy loaders that were passed
  111. args = mock_pack_cls.from_lazy_objects.call_args[0]
  112. self.assertEqual(2, len(args))
  113. # They should be callables
  114. self.assertTrue(callable(args[0]))
  115. self.assertTrue(callable(args[1]))
  116. def test_upload_pack(self):
  117. """Test _upload_pack method."""
  118. # Create mock blob objects
  119. mock_idx_blob = MagicMock()
  120. mock_data_blob = MagicMock()
  121. # Configure the bucket's blob method to return the appropriate mock
  122. def mock_blob(path):
  123. if path.endswith(".idx"):
  124. return mock_idx_blob
  125. elif path.endswith(".pack"):
  126. return mock_data_blob
  127. else:
  128. self.fail(f"Unexpected blob path: {path}")
  129. self.mock_bucket.blob.side_effect = mock_blob
  130. # Create mock file objects
  131. mock_index_file = MagicMock()
  132. mock_pack_file = MagicMock()
  133. # Call the method under test
  134. self.store._upload_pack("pack-1234", mock_pack_file, mock_index_file)
  135. # Verify the correct paths were used
  136. self.mock_bucket.blob.assert_any_call("git/pack-1234.idx")
  137. self.mock_bucket.blob.assert_any_call("git/pack-1234.pack")
  138. # Verify the uploads were called with the right files
  139. mock_idx_blob.upload_from_file.assert_called_once_with(mock_index_file)
  140. mock_data_blob.upload_from_file.assert_called_once_with(mock_pack_file)