code: def test_upsert():
      subscriber = DummySubscriber()
  
      storage = InstanceStorage(
          cluster_id="test_cluster",
          storage=InMemoryStorage(),
          status_change_subscriber=subscriber,
      )
      instance1 = create_instance("instance1")
      instance2 = create_instance("instance2")
      instance3 = create_instance("instance3")
  
      assert (True, 1) == storage.batch_upsert_instances(
          [instance1, instance2],
          expected_storage_version=None,
      )
  
      assert subscriber.events == [
          InstanceUpdateEvent(
              instance_id="instance1",
              new_instance_status=Instance.UNKNOWN,
          ),
          InstanceUpdateEvent(
              instance_id="instance2",
              new_instance_status=Instance.UNKNOWN,
          ),
      ]
  
      instance1.version = 1
      instance2.version = 1
      entries, storage_version = storage.get_instances()
  
      assert storage_version == 1
      assert entries == {
          "instance1": instance1,
          "instance2": instance2,
      }
  
      assert (False, 1) == storage.batch_upsert_instances(
          [create_instance("instance1"), create_instance("instance2")],
          expected_storage_version=0,
      )
  
      assert subscriber.events == [
          InstanceUpdateEvent(
              instance_id="instance1",
              new_instance_status=Instance.UNKNOWN,
          ),
          InstanceUpdateEvent(
              instance_id="instance2",
              new_instance_status=Instance.UNKNOWN,
          ),
      ]
  
      instance2.status = Instance.ALLOCATED
      assert (True, 2) == storage.batch_upsert_instances(
          [instance3, instance2],
          expected_storage_version=1,
      )
  
      instance1.version = 1
      instance2.version = 2
      instance3.version = 2
      entries, storage_version = storage.get_instances()
  
      assert storage_version == 2
      assert entries == {
          "instance1": instance1,
          "instance2": instance2,
          "instance3": instance3,
      }
  
      assert subscriber.events == [
          InstanceUpdateEvent(
              instance_id="instance1",
              new_instance_status=Instance.UNKNOWN,
          ),
          InstanceUpdateEvent(
              instance_id="instance2",
              new_instance_status=Instance.UNKNOWN,
          ),
          InstanceUpdateEvent(
              instance_id="instance3",
              new_instance_status=Instance.UNKNOWN,
          ),
          InstanceUpdateEvent(
              instance_id="instance2",
              new_instance_status=Instance.ALLOCATED,
          ),
      ]
  
file_path: ray/python/ray/autoscaler/v2/tests/test_instance_storage.py
imports: import copy
  import os
  import sys
  from unittest import mock
  import pytest
  from ray.autoscaler.v2.instance_manager.instance_storage import InstanceStorage, InstanceUpdatedSubscriber, InstanceUpdateEvent
  from ray.autoscaler.v2.instance_manager.storage import InMemoryStorage
  from ray.core.generated.instance_manager_pb2 import Instance
label_desc: Use image node_id 2 for calling a global function with example usage: test_upsert() without return types