[Cosmos DB] az cosmosdb restore: Fix cross-region restore by preserving source region in top-level location

src/azure-cli/HISTORY.rst

@@ -133,6 +133,7 @@ Release History
 
 * `az cosmosdb update`: Add support for Microsoft Fabric workspace resource IDs in `--network-acl-bypass-resource-ids` (#32797)
 * Fix #32608: `az cosmosdb restore`: Fix "Database Account does not exist" error during polling (#32752)
+* `az cosmosdb restore`: Preserve source region in top-level location for cross-region restore (#33274)
 
 **Maps**
 

src/azure-cli/azure/cli/command_modules/cosmosdb/custom.py

@@ -273,10 +273,21 @@ def _create_database_account(client,
         locations = []
         locations.append(Location(location_name=arm_location, failover_priority=0, is_zone_redundant=False))
 
-    for loc in locations:
-        if loc.failover_priority == 0:
-            arm_location = loc.location_name
-            break
+    # For cross-region restore (CRR), the caller intentionally passes
+    # arm_location set to the SOURCE region while locations[priority=0] is
+    # the TARGET region. The Cosmos ARM contract for restore requires the
+    # top-level `location` on DatabaseAccountCreateUpdateParameters to match
+    # the `restoreSource` URI region (the source). Overwriting arm_location
+    # with the priority-0 target here causes the backend to reject the
+    # request with "Location provided in 'restoreSource' does not match the
+    # location of the request" (BadRequest). Skip this normalization for
+    # restore requests; for regular create the loop preserves existing
+    # behavior of aligning arm_location with the priority-0 location.
+    if not is_restore_request:
+        for loc in locations:
+            if loc.failover_priority == 0:
+                arm_location = loc.location_name
+                break
 
     managed_service_identity = None
     SYSTEM_ID = '[system]'

src/azure-cli/azure/cli/command_modules/cosmosdb/tests/latest/test_cosmosdb_backuprestore_scenario.py

@@ -682,4 +682,81 @@ def test_normal_create_success(self):
         # 3. client.get should NOT be called since result() succeeded
         client.get.assert_not_called()
         # 4. Result matches
-        self.assertEqual(result, mock_created_account)
+        self.assertEqual(result, mock_created_account)
+
+    def test_restore_preserves_source_arm_location_for_cross_region(self):
+        # Regression test for cross-region restore: ensure the
+        # `failover_priority == 0` loop does NOT overwrite arm_location
+        # (source region) with the priority-0 location (target region)
+        # when is_restore_request=True. Backend rejects the request as
+        # BadRequest if top-level `location` does not match the
+        # `restoreSource` URI region.
+        from azure.cli.command_modules.cosmosdb.custom import _create_database_account
+
+        client = mock.MagicMock()
+        poller = mock.MagicMock()
+        mock_account = mock.MagicMock()
+        mock_account.provisioning_state = "Succeeded"
+        poller.result.return_value = mock_account
+        client.begin_create_or_update.return_value = poller
+
+        # Mock a Location object with a real failover_priority value (not a
+        # MagicMock) so the gate's truthiness check behaves predictably.
+        target_location = mock.MagicMock()
+        target_location.location_name = "westus2"
+        target_location.failover_priority = 0
+
+        with mock.patch(
+            'azure.cli.command_modules.cosmosdb.custom.DatabaseAccountCreateUpdateParameters'
+        ) as mock_params:
+            _create_database_account(
+                client=client,
+                resource_group_name="rg",
+                account_name="myaccount",
+                locations=[target_location],
+                is_restore_request=True,
+                arm_location="eastus2",
+                restore_source="/subscriptions/sub/providers/Microsoft.DocumentDB/locations/eastus2/restorableDatabaseAccounts/source-id",
+                restore_timestamp="2026-01-01T00:00:00+00:00"
+            )
+
+            mock_params.assert_called_once()
+            kwargs = mock_params.call_args.kwargs
+            # Source region (eastus2) must be preserved; loop must NOT
+            # overwrite it with the priority-0 target (westus2).
+            self.assertEqual(kwargs.get('location'), "eastus2")
+            self.assertEqual(kwargs.get('locations'), [target_location])
+
+    def test_normal_create_aligns_arm_location_with_priority_zero(self):
+        # Control test: for non-restore creates, the loop preserves
+        # existing behavior of aligning arm_location with the
+        # priority-0 location.
+        from azure.cli.command_modules.cosmosdb.custom import _create_database_account
+
+        client = mock.MagicMock()
+        poller = mock.MagicMock()
+        mock_account = mock.MagicMock()
+        mock_account.provisioning_state = "Succeeded"
+        poller.result.return_value = mock_account
+        client.begin_create_or_update.return_value = poller
+
+        primary_location = mock.MagicMock()
+        primary_location.location_name = "westus2"
+        primary_location.failover_priority = 0
+
+        with mock.patch(
+            'azure.cli.command_modules.cosmosdb.custom.DatabaseAccountCreateUpdateParameters'
+        ) as mock_params:
+            _create_database_account(
+                client=client,
+                resource_group_name="rg",
+                account_name="myaccount",
+                locations=[primary_location],
+                is_restore_request=False,
+                arm_location="eastus2"
+            )
+
+            mock_params.assert_called_once()
+            kwargs = mock_params.call_args.kwargs
+            # Non-restore path: priority-0 location overrides arm_location.
+            self.assertEqual(kwargs.get('location'), "westus2")