def METHOD_NAME():
    @observable_source_asset(partitions_def=StaticPartitionsDefinition(["apple", "orange", "kiwi"]))
    def foo():
        return DataVersionsByPartition({"apple": "one", "orange": DataVersion("two")})
    result = observe([foo])
    observations = result.asset_observations_for_node("foo")
    assert len(observations) == 2
    observations_by_asset_partition = {
        (observation.asset_key, observation.partition): observation for observation in observations
    }
    assert observations_by_asset_partition.keys() == {(foo.key, "apple"), (foo.key, "orange")}
    assert observations_by_asset_partition[(foo.key, "apple")].tags == {
        "dagster/data_version": "one"
    }
    assert observations_by_asset_partition[(foo.key, "orange")].tags == {
        "dagster/data_version": "two"
    }