[DOC-348] Docs for asset and op unit testing

docs/docs-beta/docs/guides/quality-testing/unit-tests-assets-and-ops.md

@@ -4,4 +4,97 @@ sidebar_position: 30
 sidebar_label: "Unit testing"
 ---
 
-# Unit tests for assets and ops
+Unit testing is an important tool for verifying that a computation works as intended. While this is traditionally difficult in the context of data pipelines, Dagster helps simplify this process by allowing you to directly invoke your underlying transforms with specific input values and mocked resources.
+
+This guide covers how to write unit tests for a variety of different assets and ops.
+
+<details>
+<summary>Prerequisites</summary>
+
+- Familiarity with [Assets](/concepts/assets)
+- Familiarity with [Ops and Jobs](/concepts/ops-jobs)
+</details>
+
+## Testing assets and ops with no arguments
+
+The simplest assets and ops to test are those with no arguments. In these cases, you can directly invoke your definition.
+
+<Tabs>
+  <TabItem value="asset" label="Using assets" default>
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/asset-no-argument.py" language="python"/>
+  </TabItem>
+  <TabItem value="op" label="Using ops">
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/op-no-argument.py" language="python"/>
+  </TabItem>
+</Tabs>
+
+## Testing assets and ops that have upstream dependencies
+
+If your asset or op has an upstream dependency, then you can directly pass a value for that dependency when invoking your definition.
+
+<Tabs>
+  <TabItem value="asset" label="Using assets" default>
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/asset-dependency.py" language="python" />
+  </TabItem>
+  <TabItem value="op" label="Using ops">
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/op-dependency.py" language="python" />
+  </TabItem>
+</Tabs>
+
+## Testing assets and ops with config
+
+If your asset or op uses [config](/todo), you can construct an instance of the required config object and pass it in directly.
+
+<Tabs>
+  <TabItem value="asset" label="Using assets" default>
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/asset-config.py" language="python" />
+  </TabItem>
+  <TabItem value="op" label="Using ops">
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/op-config.py" language="python" />
+  </TabItem>
+</Tabs>
+
+## Testing assets and ops with resources
+
+If your asset or op uses a [resource](/concepts/resources), it can be useful to create a mock instance of that resource to avoid interacting with external services.
+
+<Tabs>
+  <TabItem value="asset" label="Using assets" default>
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/asset-resource.py" language="python" />
+  </TabItem>
+  <TabItem value="op" label="Using ops">
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/op-resource.py" language="python" />
+  </TabItem>
+</Tabs>
+
+## Testing assets and ops with context
+
+If your asset or op uses a context argument, you can use `build_asset_context()` or `build_op_context()` to construct a context object.
+
+<Tabs>
+  <TabItem value="asset" label="Using assets" default>
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/asset-context.py" language="python" />
+  </TabItem>
+  <TabItem value="op" label="Using ops">
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/op-context.py" language="python" />
+  </TabItem>
+</Tabs>
+
+## Testing assets and ops with a combination of parameters
+
+If your asset or op uses a combination of multiple types of parameters, it can be convenient to use keyword arguments, rather than relying on the order of parameters.
+
+<Tabs>
+  <TabItem value="asset" label="Using assets" default>
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/asset-combo.py" language="python" />
+  </TabItem>
+  <TabItem value="op" label="Using ops">
+    <CodeExample filePath="guides/quality-testing/unit-testing-assets-and-ops/op-combo.py" language="python" />
+  </TabItem>
+</Tabs>
+## Next steps
+
+- Learn more about assets in [Understanding Assets](/concepts/assets)
+- Learn more about ops in [Understanding Assets](/concepts/ops-jobs)
+- Learn more about config in [Config](/todo)
+- Learn more about resources in [Resources](/concepts/resources)

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/asset-combo.py

@@ -0,0 +1,25 @@
+import dagster as dg
+
+
+class SeparatorConfig(dg.Config):
+    separator: str
+
+
+@dg.asset
+def processed_file(
+    primary_file: str, secondary_file: str, config: SeparatorConfig
+) -> str:
+    return f"{primary_file}{config.separator}{secondary_file}"
+
+
+# highlight-start
+def test_processed_file() -> None:
+    assert (
+        processed_file(
+            primary_file="abc",
+            secondary_file="def",
+            config=SeparatorConfig(separator=","),
+        )
+        == "abc,def"
+    )
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/asset-config.py

@@ -0,0 +1,18 @@
+import dagster as dg
+
+
+class FilepathConfig(dg.Config):
+    path: str
+
+
+@dg.asset
+def loaded_file(config: FilepathConfig) -> str:
+    with open(config.path) as file:
+        return file.read()
+
+
+# highlight-start
+def test_loaded_file() -> None:
+    assert loaded_file(FilepathConfig(path="path1.txt")) == "contents1"
+    assert loaded_file(FilepathConfig(path="path2.txt")) == "contents2"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/asset-context.py

@@ -0,0 +1,14 @@
+import dagster as dg
+
+
+@dg.asset(partitions_def=dg.DailyPartitionsDefinition("2024-01-01"))
+def loaded_file(context: dg.AssetExecutionContext) -> str:
+    with open(f"path_{context.partition_key}.txt") as file:
+        return file.read()
+
+
+# highlight-start
+def test_loaded_file() -> None:
+    context = dg.build_asset_context(partition_key="2024-08-16")
+    assert loaded_file(context) == "Contents for August 16th, 2024"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/asset-dependency.py

@@ -0,0 +1,18 @@
+import dagster as dg
+
+
+@dg.asset
+def loaded_file() -> str:
+    with open("path.txt") as file:
+        return file.read()
+
+
+@dg.asset
+def processed_file(loaded_file: str) -> str:
+    return loaded_file.strip()
+
+
+# highlight-start
+def test_processed_file() -> None:
+    assert processed_file(" contents  ") == "contents"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/asset-no-argument.py

@@ -0,0 +1,13 @@
+import dagster as dg
+
+
+@dg.asset
+def loaded_file() -> str:
+    with open("path.txt") as file:
+        return file.read()
+
+
+# highlight-start
+def test_loaded_file() -> None:
+    assert loaded_file() == "contents"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/asset-resource.py

@@ -0,0 +1,21 @@
+import mock
+from dagster_aws.s3 import S3FileHandle, S3FileManager
+
+import dagster as dg
+
+
+@dg.asset
+def loaded_file(file_manager: S3FileManager) -> str:
+    return file_manager.read_data(S3FileHandle("bucket", "path.txt"))
+
+
+# highlight-start
+def test_file() -> None:
+    mocked_resource = mock.Mock(spec=S3FileManager)
+    mocked_resource.read_data.return_value = "contents"
+
+    assert loaded_file(mocked_resource) == "contents"
+    assert mocked_resource.read_data.called_once_with(
+        S3FileHandle("bucket", "path.txt")
+    )
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/op-combo.py

@@ -0,0 +1,25 @@
+import dagster as dg
+
+
+class SeparatorConfig(dg.Config):
+    separator: str
+
+
+@dg.op
+def process_file(
+    primary_file: str, secondary_file: str, config: SeparatorConfig
+) -> str:
+    return f"{primary_file}{config.separator}{secondary_file}"
+
+
+# highlight-start
+def test_process_file() -> None:
+    assert (
+        process_file(
+            primary_file="abc",
+            secondary_file="def",
+            config=SeparatorConfig(separator=","),
+        )
+        == "abc,def"
+    )
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/op-config.py

@@ -0,0 +1,18 @@
+import dagster as dg
+
+
+class FilepathConfig(dg.Config):
+    path: str
+
+
+@dg.op
+def load_file(config: FilepathConfig) -> str:
+    with open(config.path) as file:
+        return file.read()
+
+
+# highlight-start
+def test_load_file() -> None:
+    assert load_file(FilepathConfig(path="path1.txt")) == "contents1"
+    assert load_file(FilepathConfig(path="path2.txt")) == "contents2"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/op-context.py

@@ -0,0 +1,14 @@
+import dagster as dg
+
+
+@dg.op
+def load_file(context: dg.OpExecutionContext) -> str:
+    with open(f"path_{context.partition_key}.txt") as file:
+        return file.read()
+
+
+# highlight-start
+def test_load_file() -> None:
+    context = dg.build_asset_context(partition_key="2024-08-16")
+    assert load_file(context) == "Contents for August 16th, 2024"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/op-dependency.py

@@ -0,0 +1,12 @@
+import dagster as dg
+
+
+@dg.op
+def process_file(loaded_file: str) -> str:
+    return loaded_file.strip()
+
+
+# highlight-start
+def test_process_file() -> None:
+    assert process_file(" contents  ") == "contents"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/op-no-argument.py

@@ -0,0 +1,13 @@
+import dagster as dg
+
+
+@dg.op
+def load_file() -> str:
+    with open("path.txt") as file:
+        return file.read()
+
+
+# highlight-start
+def test_load_file() -> None:
+    assert load_file() == "contents"
+    # highlight-end

examples/docs_beta_snippets/docs_beta_snippets/guides/quality-testing/unit-testing-assets-and-ops/op-resource.py

@@ -0,0 +1,21 @@
+import mock
+from dagster_aws.s3 import S3FileHandle, S3FileManager
+
+import dagster as dg
+
+
+@dg.op
+def load_file(file_manager: S3FileManager) -> str:
+    return file_manager.read_data(S3FileHandle("bucket", "path.txt"))
+
+
+# highlight-start
+def test_load_file() -> None:
+    mocked_resource = mock.Mock(spec=S3FileManager)
+    mocked_resource.read_data.return_value = "contents"
+
+    assert load_file(mocked_resource) == "contents"
+    assert mocked_resource.read_data.called_once_with(
+        S3FileHandle("bucket", "path.txt")
+    )
+    # highlight-end

examples/docs_beta_snippets/setup.py

@@ -14,10 +14,6 @@
         "Operating System :: OS Independent",
     ],
     packages=find_packages(exclude=["test"]),
-    install_requires=["dagster-cloud"],
-    extras_require={
-        "test": [
-            "pytest",
-        ]
-    },
+    install_requires=["dagster-cloud", "dagster-aws"],
+    extras_require={"test": ["pytest", "mock"]},
 )