|
| 1 | +import os |
1 | 2 | from io import StringIO
|
2 | 3 |
|
| 4 | +import pymongo |
3 | 5 | from bson import json_util
|
4 | 6 | from django.core.management import call_command
|
| 7 | +from django.db import connections |
5 | 8 | from django.test import TransactionTestCase, modify_settings, override_settings
|
| 9 | +from pymongo.encryption import AutoEncryptionOpts |
6 | 10 |
|
7 | 11 | from .routers import TestEncryptedRouter
|
8 | 12 |
|
@@ -83,5 +87,20 @@ def test_create_new_keys(self):
|
83 | 87 | command_output["encryption__patientrecord"],
|
84 | 88 | )
|
85 | 89 |
|
86 |
| - # TODO: Create a new connection to verify that the keys can be used |
87 |
| - # in a client-side configuration to migrate the encrypted fields. |
| 90 | + # Create a new connection to verify that the keys can be used in a |
| 91 | + # client-side configuration to migrate the encrypted fields. |
| 92 | + conn_params = connections["encrypted"].get_connection_params() |
| 93 | + auto_encryption_opts = AutoEncryptionOpts( |
| 94 | + key_vault_namespace="encryption.__keyvault", |
| 95 | + kms_providers={"local": {"key": os.urandom(96)}}, |
| 96 | + encrypted_fields_map=command_output, |
| 97 | + ) |
| 98 | + if conn_params.pop("auto_encryption_opts", False): |
| 99 | + # Call MongoClient instead of get_new_connection because |
| 100 | + # get_new_connection will return the encrypted connection from the |
| 101 | + # connection pool. |
| 102 | + with pymongo.MongoClient(**conn_params, auto_encryption_opts=auto_encryption_opts): |
| 103 | + call_command("migrate", "--database", "encrypted", verbosity=0) |
| 104 | + |
| 105 | + # TODO: Check the key vault to ensure that the keys created by |
| 106 | + # `showencryptedfieldsmap --create-new-keys` are in the key vault. |
0 commit comments