(self)
| 542 | ) |
| 543 | |
| 544 | def test_codec_options(self): |
| 545 | with self.assertRaisesRegex(TypeError, "codec_options must be"): |
| 546 | self.create_client_encryption( |
| 547 | KMS_PROVIDERS, |
| 548 | "keyvault.datakeys", |
| 549 | client_context.client, |
| 550 | None, # type: ignore[arg-type] |
| 551 | ) |
| 552 | |
| 553 | opts = CodecOptions(uuid_representation=UuidRepresentation.JAVA_LEGACY) |
| 554 | client_encryption_legacy = self.create_client_encryption( |
| 555 | KMS_PROVIDERS, "keyvault.datakeys", client_context.client, opts |
| 556 | ) |
| 557 | |
| 558 | # Create the encrypted field's data key. |
| 559 | key_id = client_encryption_legacy.create_data_key("local") |
| 560 | |
| 561 | # Encrypt a UUID with JAVA_LEGACY codec options. |
| 562 | value = uuid.uuid4() |
| 563 | encrypted_legacy = client_encryption_legacy.encrypt( |
| 564 | value, Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic, key_id=key_id |
| 565 | ) |
| 566 | decrypted_value_legacy = client_encryption_legacy.decrypt(encrypted_legacy) |
| 567 | self.assertEqual(decrypted_value_legacy, value) |
| 568 | |
| 569 | # Encrypt the same UUID with STANDARD codec options. |
| 570 | opts = CodecOptions(uuid_representation=UuidRepresentation.STANDARD) |
| 571 | client_encryption = self.create_client_encryption( |
| 572 | KMS_PROVIDERS, "keyvault.datakeys", client_context.client, opts |
| 573 | ) |
| 574 | encrypted_standard = client_encryption.encrypt( |
| 575 | value, Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic, key_id=key_id |
| 576 | ) |
| 577 | decrypted_standard = client_encryption.decrypt(encrypted_standard) |
| 578 | self.assertEqual(decrypted_standard, value) |
| 579 | |
| 580 | # Test that codec_options is applied during encryption. |
| 581 | self.assertNotEqual(encrypted_standard, encrypted_legacy) |
| 582 | # Test that codec_options is applied during decryption. |
| 583 | self.assertEqual( |
| 584 | client_encryption_legacy.decrypt(encrypted_standard), Binary.from_uuid(value) |
| 585 | ) |
| 586 | self.assertNotEqual(client_encryption.decrypt(encrypted_legacy), value) |
| 587 | |
| 588 | def test_close(self): |
| 589 | client_encryption = self.create_client_encryption( |
nothing calls this directly
no test coverage detected