Fix path attribute in encrypted fields schema

- Recursive calls to _get_encrypted_fields now updates the path to
  include the parent_model

Author: aclark4life

django_mongodb_backend/schema.py

@@ -488,17 +488,21 @@ def _create_collection(self, model):
             # Unencrypted path
             db.create_collection(db_table)
 
-    def _get_encrypted_fields(self, model, create_data_keys=False, key_alt_name=None):
+    def _get_encrypted_fields(
+        self, model, create_data_keys=False, key_alt_name=None, parent_model=None
+    ):
         """
         Recursively collect encryption schema data for only encrypted fields in a model.
         Returns None if no encrypted fields are found anywhere in the model hierarchy.
 
-        key_alt_name is the base path for this level, typically model._meta.db_table.
+        key_alt_name is the base path used for keyAltNames.
+        parent_model is the dot-notated path inside the document for schema mapping.
         """
         connection = self.connection
         client = connection.connection
         fields = model._meta.fields
         key_alt_name = key_alt_name or model._meta.db_table
+        parent_model = parent_model or ""
 
         options = client._options
         auto_encryption_opts = getattr(options, "auto_encryption_opts", None)
@@ -515,49 +519,51 @@ def _get_encrypted_fields(self, model, create_data_keys=False, key_alt_name=None
         field_list = []
 
         for field in fields:
-            new_path = f"{key_alt_name}.{field.column}"
+            new_key_alt_name = f"{key_alt_name}.{field.column}"
+            new_parent_model = f"{parent_model}.{field.column}" if parent_model else field.column
 
             # --- EmbeddedModelField ---
             if isinstance(field, EmbeddedModelField):
                 if getattr(field, "encrypted", False):
-                    # Entire sub-object is encrypted
+                    # Entire sub-object encrypted
                     if create_data_keys:
                         if not client_encryption:
                             raise ImproperlyConfigured("client_encryption is not configured.")
                         data_key = client_encryption.create_data_key(
                             kms_provider=kms_provider,
                             master_key=master_key,
-                            key_alt_names=[new_path],
+                            key_alt_names=[new_key_alt_name],
                         )
                     else:
                         if key_vault_collection is None:
                             raise ImproperlyConfigured(
-                                f"Encrypted field {new_path} detected but no key vault configured"
+                                f"Encrypted field {new_key_alt_name} detected "
+                                "but no key vault configured"
                             )
-                        key_doc = key_vault_collection.find_one({"keyAltNames": new_path})
+                        key_doc = key_vault_collection.find_one({"keyAltNames": new_key_alt_name})
                         if not key_doc:
                             raise ValueError(
-                                f"No key found in keyvault for keyAltName={new_path}. "
+                                f"No key found in keyvault for keyAltName={new_key_alt_name}. "
                                 "Run with '--create-data-keys' to create missing keys."
                             )
                         data_key = key_doc["_id"]
 
                     field_dict = {
                         "bsonType": "object",
-                        "path": field.column,
+                        "path": new_parent_model,
                         "keyId": data_key,
                     }
                     if getattr(field, "queries", False):
                         field_dict["queries"] = field.queries
 
                     field_list.append(field_dict)
                 else:
-                    # Not encrypting whole object — recurse first then
-                    # conditionally extend field list
+                    # Recurse into embedded model
                     embedded_result = self._get_encrypted_fields(
                         field.embedded_model,
                         create_data_keys=create_data_keys,
-                        key_alt_name=new_path,
+                        key_alt_name=new_key_alt_name,
+                        parent_model=new_parent_model,
                     )
                     if embedded_result and embedded_result.get("fields"):
                         field_list.extend(embedded_result["fields"])
@@ -571,24 +577,25 @@ def _get_encrypted_fields(self, model, create_data_keys=False, key_alt_name=None
                     data_key = client_encryption.create_data_key(
                         kms_provider=kms_provider,
                         master_key=master_key,
-                        key_alt_names=[new_path],
+                        key_alt_names=[new_key_alt_name],
                     )
                 else:
                     if key_vault_collection is None:
                         raise ImproperlyConfigured(
-                            f"Encrypted field {new_path} detected but no key vault configured"
+                            f"Encrypted field {new_key_alt_name} detected "
+                            "but no key vault configured"
                         )
-                    key_doc = key_vault_collection.find_one({"keyAltNames": new_path})
+                    key_doc = key_vault_collection.find_one({"keyAltNames": new_key_alt_name})
                     if not key_doc:
                         raise ValueError(
-                            f"No key found in keyvault for keyAltName={new_path}. "
+                            f"No key found in keyvault for keyAltName={new_key_alt_name}. "
                             "Run with '--create-data-keys' to create missing keys."
                         )
                     data_key = key_doc["_id"]
 
                 field_dict = {
                     "bsonType": field.db_type(connection),
-                    "path": field.column,
+                    "path": new_parent_model,
                     "keyId": data_key,
                 }
                 if getattr(field, "queries", False):

tests/encryption_/test_management.py

@@ -13,8 +13,12 @@ class CommandTests(EncryptionTestCase):
     expected_maps = {
         "encryption__patient": {
             "fields": [
-                {"bsonType": "string", "path": "ssn", "queries": {"queryType": "equality"}},
-                {"bsonType": "object", "path": "billing"},
+                {
+                    "bsonType": "string",
+                    "path": "patient_record.ssn",
+                    "queries": {"queryType": "equality"},
+                },
+                {"bsonType": "object", "path": "patient_record.billing"},
             ]
         },
         # Equality-queryable fields

tests/encryption_/test_schema.py

@@ -10,8 +10,12 @@ class SchemaTests(EncryptionTestCase):
     expected_map = {
         "Patient": {
             "fields": [
-                {"bsonType": "string", "path": "ssn", "queries": {"queryType": "equality"}},
-                {"bsonType": "object", "path": "billing"},
+                {
+                    "bsonType": "string",
+                    "path": "patient_record.ssn",
+                    "queries": {"queryType": "equality"},
+                },
+                {"bsonType": "object", "path": "patient_record.billing"},
             ]
         },
         "EncryptedBinaryTest": {