unit tests

tests/test_upload.py

@@ -872,6 +872,75 @@ def test_process_bidi_write_grpc_invalid_checksum(self):
             grpc.StatusCode.FAILED_PRECONDITION, unittest.mock.ANY
         )
 
+    def test_process_bidi_write_grpc_checksums(self):
+        media = b"The quick brown fox jumps over the lazy dog"
+        proto_crc32c = crc32c.crc32c(media)
+        proto_md5_hash = hashlib.md5(media).digest()
+
+        TEST_CASES = {
+            "both": {
+                "checksums": storage_pb2.ObjectChecksums(
+                    crc32c=proto_crc32c, md5_hash=proto_md5_hash
+                ),
+                "expected": {
+                    "x_emulator_crc32c": testbench.common.rest_crc32c_from_proto(
+                        proto_crc32c
+                    ),
+                    "x_emulator_md5": testbench.common.rest_md5_from_proto(
+                        proto_md5_hash
+                    ),
+                },
+            },
+            "only md5": {
+                "checksums": storage_pb2.ObjectChecksums(md5_hash=proto_md5_hash),
+                "expected": {
+                    "x_emulator_md5": testbench.common.rest_md5_from_proto(
+                        proto_md5_hash
+                    ),
+                    "x_emulator_no_crc32c": "true",
+                },
+            },
+            "only crc32c": {
+                "checksums": storage_pb2.ObjectChecksums(crc32c=proto_crc32c),
+                "expected": {
+                    "x_emulator_crc32c": testbench.common.rest_crc32c_from_proto(
+                        proto_crc32c
+                    ),
+                    "x_emulator_no_md5": "true",
+                },
+            },
+        }
+        for name, test in TEST_CASES.items():
+            request = testbench.common.FakeRequest(
+                args={}, data=json.dumps({"name": "bucket-name"})
+            )
+            bucket, _ = gcs.bucket.Bucket.init(request, None)
+            spec = storage_pb2.WriteObjectSpec(
+                resource={"name": "object", "bucket": "projects/_/buckets/bucket-name"},
+            )
+            request = storage_pb2.BidiWriteObjectRequest(
+                write_object_spec=spec,
+                write_offset=0,
+                object_checksums=test["checksums"],
+                finish_write=True,
+            )
+
+            context = self.mock_context()
+            db = unittest.mock.Mock()
+            db.get_bucket = unittest.mock.MagicMock(return_value=bucket)
+            responses = list(
+                gcs.upload.Upload.process_bidi_write_object_grpc(db, [request], context)
+            )
+            # Verify the annotations inserted by the testbench.
+            annotations = responses[0].resource.metadata
+            expected = test["expected"]
+            self.maxDiff = None
+            self.assertEqual(
+                annotations,
+                {**annotations, **expected},
+                msg="Testing with %s checksums" % name,
+            )
+
     def test_process_bidi_write_grpc_empty(self):
         db = unittest.mock.Mock()
         context = self.mock_context()