Skip to main content

Testing configurable resources

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

from dagster import ConfigurableResource

class MyResource(ConfigurableResource):
value: str

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

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

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

from dagster import ConfigurableResource

class StringHolderResource(ConfigurableResource):
value: str

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

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:

from dagster import (
ConfigurableResource,
build_init_resource_context,
DagsterInstance,
)
from typing import Optional

class MyContextResource(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()

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