code: def test_recover_during_update(mocked_application_state_manager):
      """Test that application and deployment states are recovered if
      controller crashed in the middle of a redeploy.
  
      Target state is checkpointed in the application state manager,
      but not yet the deployment state manager when the controller crashes
      Then the deployment state manager should recover an old version of
      the deployment during initial recovery, but the application state
      manager should eventually reconcile this.
      """
      (
          app_state_manager,
          deployment_state_manager,
          kv_store,
      ) = mocked_application_state_manager
      deployment_id = DeploymentID("d1", "test_app")
      app_name = "test_app"
  
      # DEPLOY application with deployment "d1"
      params = deployment_params("d1")
      app_state_manager.apply_deployment_args(app_name, [params])
      app_state = app_state_manager._application_states[app_name]
      assert app_state.status == ApplicationStatus.DEPLOYING
  
      # Once deployment is healthy, app should be running
      app_state_manager.update()
      assert deployment_state_manager.get_deployment(deployment_id)
      deployment_state_manager.set_deployment_healthy(deployment_id)
      app_state_manager.update()
      assert app_state.status == ApplicationStatus.RUNNING
  
      # Deploy new version of "d1" (this auto generates new random version)
      params2 = deployment_params("d1")
      app_state_manager.apply_deployment_args(app_name, [params2])
      assert app_state.status == ApplicationStatus.DEPLOYING
  
      # Before application state manager could propagate new version to
      # deployment state manager, controller crashes.
      # Create new deployment state manager. It should recover the old
      # version of the deployment from the kv store
      new_deployment_state_manager = MockDeploymentStateManager(
          kv_store
      )
      dr_version = new_deployment_state_manager.deployment_infos[
          deployment_id
      ].version
  
      # Create new application state manager, and it should call _recover_from_checkpoint
      new_app_state_manager = ApplicationStateManager(
          new_deployment_state_manager, MockEndpointState(), kv_store
      )
      app_state = new_app_state_manager._application_states[app_name]
      ar_version = app_state._target_state.deployment_infos[
          "d1"
      ].version
      assert app_state.status == ApplicationStatus.DEPLOYING
      assert ar_version != dr_version
  
      new_app_state_manager.update()
      assert (
          new_deployment_state_manager.deployment_infos[
              deployment_id
          ].version
          == ar_version
      )
      assert app_state.status == ApplicationStatus.DEPLOYING
  
      new_deployment_state_manager.set_deployment_healthy(deployment_id)
      new_app_state_manager.update()
      assert app_state.status == ApplicationStatus.RUNNING
  
imports: import sys
  from typing import Dict, List, Tuple
  from unittest.mock import Mock, PropertyMock, patch
  import pytest
  from ray.exceptions import RayTaskError
  from ray.serve._private.application_state import ApplicationState, ApplicationStateManager, override_deployment_info
  from ray.serve._private.common import ApplicationStatus, DeploymentID, DeploymentStatus, DeploymentStatusInfo, DeploymentStatusTrigger
  from ray.serve._private.config import DeploymentConfig, ReplicaConfig
  from ray.serve._private.deploy_utils import deploy_args_to_deployment_info
  from ray.serve._private.deployment_info import DeploymentInfo
  from ray.serve._private.test_utils import MockKVStore
  from ray.serve._private.utils import get_random_string
  from ray.serve.exceptions import RayServeException
  from ray.serve.schema import DeploymentSchema, ServeApplicationSchema
label_desc: Use image node_id 15 for calling a global function with example usage: test_recover_during_update(mocked_application_state_manager) without return types