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

Alex Berndt

03/06/2024, 2:54 PM
How can I test a job which consists of ops using the new pydantic-based config schemas?
s

Scout

03/06/2024, 2:54 PM
To test a job that consists of ops using the new Pydantic-based config schemas, you would follow a similar approach to testing any Dagster job, but with the added convenience that Pydantic-based config schemas provide more straightforward and type-safe access to configuration values. Here's a step-by-step guide based on the documentation: 1. Define Your Config Schema: Use Pydantic to define a strongly-typed config schema for your assets or ops. This allows you to access configuration values directly as attributes of the config object, rather than using a dictionary.
Copy code
python    from pydantic import BaseModel     class MyOpConfig(BaseModel):        a_string: str        an_int: int
2. Define Your Ops: When defining your ops, you can now use the Pydantic config schema directly as a parameter to the op function.
Copy code
python    @op    def my_op(config: MyOpConfig):        # Access config directly        assert config.a_string == "expected_string"        assert config.an_int == 42
3. Define Your Job: Create a job that includes the ops you've defined.
Copy code
python    @job    def my_job():        my_op()
4. Write Tests: Write unit tests for your ops and job. You can directly construct instances of your Pydantic config models with the desired test values and pass them to your ops or job when invoking them in tests.
Copy code
python    def test_my_op():        test_config = MyOpConfig(a_string="expected_string", an_int=42)        my_op(config=test_config)
5. Execute Job in Tests: Use the
execute_in_process
method to execute the job within your test environment. You can pass the Pydantic config object directly to the
run_config
parameter.
Copy code
python    def test_my_job():        test_config = {"ops": {"my_op": MyOpConfig(a_string="expected_string", an_int=42)}}        result = my_job.execute_in_process(run_config=test_config)        assert result.success
This approach leverages the type safety and ease of use provided by Pydantic models, making your tests more readable and maintainable. Remember that the exact syntax for invoking ops and jobs in tests may vary depending on the version of Dagster you are using and the specific features of the Pydantic-based config system that you are utilizing. Please note that the code examples provided are based on the documentation's description of using Pydantic for config schemas and are meant to illustrate the testing process. You should adapt these examples to fit the specifics of your job and ops.
a

Alex Berndt

03/06/2024, 2:57 PM
Copy code
def test_my_job():
       test_config = {"ops": {"my_op": MyOpConfig(a_string="expected_string", an_int=42)}}
       result = my_job.execute_in_process(run_config=test_config)
       assert result.success
This yields an error saying
Value must be a dict
. Please correct the example.
@Scout
Copy code
def test_my_job():
       test_config = {"ops": {"my_op": MyOpConfig(a_string="expected_string", an_int=42)}}
       result = my_job.execute_in_process(run_config=test_config)
       assert result.success
This yields an error saying
Value must be a dict
. Please correct the example.
s

Scout

03/06/2024, 2:58 PM
Apologies for the confusion. When you're using Pydantic models for configuration, you need to convert the Pydantic model to a dictionary before passing it to the
run_config
parameter. You can use the Pydantic model's
.dict()
method to achieve this. Here's the corrected example:
Copy code
python def test_my_job():     test_config = {         "ops": {             "my_op": {                 "config": MyOpConfig(a_string="expected_string", an_int=42).dict()             }         }     }     result = my_job.execute_in_process(run_config=test_config)     assert result.success
In this example,
MyOpConfig(a_string="expected_string", an_int=42).dict()
creates an instance of the
MyOpConfig
Pydantic model with the specified values and then converts it to a dictionary, which is the expected format for the
run_config
parameter when executing a job in process.
3 Views