Method setUp

test/test_encryption.py:2273–2322

(self)

    @client_context.require_no_standalone
    @client_context.require_version_min(7, 0, -1)
    def setUp(self):
        super().setUp()
        encrypted_client = self.rs_or_single_client(
            auto_encryption_opts=AutoEncryptionOpts(
                key_vault_namespace="db.keyvault",
                kms_providers={"local": {"key": LOCAL_MASTER_KEY}},
            )
        )
        encrypted_client.drop_database("db")

        key_doc = json_data("etc", "data", "lookup", "key-doc.json")
        create_key_vault(encrypted_client.db.keyvault, key_doc)
        self.addCleanup(client_context.client.drop_database, "db")

        encrypted_client.db.create_collection(
            "csfle",
            validator={"$jsonSchema": json_data("etc", "data", "lookup", "schema-csfle.json")},
        )
        encrypted_client.db.create_collection(
            "csfle2",
            validator={"$jsonSchema": json_data("etc", "data", "lookup", "schema-csfle2.json")},
        )
        encrypted_client.db.create_collection(
            "qe", encryptedFields=json_data("etc", "data", "lookup", "schema-qe.json")
        )
        encrypted_client.db.create_collection(
            "qe2", encryptedFields=json_data("etc", "data", "lookup", "schema-qe2.json")
        )
        encrypted_client.db.create_collection("no_schema")
        encrypted_client.db.create_collection("no_schema2")

        unencrypted_client = self.rs_or_single_client()

        encrypted_client.db.csfle.insert_one({"csfle": "csfle"})
        doc = unencrypted_client.db.csfle.find_one()
        self.assertIsInstance(doc["csfle"], Binary)
        encrypted_client.db.csfle2.insert_one({"csfle2": "csfle2"})
        doc = unencrypted_client.db.csfle2.find_one()
        self.assertIsInstance(doc["csfle2"], Binary)
        encrypted_client.db.qe.insert_one({"qe": "qe"})
        doc = unencrypted_client.db.qe.find_one()
        self.assertIsInstance(doc["qe"], Binary)
        encrypted_client.db.qe2.insert_one({"qe2": "qe2"})
        doc = unencrypted_client.db.qe2.find_one()
        self.assertIsInstance(doc["qe2"], Binary)
        encrypted_client.db.no_schema.insert_one({"no_schema": "no_schema"})
        encrypted_client.db.no_schema2.insert_one({"no_schema2": "no_schema2"})

        encrypted_client.close()
        unencrypted_client.close()

    @client_context.require_version_min(8, 1, -1)
    def test_1_csfle_joins_no_schema(self):