Testing configurable resources

tip

You can scaffold resources from the command line by running dg scaffold defs dagster.resources <path/to/resources_file.py>. For more information, see the dg CLI docs.

You can test the initialization of a ConfigurableResource by constructing it manually. In most cases, the resource can be constructed directly:

src/<project_name>/defs/resources.py

import dagster as dg

class MyResource(dg.ConfigurableResource):
    value: str

    def get_value(self) -> str:
        return self.value

tests/test_resource.py

def test_my_resource():
    assert MyResource(value="foo").get_value() == "foo"

If the resource requires other resources, you can pass them as constructor arguments:

src/<project_name>/defs/resources.py

import dagster as dg

class StringHolderResource(dg.ConfigurableResource):
    value: str

class MyResourceRequiresAnother(dg.ConfigurableResource):
    foo: StringHolderResource
    bar: str

tests/test_resource.py

def test_my_resource_with_nesting():
    string_holder = StringHolderResource(value="foo")
    resource = MyResourceRequiresAnother(foo=string_holder, bar="bar")
    assert resource.foo.value == "foo"
    assert resource.bar == "bar"

Testing with resource context

In the case that a resource uses the resource initialization context, you can use the build_init_resource_context utility alongside the with_init_resource_context helper on the resource class:

src/<project_name>/defs/resources.py

import dagster as dg

from typing import Optional

class MyContextResource(dg.ConfigurableResource[GitHub]):
    base_path: Optional[str] = None

    def effective_base_path(self) -> str:
        if self.base_path:
            return self.base_path
        instance = self.get_resource_context().instance
        assert instance
        return instance.storage_directory()

tests/test_resource.py

def test_my_context_resource():
    with dg.DagsterInstance.ephemeral() as instance:
        context = dg.build_init_resource_context(instance=instance)
        assert (
            MyContextResource(base_path=None)
            .with_resource_context(context)
            .effective_base_path()
            == instance.storage_directory()
        )