(self, provider, master_key)
| 3050 | conn.close() |
| 3051 | |
| 3052 | def _test(self, provider, master_key): |
| 3053 | self.http_post("/reset") |
| 3054 | # Case 1: createDataKey and encrypt with TCP retry |
| 3055 | self.http_post("/set_failpoint/network", {"count": 1}) |
| 3056 | key_id = self.client_encryption.create_data_key(provider, master_key=master_key) |
| 3057 | self.http_post("/set_failpoint/network", {"count": 1}) |
| 3058 | self.client_encryption.encrypt( |
| 3059 | 123, Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic, key_id |
| 3060 | ) |
| 3061 | |
| 3062 | # Case 2: createDataKey and encrypt with HTTP retry |
| 3063 | self.http_post("/set_failpoint/http", {"count": 1}) |
| 3064 | key_id = self.client_encryption.create_data_key(provider, master_key=master_key) |
| 3065 | self.http_post("/set_failpoint/http", {"count": 1}) |
| 3066 | self.client_encryption.encrypt( |
| 3067 | 123, Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic, key_id |
| 3068 | ) |
| 3069 | |
| 3070 | # Case 3: createDataKey fails after too many retries |
| 3071 | self.http_post("/set_failpoint/network", {"count": 4}) |
| 3072 | with self.assertRaisesRegex(EncryptionError, "KMS request failed after"): |
| 3073 | self.client_encryption.create_data_key(provider, master_key=master_key) |
| 3074 | |
| 3075 | def test_kms_retry(self): |
| 3076 | if IS_PYOPENSSL: |
no test coverage detected