Philipp G
10/31/2019, 1:56 PMimport unittest
from dagster import execute_solid, lambda_solid
@lambda_solid
def add_one(x):
return x + 1
class AddOneTest(unittest.TestCase):
def test_for_zero(self):
result = execute_solid(add_one, input_values={"x": 0})
self.assertEqual(1, result.output_value())
When I run the test with unittest
on the command line, the standard output of the test is quite extensive: (continue to read in the thread…)2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - PIPELINE_START - Started execution of pipeline "ephemeral_add_one_solid_pipeline".
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - ENGINE_EVENT - Executing steps in process (pid: 10455)
event_specific_data = {"metadata_entries": [["pid", null, ["10455"]], ["step_keys", null, ["{'add_one.compute', 'x.compute'}"]]]}
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - STEP_START - Started execution of step "x.compute".
solid = "x"
solid_definition = "x"
step_key = "x.compute"
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - STEP_OUTPUT - Yielded output "result" of type "Any". (Type check passed).
event_specific_data = {"intermediate_materialization": null, "step_output_handle": ["x.compute", "result"], "type_check_data": [true, "result", null, []]}
solid = "x"
solid_definition = "x"
step_key = "x.compute"
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - STEP_SUCCESS - Finished execution of step "x.compute" in 0.48ms.
event_specific_data = {"duration_ms": 0.48067804891616106}
solid = "x"
solid_definition = "x"
step_key = "x.compute"
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - STEP_START - Started execution of step "add_one.compute".
solid = "add_one"
solid_definition = "add_one"
step_key = "add_one.compute"
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - STEP_INPUT - Got input "x" of type "Any". (Type check passed).
event_specific_data = {"input_name": "x", "type_check_data": [true, "x", null, []]}
solid = "add_one"
solid_definition = "add_one"
step_key = "add_one.compute"
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - STEP_OUTPUT - Yielded output "result" of type "Any". (Type check passed).
event_specific_data = {"intermediate_materialization": null, "step_output_handle": ["add_one.compute", "result"], "type_check_data": [true, "result", null, []]}
solid = "add_one"
solid_definition = "add_one"
step_key = "add_one.compute"
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - STEP_SUCCESS - Finished execution of step "add_one.compute" in 0.44ms.
event_specific_data = {"duration_ms": 0.4425259539857507}
solid = "add_one"
solid_definition = "add_one"
step_key = "add_one.compute"
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - ENGINE_EVENT - Finished steps in process (pid: 10455) in 2.32ms
event_specific_data = {"metadata_entries": [["pid", null, ["10455"]], ["step_keys", null, ["{'add_one.compute', 'x.compute'}"]]]}
2019-10-31 13:25:28 - dagster - DEBUG - ephemeral_add_one_solid_pipeline - 4573124d-2990-413a-90bb-24b9671d2a04 - PIPELINE_SUCCESS - Finished execution of pipeline "ephemeral_add_one_solid_pipeline".
.
----------------------------------------------------------------------
Ran 1 test in 0.071s
OK
import unittest
from dagster import execute_solid, lambda_solid
@lambda_solid
def add_one(x):
return x + 1
class AddOneTest(unittest.TestCase):
def setUp(self):
self.silent_config = {"loggers": {"console": {"config": {"log_level": "CRITICAL"}}}}
def test_for_zero(self):
result = execute_solid(add_one, input_values={"x": 0}, environment_dict=self.silent_config)
self.assertEqual(1, result.output_value())
import unittest
from dagster import Field, execute_solid, solid
@solid(config={"add_more": Field(int, is_optional=True, default_value=0)})
def add_one(context, x):
return x + 1 + context.solid_config["add_more"]
class AddOneTest(unittest.TestCase):
def setUp(self):
self.silent_config = {
"loggers": {"console": {"config": {"log_level": "CRITICAL"}}}
}
def test_for_zero(self):
result = execute_solid(
add_one, input_values={"x": 0}, environment_dict=self.silent_config
)
self.assertEqual(1, result.output_value())
def test_add_more_to_zero(self):
test_config = {
"loggers": self.silent_config["loggers"],
"solids": {"add_one": {"config": {"add_more": {"value": 1}}}},
}
result = execute_solid(
add_one, input_values={"x": 0}, environment_dict=test_config
)
self.assertEqual(2, result.output_value())
CRITICAL
to silence output and the way I merge the silent_config
with the test_config
feel a bit weird. Does anyone have a better suggestion here? This may be a very specific problem with unittest
and not occurr in other testing modules like pytest
.alex
10/31/2019, 3:20 PM-b
option for unittest not work for you https://docs.python.org/2/library/unittest.html#cmdoption-unittest-b ?max
10/31/2019, 4:17 PMalex
10/31/2019, 4:20 PM-b
flag for unittest
does as wellPhilipp G
11/04/2019, 9:51 AM-b
works here, I didn’t think of it. I also can’t see any reason not to use it here. Thank you for the suggestion!