Changeset View
Changeset View
Standalone View
Standalone View
python_modules/libraries/dagstermill/dagstermill_tests/test_solids.py
Show First 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | |||||
@contextmanager | @contextmanager | ||||
def exec_for_test(fn_name, env=None, raise_on_error=True, **kwargs): | def exec_for_test(fn_name, env=None, raise_on_error=True, **kwargs): | ||||
result = None | result = None | ||||
recon_pipeline = ReconstructablePipeline.for_module("dagstermill.examples.repository", fn_name) | recon_pipeline = ReconstructablePipeline.for_module("dagstermill.examples.repository", fn_name) | ||||
with instance_for_test() as instance: | with instance_for_test() as instance: | ||||
try: | try: | ||||
result = execute_pipeline( | result = execute_pipeline( | ||||
recon_pipeline, env, instance=instance, raise_on_error=raise_on_error, **kwargs | recon_pipeline, | ||||
env, | |||||
instance=instance, | |||||
raise_on_error=raise_on_error, | |||||
**kwargs, | |||||
) | ) | ||||
yield result | yield result | ||||
finally: | finally: | ||||
if result: | if result: | ||||
cleanup_result_notebook(result) | cleanup_result_notebook(result) | ||||
@pytest.mark.notebook_test | @pytest.mark.notebook_test | ||||
▲ Show 20 Lines • Show All 351 Lines • ▼ Show 20 Lines | |||||
def test_custom_description(): | def test_custom_description(): | ||||
test_description = "custom description" | test_description = "custom description" | ||||
test_solid = define_dagstermill_solid( | test_solid = define_dagstermill_solid( | ||||
BACKING_NB_NAME, BACKING_NB_PATH, description=test_description | BACKING_NB_NAME, BACKING_NB_PATH, description=test_description | ||||
) | ) | ||||
assert test_solid.description == test_description | assert test_solid.description == test_description | ||||
@pytest.mark.notebook_test | |||||
def test_retries(capsys): | |||||
with exec_for_test("retries_pipeline", raise_on_error=False) as result: | |||||
assert result.result_for_solid("yield_retry").retry_attempts == 1 | |||||
# the raise_retry solid should trigger a warning to use yield_event | |||||
warn_found = False | |||||
captured = capsys.readouterr() | |||||
for line in captured.err.split("\n"): | |||||
if "Use dagstermill.yield_event with RetryRequested or Failure" in line: | |||||
warn_found = True | |||||
assert warn_found | |||||
@pytest.mark.notebook_test | |||||
def test_failure(capsys): | |||||
with exec_for_test("failure_pipeline", raise_on_error=False) as result: | |||||
assert ( | |||||
result.result_for_solid("yield_failure").failure_data.user_failure_data.description | |||||
== "bad bad notebook" | |||||
) | |||||
# the raise_failure solid should trigger a warning to use yield_event | |||||
warn_found = False | |||||
captured = capsys.readouterr() | |||||
for line in captured.err.split("\n"): | |||||
if "Use dagstermill.yield_event with RetryRequested or Failure" in line: | |||||
warn_found = True | |||||
assert warn_found |