def METHOD_NAME(is_valid, resource_defs):
    @observable_source_asset(
        required_resource_keys={"bar"},
        resource_defs=resource_defs,
    )
    def foo(context) -> DataVersion:
        return DataVersion(f"{context.resources.bar}-alpha")
    instance = DagsterInstance.ephemeral()
    if is_valid:
        observe([foo], instance=instance)
        assert _get_current_data_version(AssetKey("foo"), instance) == DataVersion("bar-alpha")
    else:
        with pytest.raises(
            DagsterInvalidDefinitionError,
            match="resource with key 'bar' required by op 'foo' was not provided",
        ):
            observe([foo], instance=instance)