code: def test_ttl_duration(
      feature_store_session,
      role,
      feature_group_name,
      pandas_data_frame,
      record,
  ):
      feature_group = FeatureGroup(
          name=feature_group_name,
          sagemaker_session=feature_store_session,
      )
      feature_group.load_feature_definitions(
          data_frame=pandas_data_frame
      )
  
      record_identifier_value_as_string = record[0].value_as_string
  
      ttl_duration = TtlDuration(unit="Seconds", value=30)
  
      with cleanup_feature_group(feature_group):
          output = feature_group.create(
              s3_uri=False,
              record_identifier_name="feature1",
              event_time_feature_name="feature3",
              role_arn=role,
              enable_online_store=True,
              ttl_duration=ttl_duration,
          )
          _wait_for_feature_group_create(feature_group)
  
          feature_store = FeatureStore(
              sagemaker_session=feature_store_session
          )
  
          describe = feature_group.describe()
  
          assert ttl_duration.unit == (
              describe.get("OnlineStoreConfig")
              .get("TtlDuration")
              .get("Unit")
          )
  
          assert ttl_duration.value == (
              describe.get("OnlineStoreConfig")
              .get("TtlDuration")
              .get("Value")
          )
  
          record[2].value_as_string = (
              datetime.datetime.now(datetime.timezone.utc)
              .replace(microsecond=0)
              .isoformat()
              .split("+")[0]
              + "Z"
          )
  
          feature_group.put_record(record=record)
  
          sample_record = feature_store_session.get_record(
              feature_group_name=feature_group_name,
              record_identifier_value_as_string=record_identifier_value_as_string,
              feature_names=None,
              expiration_time_response="Enabled",
          )
          assert sample_record.get("Record") is not None
          _verify_expires_at(
              sample_record.get("ExpiresAt"),
              record[2].value_as_string,
              ttl_duration,
          )
  
          batch_records = feature_store.batch_get_record(
              identifiers=[
                  Identifier(
                      feature_group_name=feature_group_name,
                      record_identifiers_value_as_string=[
                          record_identifier_value_as_string
                      ],
                  )
              ],
              expiration_time_response="Enabled",
          ).get("Records")
          assert len(batch_records) == 1
          assert batch_records[0].get("Record") == sample_record.get(
              "Record"
          )
          assert batch_records[0].get("ExpiresAt") == sample_record.get(
              "ExpiresAt"
          )
  
          retrieved_record = feature_group.get_record(
              record_identifier_value_as_string=record_identifier_value_as_string
          )
  
          assert sample_record.get("Record") == retrieved_record
  
          sample_record = feature_store_session.get_record(
              feature_group_name=feature_group_name,
              record_identifier_value_as_string=record_identifier_value_as_string,
              feature_names=None,
          )
  
          assert sample_record.get("Record") is not None
          assert sample_record.get("ExpiresAt") is None
  
          batch_records = feature_store.batch_get_record(
              identifiers=[
                  Identifier(
                      feature_group_name=feature_group_name,
                      record_identifiers_value_as_string=[
                          record_identifier_value_as_string
                      ],
                  )
              ],
              expiration_time_response="Disabled",
          ).get("Records")
  
          assert len(batch_records) == 1
          assert batch_records[0].get("Record") == sample_record.get(
              "Record"
          )
          assert batch_records[0].get("ExpiresAt") is None
  
          time.sleep(35)
  
          sample_record = feature_store_session.get_record(
              feature_group_name=feature_group_name,
              record_identifier_value_as_string=record_identifier_value_as_string,
              feature_names=None,
          )
          assert sample_record.get("Record") is None
  
          retrieved_record = feature_group.get_record(
              record_identifier_value_as_string=record_identifier_value_as_string
          )
          assert retrieved_record is None
  
          batch_records = feature_store.batch_get_record(
              identifiers=[
                  Identifier(
                      feature_group_name=feature_group_name,
                      record_identifiers_value_as_string=[
                          record_identifier_value_as_string
                      ],
                  )
              ],
              expiration_time_response="Enabled",
          )["Records"]
          assert len(batch_records) == 0
  
          # customer record level ttl
          ttl_duration = TtlDuration(unit="Seconds", value=15)
  
          record[2].value_as_string = (
              datetime.datetime.now(datetime.timezone.utc)
              .replace(microsecond=0)
              .isoformat()
              .split("+")[0]
              + "Z"
          )
  
          feature_group.put_record(
              record=record, ttl_duration=ttl_duration
          )
  
          sample_record = feature_store_session.get_record(
              feature_group_name=feature_group_name,
              record_identifier_value_as_string=record_identifier_value_as_string,
              feature_names=None,
              expiration_time_response="Enabled",
          )
          assert sample_record.get("Record") is not None
          _verify_expires_at(
              sample_record.get("ExpiresAt"),
              record[2].value_as_string,
              ttl_duration,
          )
  
          time.sleep(15)
  
          sample_record = feature_store_session.get_record(
              feature_group_name=feature_group_name,
              record_identifier_value_as_string=record_identifier_value_as_string,
              feature_names=None,
          )
          assert sample_record.get("Record") is None
  
          # update default ttl for feature group
          ttl_duration = TtlDuration(unit="Seconds", value=40)
  
          online_store_config = OnlineStoreConfigUpdate(
              ttl_duration=ttl_duration
          )
          feature_group.update(online_store_config=online_store_config)
          # cache feature group for 5min, wait for new ttl to take effect
          time.sleep(300)
  
          describe = feature_group.describe()
  
          assert ttl_duration.unit == (
              describe.get("OnlineStoreConfig")
              .get("TtlDuration")
              .get("Unit")
          )
  
          assert ttl_duration.value == (
              describe.get("OnlineStoreConfig")
              .get("TtlDuration")
              .get("Value")
          )
  
          record[2].value_as_string = (
              datetime.datetime.now(datetime.timezone.utc)
              .replace(microsecond=0)
              .isoformat()
              .split("+")[0]
              + "Z"
          )
  
          feature_group.put_record(record=record)
  
          sample_record = feature_store_session.get_record(
              feature_group_name=feature_group_name,
              record_identifier_value_as_string=record_identifier_value_as_string,
              feature_names=None,
              expiration_time_response="Enabled",
          )
          assert sample_record.get("Record") is not None
          _verify_expires_at(
              sample_record.get("ExpiresAt"),
              record[2].value_as_string,
              ttl_duration,
          )
  
          time.sleep(40)
  
          sample_record = feature_store_session.get_record(
              feature_group_name=feature_group_name,
              record_identifier_value_as_string=record_identifier_value_as_string,
              feature_names=None,
          )
          assert sample_record.get("Record") is None
  
      assert output["FeatureGroupArn"].endswith(
          f"feature-group/{feature_group_name}"
      )
  
imports: from __future__ import absolute_import
  import datetime
  import json
  import time
  import dateutil.parser
  from contextlib import contextmanager
  import boto3
  import numpy
  import pandas
  import pytest
  from pandas import DataFrame
  from sagemaker.feature_store.feature_utils import get_feature_group_as_dataframe
  from sagemaker.feature_store.feature_definition import FractionalFeatureDefinition, StringFeatureDefinition, ListCollectionType
  from sagemaker.feature_store.feature_group import FeatureGroup
  from sagemaker.feature_store.feature_store import FeatureStore
  from sagemaker.feature_store.inputs import FeatureValue, FeatureParameter, TableFormatEnum, Filter, ResourceEnum, Identifier, DeletionModeEnum, TtlDuration, OnlineStoreConfigUpdate, OnlineStoreStorageTypeEnum
  from sagemaker.feature_store.dataset_builder import JoinTypeEnum
  from sagemaker.session import get_execution_role, Session
  from tests.integ.timeout import timeout
  from urllib.parse import urlparse
label_desc: Use image node_id 21 for calling a global function with example usage: test_ttl_duration(feature_store_session, role, feature_group_name, pandas_data_frame, record) without return types