https://dagster.io/ logo
#ask-ai
Title
# ask-ai
j

James O'Toole

02/08/2024, 9:16 PM
@Scout if i'm using
build_sensor_context
in a test, how do i use it along with a mock
DagsterRun
instance?
s

Scout

02/08/2024, 9:16 PM
When using
build_sensor_context
in a test, you can create a mock
DagsterRun
instance and pass it to the context builder function to simulate a run context for your sensor. However, the documentation does not provide a direct example of using
build_sensor_context
with a mock
DagsterRun
instance. Instead, it provides guidance on how to build a sensor execution context using the provided parameters, which can be used to invoke a sensor definition. Here's a general approach based on the documentation: 1. Create an ephemeral Dagster instance using
DagsterInstance.ephemeral()
. 2. Execute a job or an operation to generate a
DagsterRun
object. 3. Use the
DagsterRun
object to build the sensor context. Here is a code snippet from the documentation that demonstrates how to build the context for a run status sensor, which you can adapt for your use case with a mock `DagsterRun`:
Copy code
python from dagster import DagsterInstance, build_run_status_sensor_context  # Create an ephemeral Dagster instance instance = DagsterInstance.ephemeral()  # Execute a job to generate a DagsterRun object (this is where you would use your mock instead) result = my_job.execute_in_process(instance=instance) dagster_run = result.dagster_run  # Create the sensor context sensor_context = build_run_status_sensor_context(     sensor_name="my_sensor",     dagster_instance=instance,     dagster_run=dagster_run,     dagster_event=result.get_job_success_event(),  # or get_job_failure_event() )  # Now you can invoke your sensor with the created context my_sensor(sensor_context)
In this example,
my_job
would be a job that you expect to succeed and generate a
DagsterRun
object. If you want to use a mock
DagsterRun
, you would replace the execution of
my_job
with the creation of a mock
DagsterRun
object that suits your testing needs. Remember that the
build_sensor_context
function is used to build a context for sensor invocation, and it can accept parameters such as
instance
,
cursor
,
repository_name
,
repository_def
,
resources
, and
definitions
. If you need to mock the
DagsterRun
for an asset sensor, you would need to provide an instance parameter when building your context, as asset sensors rely on querying for new events from the
DagsterInstance
. Please adapt the above example to fit your specific testing scenario, and if you need to mock specific behaviors or attributes of the
DagsterRun
, you can do so using standard Python mocking libraries like
unittest.mock
.
j

James O'Toole

02/09/2024, 3:28 PM
@Scout here's an excerpt from my current test setup:
Copy code
self.run_id = "abc123"
        self.tags = {
            "job_completion_callback_url": "<https://www.foo.com>",
            "job_completion_callback_auth": "auth_token"
        }
        self.mock_dagster_instance = DagsterInstance.ephemeral()
        self.job_name = "some_job"
        self.mock_dagster_run = DagsterRun(run_id=self.run_id, tags=self.tags, job_name=self.job_name)

        result = self.job_name.execute_in_process(instance=self.mock_dagster_instance)
        dagster_run = result.mock_dagster_run

        # Create the sensor context
        self.sensor_context = build_run_status_sensor_context(
            sensor_name="my_sensor",
            dagster_instance=self.mock_dagster_instance,
            dagster_run=dagster_run,
            dagster_event=result.get_job_success_event(),  # or get_job_failure_event()
        )
could you help me fix the following error i get when running my test suite?
Copy code
AttributeError: 'str' object has no attribute 'execute_in_process'
s

Scout

02/09/2024, 3:28 PM
The error you're encountering,
AttributeError: 'str' object has no attribute 'execute_in_process'
, is occurring because
self.job_name
is a string and not a job object. The
execute_in_process
method is a method of a job object, not a string. To fix this error, you need to ensure that
self.job_name
is assigned to an actual job object that you've defined using the
@job
decorator. Here's an example of how you might define a job:
Copy code
python from dagster import job, op  @op def my_op():     # Your operation logic here     pass  @job def some_job():     my_op()
Once you have defined your job, you can reference it when setting `self.job_name`:
Copy code
python self.job_name = some_job
Now, when you call
execute_in_process
on
self.job_name
, it will be calling the method on the job object, not a string:
Copy code
python result = self.job_name.execute_in_process(instance=self.mock_dagster_instance)
Make sure to replace
some_job
with the actual job object you have defined in your code. Here's how your test setup should look after the correction:
Copy code
python from dagster import DagsterInstance, build_run_status_sensor_context, DagsterRun  # Assuming some_job is a job object you have defined elsewhere in your code self.run_id = "abc123" self.tags = {     "job_completion_callback_url": "<https://www.foo.com>",     "job_completion_callback_auth": "auth_token" } self.mock_dagster_instance = DagsterInstance.ephemeral() self.job_name = some_job  # Replace with your actual job object self.mock_dagster_run = DagsterRun(run_id=self.run_id, tags=self.tags, job_name=self.job_name.name)  result = self.job_name.execute_in_process(instance=self.mock_dagster_instance) dagster_run = result.dagster_run  # Use the result's dagster_run, not mock_dagster_run  # Create the sensor context self.sensor_context = build_run_status_sensor_context(     sensor_name="my_sensor",     dagster_instance=self.mock_dagster_instance,     dagster_run=dagster_run,     dagster_event=result.get_job_success_event(),  # or get_job_failure_event() )
Make sure that
some_job
is the job you want to test, and replace
my_sensor
with the actual sensor name you are testing. This should resolve the
AttributeError
you're encountering.
j

James O'Toole

02/09/2024, 3:31 PM
@Scout i don't use the @job decorator when defining my job; i instead do so as follows:
Copy code
all_assets = load_assets_from_modules([assets], group_name=GROUP_NAME)

materialize_all_assets_job = define_asset_job(
    name="some_job",
    selection=AssetSelection.all(),
    # auto-terminate run after 20 minutes; we can toggle this up or down as needed
    tags={MAX_RUNTIME_SECONDS_TAG: 1200},
)

defs = Definitions(
    assets=all_assets, 
    jobs=[materialize_all_assets_job], 
    sensors=[success_callback, failure_callback]
)
how should i therefore fix my test setup?
s

Scout

02/09/2024, 3:31 PM
Given that you're using
define_asset_job
to create your job, you'll need to resolve the job from the
Definitions
object before you can execute it in process. The
define_asset_job
function returns an
UnresolvedAssetJobDefinition
, which needs to be resolved with the actual asset definitions before it can be executed. Here's how you can adjust your test setup: 1. Use the
Definitions
object to get the resolved job definition. 2. Execute the resolved job in process. 3. Use the result of the execution to create the sensor context. Here's an example of how you might adjust your test setup:
Copy code
python from dagster import DagsterInstance, build_run_status_sensor_context, DagsterRun, Definitions, define_asset_job, AssetSelection  # Define your assets and job as you have in your actual code all_assets = load_assets_from_modules([assets], group_name=GROUP_NAME)  materialize_all_assets_job = define_asset_job(     name="some_job",     selection=AssetSelection.all(),     tags={MAX_RUNTIME_SECONDS_TAG: 1200}, )  # Create a Definitions object with your assets and job defs = Definitions(     assets=all_assets,      jobs=[materialize_all_assets_job],      sensors=[success_callback, failure_callback] )  # Resolve the job from the Definitions object resolved_job = defs.get_job_def("some_job")  # Create an ephemeral Dagster instance mock_dagster_instance = DagsterInstance.ephemeral()  # Execute the resolved job in process result = resolved_job.execute_in_process(instance=mock_dagster_instance) dagster_run = result.dagster_run  # Create the sensor context sensor_context = build_run_status_sensor_context(     sensor_name="my_sensor",     dagster_instance=mock_dagster_instance,     dagster_run=dagster_run,     dagster_event=result.get_job_success_event(),  # or get_job_failure_event() )
In this setup,
resolved_job
is the job definition that you can execute in process. The
execute_in_process
method will return a result object that contains the
dagster_run
and the events that you can use to create the sensor context. Make sure to replace
success_callback
,
failure_callback
,
assets
, and
GROUP_NAME
with the actual values from your code. This should allow you to execute your job in the test and create the sensor context without encountering the
AttributeError
.
j

James O'Toole

02/09/2024, 3:43 PM
why is this so complicated?