(self, opts)
| 907 | return corpus_encrypted |
| 908 | |
| 909 | def _test_corpus(self, opts): |
| 910 | # Drop and create the collection 'db.coll' with jsonSchema. |
| 911 | coll = create_with_schema( |
| 912 | self.client.db.coll, self.fix_up_schema(json_data("corpus", "corpus-schema.json")) |
| 913 | ) |
| 914 | self.addCleanup(coll.drop) |
| 915 | |
| 916 | vault = create_key_vault( |
| 917 | self.client.keyvault.datakeys, |
| 918 | json_data("corpus", "corpus-key-local.json"), |
| 919 | json_data("corpus", "corpus-key-aws.json"), |
| 920 | json_data("corpus", "corpus-key-azure.json"), |
| 921 | json_data("corpus", "corpus-key-gcp.json"), |
| 922 | json_data("corpus", "corpus-key-kmip.json"), |
| 923 | ) |
| 924 | self.addCleanup(vault.drop) |
| 925 | |
| 926 | client_encrypted = self.rs_or_single_client(auto_encryption_opts=opts) |
| 927 | |
| 928 | client_encryption = self.create_client_encryption( |
| 929 | self.kms_providers(), |
| 930 | "keyvault.datakeys", |
| 931 | client_context.client, |
| 932 | OPTS, |
| 933 | kms_tls_options=DEFAULT_KMS_TLS, |
| 934 | ) |
| 935 | |
| 936 | corpus = self.fix_up_curpus(json_data("corpus", "corpus.json")) |
| 937 | corpus_copied: SON = SON() |
| 938 | for key, value in corpus.items(): |
| 939 | corpus_copied[key] = copy.deepcopy(value) |
| 940 | if key in ( |
| 941 | "_id", |
| 942 | "altname_aws", |
| 943 | "altname_azure", |
| 944 | "altname_gcp", |
| 945 | "altname_local", |
| 946 | "altname_kmip", |
| 947 | ): |
| 948 | continue |
| 949 | if value["method"] == "auto": |
| 950 | continue |
| 951 | if value["method"] == "explicit": |
| 952 | identifier = value["identifier"] |
| 953 | self.assertIn(identifier, ("id", "altname")) |
| 954 | kms = value["kms"] |
| 955 | self.assertIn(kms, ("local", "aws", "azure", "gcp", "kmip")) |
| 956 | if identifier == "id": |
| 957 | if kms == "local": |
| 958 | kwargs = {"key_id": LOCAL_KEY_ID} |
| 959 | elif kms == "aws": |
| 960 | kwargs = {"key_id": AWS_KEY_ID} |
| 961 | elif kms == "azure": |
| 962 | kwargs = {"key_id": AZURE_KEY_ID} |
| 963 | elif kms == "gcp": |
| 964 | kwargs = {"key_id": GCP_KEY_ID} |
| 965 | else: |
| 966 | kwargs = {"key_id": KMIP_KEY_ID} |
no test coverage detected