Hi All. Is there a proper way to test an asset sensor by any chance?

There’s a CLI endpoint and some utils for writing unit tests: https://docs.dagster.io/concepts/partitions-schedules-sensors/sensors#testing-sensors

Thank you. My specific problem is with asset sensor since in their case I need to create a mock event somehow

Got it. I wonder if you could run a job first to generate a fake materialization event

Yep, You can do something like this:

@asset
def my_asset():
    return 1


@asset_sensor(asset_key=AssetKey("my_asset"), job=my_job)
def sensor_to_test(context, asset_event):
    yield RunRequest(
        run_key=context.cursor,
        run_config={
            "ops": {
                "read_materialization": {
                    "config": {
                        "asset_key": asset_event.dagster_event.asset_key.path,
                    }
                }
            }
        },
    )


def test_sensor():
    with instance_for_test() as instance:
        build_assets_job("yields_my_asset", [my_asset]).execute_in_process(instance=instance)
        for run_request in sensor_to_test(build_sensor_context(instance=instance)):
            assert run_request.run_key

Thank you, I'll try it out