new test

python_modules/dagster-pipes/dagster_pipes_tests/test_context.py

@@ -4,6 +4,10 @@
 
 import jsonschema
 import pytest
+from dagster._core.definitions.decorators.asset_decorator import asset
+from dagster._core.definitions.materialize import materialize
+from dagster._core.execution.context.compute import AssetExecutionContext
+from dagster._core.pipes.context import build_external_execution_context_data
 from dagster_pipes import (
     PIPES_PROTOCOL_VERSION,
     PIPES_PROTOCOL_VERSION_FIELD,
@@ -263,6 +267,19 @@ def test_message_json_schema_validation():
     jsonschema.validate(message, get_pipes_json_schema("message"))
 
 
+def test_context_json_schema_validation():
+    @asset
+    def foo(context: AssetExecutionContext):
+        context_data_no_extras = build_external_execution_context_data(context, extras=None)
+        jsonschema.validate(context_data_no_extras, get_pipes_json_schema("context"))
+        context_data_with_extras = build_external_execution_context_data(
+            context, extras={"foo": "bar"}
+        )
+        jsonschema.validate(context_data_with_extras, get_pipes_json_schema("context"))
+
+    materialize([foo])
+
+
 def test_json_schema_rejects_invalid():
     with pytest.raises(jsonschema.ValidationError):
         jsonschema.validate({"foo": "bar"}, get_pipes_json_schema("context"))

python_modules/dagster/dagster/_core/execution/context/invocation.py

@@ -218,6 +218,10 @@ def op_def(self) -> OpDefinition:
     def assets_def(self) -> AssetsDefinition:
         raise DagsterInvalidPropertyError(_property_msg("assets_def", "property"))
 
+    @property
+    def has_assets_def(self) -> bool:
+        return self._assets_def is not None
+
     @property
     def has_partition_key(self) -> bool:
         return self._partition_key is not None