[DA][testing] RFC: Create AutomationConditionTester class

[OwenKephart]

Summary & Motivation

This is an initial forray into exposing a testing API for AutomationConditions. It's quite minimal at the moment, but contains the necessary functions for a user to test how either new materializations or the passage of time might impact the decisions made. It's inspired by the testing framework we use internally, but with a bit less magic.

One constraint here was that any return value from this must only use public APIs, meaning no AssetKeyPartitionKey or AssetSubset, hence the somewhat weird shape of the AutomationConditionTesterResult object, and the fact that we just log the evaluation results rather than return them outright.

This contains a single test at the moment, roughly indicating how a user might interface with this, but just wanted to put this out to assess the vibes on this type of pattern.

How I Tested These Changes

[OwenKephart]

• [DA][testing] RFC: Create AutomationConditionTester class #22292 👈
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @OwenKephart and the rest of your teammates on Graphite

[schrockn]

I think this is too complicated for a public API. The primary interface should be a simple function with this as an opt-in advanced API.

[OwenKephart]

@schrockn hm so the alternative testing path with a single function would look like:

from dagster import execute_automation_tick, AutomationTickResult

def my_test():

    instance=DagsterInstance.ephemeral()

    result = execute_automation_tick(defs=defs, current_time=..., instance=instance)

    assert result.num_requested == 1

    result = execute_automation_tick(defs=defs, current_time=..., cursor=result.cursor, instance=instance)

    assert result.num_requested == 0

This isn't the end of the world but is a bit more verbose because you have to explicitly shuttle the instance and cursor between ticks.

The main thing I was trying to accomplish with the class was to avoid that, but there's definitely no harm in starting lightweight.

[OwenKephart]

@schrockn updated the PR to move the class thing out of the public API and add an evaluate_automation_conditions method.

[schrockn]

Yeah I think the function version is simpler and easier to understand. If we want to handle instance and context management I think we could do something like

with automation_testing(context=context, instance=instance):
    result = execute_automation_tick(defs=defs, current_time=...)

or

with automation_state(context=context, instance=instance, current_time=...) as state:
    result = state.execute_automation_tick(defs=defs)

and it just is a partial underneath the hood rather than a stateful builder thing.

[schrockn]

I think we should figure out a way to allow for passing a list of AssetsDefinitions

[OwenKephart]

@schrockn gotcha -- removed the class-based thing, made defs accept a Sequence[AssetsDefinition], and added a docstring

[schrockn]

Definitely should be a goal to have this return a range representation soon. cc: @smackesey

[schrockn]

cool thanks