Differential D4787 Diff 23775 python_modules/dagster/dagster_tests/core_tests/definitions_tests/test_decorators.py

Changeset View

Standalone View

python_modules/dagster/dagster_tests/core_tests/definitions_tests/test_decorators.py

Show First 20 Lines • Show All 438 Lines • ▼ Show 20 Lines


def test_partitions_for_hourly_schedule_decorators_without_timezone():		def test_partitions_for_hourly_schedule_decorators_without_timezone():
with instance_for_test() as instance:		with instance_for_test() as instance:
with pendulum.test(pendulum.create(2019, 2, 27, 0, 1, 1, tz="US/Central")):		with pendulum.test(pendulum.create(2019, 2, 27, 0, 1, 1, tz="US/Central")):

context_without_time = ScheduleExecutionContext(instance, None)		context_without_time = ScheduleExecutionContext(instance, None)

start_date = datetime(year=2019, month=1, day=1, minute=1)		start_date = datetime(year=2019, month=1, day=1)

@hourly_schedule(		@hourly_schedule(
pipeline_name="foo_pipeline",		pipeline_name="foo_pipeline",
start_date=start_date,		start_date=start_date,
execution_time=time(hour=0, minute=25),		execution_time=time(hour=0, minute=25),
)		)
def hourly_foo_schedule(hourly_time):		def hourly_foo_schedule(hourly_time):
return {"hourly_time": hourly_time.isoformat()}		return {"hourly_time": hourly_time.isoformat()}

_check_partitions(		_check_partitions(
hourly_foo_schedule,		hourly_foo_schedule,
HOURS_UNTIL_FEBRUARY_27,		HOURS_UNTIL_FEBRUARY_27,
start_date,		start_date,
DEFAULT_HOURLY_FORMAT_WITHOUT_TIMEZONE,		DEFAULT_HOURLY_FORMAT_WITHOUT_TIMEZONE,
relativedelta(hours=1),		relativedelta(hours=1),
)		)

assert hourly_foo_schedule.get_run_config(context_without_time) == {		assert hourly_foo_schedule.get_run_config(context_without_time) == {
"hourly_time": pendulum.create(		"hourly_time": pendulum.create(
year=2019, month=2, day=26, hour=23, minute=1, tz="US/Central"		year=2019, month=2, day=26, hour=23, tz="US/Central"
).isoformat()		).isoformat()
}		}
assert hourly_foo_schedule.should_execute(context_without_time)		assert hourly_foo_schedule.should_execute(context_without_time)

# time that's invalid since it corresponds to a partition that hasn't happened yet		# time that's invalid since it corresponds to a partition that hasn't happened yet
# should not execute and should throw if it tries to generate run config		# should not execute and should throw if it tries to generate run config
execution_time_with_invalid_partition = datetime(		execution_time_with_invalid_partition = datetime(
year=2019, month=2, day=27, hour=3, minute=25		year=2019, month=2, day=27, hour=3, minute=25
Show All 10 Lines	with instance_for_test() as instance:
):		):
hourly_foo_schedule.get_run_config(context_with_invalid_time)		hourly_foo_schedule.get_run_config(context_with_invalid_time)

valid_time = datetime(year=2019, month=1, day=27, hour=1, minute=25)		valid_time = datetime(year=2019, month=1, day=27, hour=1, minute=25)
context_with_valid_time = ScheduleExecutionContext(instance, valid_time)		context_with_valid_time = ScheduleExecutionContext(instance, valid_time)

assert hourly_foo_schedule.get_run_config(context_with_valid_time) == {		assert hourly_foo_schedule.get_run_config(context_with_valid_time) == {
"hourly_time": pendulum.create(		"hourly_time": pendulum.create(
year=2019, month=1, day=27, hour=0, minute=1, tz="US/Central"		year=2019, month=1, day=27, hour=0, tz="US/Central"
).isoformat()		).isoformat()
}		}

assert hourly_foo_schedule.should_execute(context_with_valid_time)		assert hourly_foo_schedule.should_execute(context_with_valid_time)


def test_partitions_for_hourly_schedule_decorators_with_timezone():		def test_partitions_for_hourly_schedule_decorators_with_timezone():
with instance_for_test() as instance:		with instance_for_test() as instance:
with pendulum.test(pendulum.create(2019, 2, 27, 0, 1, 1, tz="US/Central")):		with pendulum.test(pendulum.create(2019, 2, 27, 0, 1, 1, tz="US/Central")):
start_date = datetime(year=2019, month=1, day=1, minute=1)		start_date = datetime(year=2019, month=1, day=1)

# You can specify a start date with no timezone and it will be assumed to be		# You can specify a start date with no timezone and it will be assumed to be
# in the execution timezone		# in the execution timezone

@hourly_schedule(		@hourly_schedule(
pipeline_name="foo_pipeline",		pipeline_name="foo_pipeline",
start_date=start_date,		start_date=start_date,
execution_time=time(hour=0, minute=25),		execution_time=time(hour=0, minute=25),
Show All 14 Lines	with instance_for_test() as instance:

valid_time = pendulum.create(		valid_time = pendulum.create(
year=2019, month=1, day=27, hour=1, minute=25, tz="US/Central"		year=2019, month=1, day=27, hour=1, minute=25, tz="US/Central"
)		)
context_with_valid_time = ScheduleExecutionContext(instance, valid_time)		context_with_valid_time = ScheduleExecutionContext(instance, valid_time)

assert hourly_central_schedule.get_run_config(context_with_valid_time) == {		assert hourly_central_schedule.get_run_config(context_with_valid_time) == {
"hourly_time": pendulum.create(		"hourly_time": pendulum.create(
year=2019, month=1, day=27, hour=0, minute=1, tz="US/Central"		year=2019, month=1, day=27, hour=0, tz="US/Central"
).isoformat()		).isoformat()
}		}

assert hourly_central_schedule.should_execute(context_with_valid_time)		assert hourly_central_schedule.should_execute(context_with_valid_time)

# You can specify a start date in a different timezone and it will be transformed into the		# You can specify a start date in a different timezone and it will be transformed into the
# execution timezone		# execution timezone
start_date_with_different_timezone = pendulum.create(2019, 1, 1, 0, 1, tz="US/Pacific")		start_date_with_different_timezone = pendulum.create(2019, 1, 1, 0, tz="US/Pacific")

@hourly_schedule(		@hourly_schedule(
pipeline_name="foo_pipeline",		pipeline_name="foo_pipeline",
start_date=start_date_with_different_timezone,		start_date=start_date_with_different_timezone,
execution_time=time(hour=0, minute=25),		execution_time=time(hour=0, minute=25),
execution_timezone="US/Central",		execution_timezone="US/Central",
)		)
def hourly_central_schedule_with_timezone_start_time(hourly_time):		def hourly_central_schedule_with_timezone_start_time(hourly_time):
▲ Show 20 Lines • Show All 259 Lines • ▼ Show 20 Lines	with pytest.raises(DagsterInvalidDefinitionError):
@monthly_schedule(		@monthly_schedule(
pipeline_name="foo_pipeline",		pipeline_name="foo_pipeline",
execution_day_of_month=32,		execution_day_of_month=32,
start_date=datetime(year=2019, month=1, day=1),		start_date=datetime(year=2019, month=1, day=1),
)		)
def monthly_foo_schedule_over():		def monthly_foo_schedule_over():
return {}		return {}

		with pytest.raises(
		DagsterInvalidDefinitionError,
		match=re.escape("`start_date` must be at the beginning of a month for a monthly schedule."),
		):

		@monthly_schedule(
		pipeline_name="foo_pipeline",
		execution_day_of_month=7,
		start_date=datetime(year=2019, month=1, day=5),
		)
		def monthly_foo_schedule_later_in_month():
		return {}

with pytest.raises(DagsterInvalidDefinitionError):		with pytest.raises(DagsterInvalidDefinitionError):

@monthly_schedule(		@monthly_schedule(
pipeline_name="foo_pipeline",		pipeline_name="foo_pipeline",
execution_day_of_month=0,		execution_day_of_month=0,
start_date=datetime(year=2019, month=1, day=1),		start_date=datetime(year=2019, month=1, day=1),
)		)
def monthly_foo_schedule_under():		def monthly_foo_schedule_under():
return {}		return {}

with pytest.raises(DagsterInvalidDefinitionError):		with pytest.raises(DagsterInvalidDefinitionError):

@weekly_schedule(		@weekly_schedule(
pipeline_name="foo_pipeline",		pipeline_name="foo_pipeline",
execution_day_of_week=7,		execution_day_of_week=7,
start_date=datetime(year=2019, month=1, day=1),		start_date=datetime(year=2019, month=1, day=1),
)		)
def weekly_foo_schedule_over():		def weekly_foo_schedule_over():
return {}		return {}

		with pytest.raises(
		DagsterInvalidDefinitionError,
		match=re.escape("`start_date` must be at the beginning of a day for a weekly schedule."),
		):

		@weekly_schedule(
		pipeline_name="foo_pipeline",
		execution_day_of_week=7,
		start_date=datetime(year=2019, month=1, day=1, hour=2),
		)
		def weekly_foo_schedule_start_later_in_day():
		return {}

		with pytest.raises(
		DagsterInvalidDefinitionError,
		match=re.escape("`start_date` must be at the beginning of a day for a daily schedule."),
		):

		@daily_schedule(
		pipeline_name="foo_pipeline", start_date=datetime(year=2019, month=1, day=1, hour=2),
		)
		def daily_foo_schedule_start_later_in_day():
		return {}

		with pytest.raises(
		DagsterInvalidDefinitionError,
		match=re.escape("`start_date` must be at the beginning of an hour for an hourly schedule."),
		):

		@hourly_schedule(
		pipeline_name="foo_pipeline",
		start_date=datetime(year=2019, month=1, day=1, hour=2, minute=30),
		)
		def hourly_foo_schedule_start_later_in_hour():
		return {}


def test_solid_docstring():		def test_solid_docstring():
@solid		@solid
def foo_solid(_):		def foo_solid(_):
"""FOO_DOCSTRING"""		"""FOO_DOCSTRING"""
return		return

@lambda_solid		@lambda_solid
▲ Show 20 Lines • Show All 125 Lines • Show Last 20 Lines