Add docs for top level job inputs (#8212)

* Add docs for top level job inputs

* Move docs example

* lint

docs/content/concepts/testing.mdx

@@ -268,3 +268,35 @@ def test_event_stream():
     assert isinstance(materialization, AssetMaterialization)
     assert materialization.label == "persisted_string"
 ```
+
+### Testing jobs with top-level inputs
+
+You can wire inputs from the top-level of a job to the constituent ops. Consider the following op and graph:
+
+```python file=/concepts/ops_jobs_graphs/jobs.py startafter=start_top_level_input_graph endbefore=end_top_level_input_graph
+from dagster import graph, op
+
+
+@op
+def op_with_input(x):
+    return do_something(x)
+
+
+@graph
+def wires_input(x):
+    op_with_input(x)
+```
+
+Turn the graph into a job by calling <PyObject object="GraphDefinition" method="to_job" />, and provide a value to the input `x` using the `input_values` argument:
+
+```python file=/concepts/ops_jobs_graphs/jobs.py startafter=start_top_level_input_job endbefore=end_top_level_input_job
+the_job = wires_input.to_job(input_values={"x": 5})
+```
+
+You can also provide input values using <PyObject object="JobDefinition" method="execute_in_process" /> or <PyObject object="GraphDefinition" method="execute_in_process" />:
+
+```python file=/concepts/ops_jobs_graphs/jobs.py startafter=start_execute_in_process_input endbefore=end_execute_in_process_input
+graph_result = wires_input.execute_in_process(input_values={"x": 5})
+
+job_result = the_job.execute_in_process(input_values={"x": 6})  # Overrides existing input value
+```

examples/docs_snippets/docs_snippets/concepts/ops_jobs_graphs/jobs.py

@@ -1,4 +1,6 @@
-# pylint: disable=unused-argument
+# isort: skip_file
+
+# pylint: disable=unused-argument,reimported
 
 from dagster import DependencyDefinition, GraphDefinition, job, op
 
@@ -68,3 +70,34 @@ def my_tags_job():
 
 
 # end_tags_pipeline
+
+
+def do_something(x):
+    return x
+
+
+# start_top_level_input_graph
+from dagster import graph, op
+
+
+@op
+def op_with_input(x):
+    return do_something(x)
+
+
+@graph
+def wires_input(x):
+    op_with_input(x)
+
+
+# end_top_level_input_graph
+
+# start_top_level_input_job
+the_job = wires_input.to_job(input_values={"x": 5})
+# end_top_level_input_job
+
+# start_execute_in_process_input
+graph_result = wires_input.execute_in_process(input_values={"x": 5})
+
+job_result = the_job.execute_in_process(input_values={"x": 6})  # Overrides existing input value
+# end_execute_in_process_input

examples/docs_snippets/docs_snippets_tests/concepts_tests/ops_jobs_graphs_tests/test_jobs.py

@@ -11,9 +11,12 @@
 from docs_snippets.concepts.ops_jobs_graphs.fan_in_job import fan_in
 from docs_snippets.concepts.ops_jobs_graphs.jobs import (
     alias,
+    graph_result,
+    job_result,
     one_plus_one,
     one_plus_one_from_constructor,
     tagged_add_one,
+    the_job,
 )
 from docs_snippets.concepts.ops_jobs_graphs.jobs_from_graphs import local_job, prod_job
 from docs_snippets.concepts.ops_jobs_graphs.linear_job import linear
@@ -87,3 +90,10 @@ def test_retry_examples():
 def test_jobs_from_graphs():
     assert local_job.execute_in_process()
     assert prod_job.execute_in_process()
+
+
+def test_input_values():
+    assert the_job.execute_in_process().success
+    assert graph_result.success
+    assert job_result.success
+    assert job_result.output_for_node("op_with_input") == 6

examples/hacker_news_assets/hacker_news_assets_tests/test_core.py

@@ -8,8 +8,8 @@
 from dagster import (
     AssetGroup,
     ResourceDefinition,
-    load_assets_from_package_module,
     fs_io_manager,
+    load_assets_from_package_module,
     mem_io_manager,
 )