Hi all, I'm trying to implement some tests for my code but am having difficulty mocking input assets. I'm sure there's a way to do this, but mocking the asset that is being used in

materialize_to_memory

results in a

CheckError

.

dagster._check.CheckError: Member of sequence mismatches type. Expected (<class 'dagster._core.definitions.assets.AssetsDefinition'>, <class 'dagster._core.definitions.source_asset.SourceAsset'>). Got <MagicMock name='run_first' id='...'> of type <class 'unittest.mock.MagicMock'>.

Does anyone have any suggestions for me? Thanks in advance! Full code below:

import unittest
from unittest import mock
from dagster import asset, resource, materialize_to_memory


@resource
def my_first_resource():
    return "1st resource"


@resource
def my_second_resource():
    return "2nd resource"


@asset(required_resource_keys={"rsc"})
def run_first(context):
    <http://context.log.info|context.log.info>(f"Running 'run_first' with resource '{context.resources.rsc}'")
    return True


@asset(required_resource_keys={"rsc2"})
def run_second(context, run_first):
    <http://context.log.info|context.log.info>(f"Running 'run_second' with resource '{context.resources.rsc2}'")
    if run_first:
        return "run second succeeds"
    else:
        return "run second fails"


class TestAssets(unittest.TestCase):
    def test_without_mocking(self):
        """This works fine."""
        results = materialize_to_memory(
            [run_first, run_second],
            resources={"rsc": my_first_resource, "rsc2": my_second_resource},
        )
        self.assertEqual(results.output_for_node("run_second"), "run second succeeds")

    @mock.patch("ignore2.run_first")
    def test_with_mock(self, run_first_mock):
        """Raises a dagster._check.CheckError: Member of sequence mismatches type.
        Expected (<class 'dagster._core.definitions.assets.AssetsDefinition'>,
        <class 'dagster._core.definitions.source_asset.SourceAsset'>). Got <MagicMock
        name='run_first' id='...'> of type <class 'unittest.mock.MagicMock'>.
        """
        run_first_mock.return_value = True
        results = materialize_to_memory(
            [run_first, run_second],
            resources={"rsc": my_first_resource, "rsc2": my_second_resource},
        )
        self.assertEqual(results.output_for_node("run_second"), "run second succeeds")


if __name__ == "__main__":
    unittest.main()

mocking isn’t working bc dagster does some under-the-hood runtime type checking on provided inputs to user-facing functions; you could make this work by providing a

spec

to the mock, but I actually recommend just making a little utility function for creating simple value-generating assets that you can re-use for your tests like so:

def create_mock_asset(val, key):
    @asset(name=key)
    def mock_asset():
        return val

    return mock_asset

then you would replace

run_first

with a call to

create_mock_asset(True, "run_first")

Clever solution! Thanks so much for your help!