text: from datetime import datetime
  
  import pendulum
  from dagster import (
      DailyPartitionsDefinition,
      HourlyPartitionsDefinition,
      MonthlyPartitionsDefinition,
      daily_partitioned_config,
      hourly_partitioned_config,
      monthly_partitioned_config,
  )
  from dagster.core.definitions.time_window_partitions import TimeWindow
  from dagster.utils.partitions import DEFAULT_HOURLY_FORMAT_WITHOUT_TIMEZONE
  
  DATE_FORMAT = "%Y-%m-%d"
  
  
  def time_window(start: str, end: str) -> TimeWindow:
      return TimeWindow(pendulum.parse(start), pendulum.parse(end))
  
  
  def test_daily_partitions():
      @daily_partitioned_config(start_date="2021-05-05")
      def my_partitioned_config(_start, _end):
          return {}
  
      partitions_def = my_partitioned_config.partitions_def
      assert partitions_def == DailyPartitionsDefinition(start_date="2021-05-05")
  
      assert [
          partition.value
          for partition in partitions_def.get_partitions(datetime.strptime("2021-05-07", DATE_FORMAT))
      ] == [
          time_window("2021-05-05", "2021-05-06"),
          time_window("2021-05-06", "2021-05-07"),
      ]
  
      assert partitions_def.time_window_for_partition_key("2021-05-08") == time_window(
          "2021-05-08", "2021-05-09"
      )
  
  
  def test_daily_partitions_with_end_offset():
      @daily_partitioned_config(start_date="2021-05-05", end_offset=2)
      def my_partitioned_config(_start, _end):
          return {}
  
      assert [
          partition.value
          for partition in my_partitioned_config.partitions_def.get_partitions(
              datetime.strptime("2021-05-07", DATE_FORMAT)
          )
      ] == [
          time_window("2021-05-05", "2021-05-06"),
          time_window("2021-05-06", "2021-05-07"),
          time_window("2021-05-07", "2021-05-08"),
          time_window("2021-05-08", "2021-05-09"),
      ]
  
  
  def test_daily_partitions_with_negative_end_offset():
      @daily_partitioned_config(start_date="2021-05-01", end_offset=-2)
      def my_partitioned_config(_start, _end):
          return {}
  
      assert [
          partition.value
          for partition in my_partitioned_config.partitions_def.get_partitions(
              datetime.strptime("2021-05-07", DATE_FORMAT)
          )
      ] == [
          time_window("2021-05-01", "2021-05-02"),
          time_window("2021-05-02", "2021-05-03"),
          time_window("2021-05-03", "2021-05-04"),
          time_window("2021-05-04", "2021-05-05"),
      ]
  
  
  def test_monthly_partitions():
      @monthly_partitioned_config(start_date="2021-05-01")
      def my_partitioned_config(_start, _end):
          return {}
  
      partitions_def = my_partitioned_config.partitions_def
      assert partitions_def == MonthlyPartitionsDefinition(start_date="2021-05-01")
  
      assert [
          partition.value
          for partition in partitions_def.get_partitions(datetime.strptime("2021-07-03", DATE_FORMAT))
      ] == [
          time_window("2021-05-01", "2021-06-01"),
          time_window("2021-06-01", "2021-07-01"),
      ]
  
      assert partitions_def.time_window_for_partition_key("2021-05-01") == time_window(
          "2021-05-01", "2021-06-01"
      )
  
  
  def test_monthly_partitions_with_end_offset():
      @monthly_partitioned_config(start_date="2021-05-01", end_offset=2)
      def my_partitioned_config(_start, _end):
          return {}
  
      assert [
          partition.value
          for partition in my_partitioned_config.partitions_def.get_partitions(
              datetime.strptime("2021-07-03", DATE_FORMAT)
          )
      ] == [
          time_window("2021-05-01", "2021-06-01"),
          time_window("2021-06-01", "2021-07-01"),
          time_window("2021-07-01", "2021-08-01"),
          time_window("2021-08-01", "2021-09-01"),
      ]
  
  
  def test_hourly_partitions():
      @hourly_partitioned_config(start_date="2021-05-05-01:00")
      def my_partitioned_config(_start, _end):
          return {}
  
      partitions_def = my_partitioned_config.partitions_def
      assert partitions_def == HourlyPartitionsDefinition(start_date="2021-05-05-01:00")
  
      partitions = partitions_def.get_partitions(
          datetime.strptime("2021-05-05-03:00", DEFAULT_HOURLY_FORMAT_WITHOUT_TIMEZONE)
      )
  
      assert [partition.value for partition in partitions] == [
          time_window("2021-05-05T01:00:00", "2021-05-05T02:00:00"),
          time_window("2021-05-05T02:00:00", "2021-05-05T03:00:00"),
      ]
  
      assert [partition.name for partition in partitions] == [
          "2021-05-05-01:00",
          "2021-05-05-02:00",
      ]
  
      assert partitions_def.time_window_for_partition_key("2021-05-05-01:00") == time_window(
          "2021-05-05T01:00:00", "2021-05-05T02:00:00"
      )
  