code: def mf_test_session_state(  # noqa: D
      request: FixtureRequest,
      disable_sql_alchemy_deprecation_warning: None,
      source_table_snapshot_repository: SqlTableSnapshotRepository,
  ) -> MetricFlowTestSessionState:
      engine_url = os.environ.get("MF_SQL_ENGINE_URL")
      assert engine_url is not None, (
          "MF_SQL_ENGINE_URL environment variable has not been set! Are you running in a properly configured "
          "environment? Check out our CONTRIBUTING.md for pointers to our environment configurations."
      )
      engine_password = os.environ.get("MF_SQL_ENGINE_PASSWORD", "")
  
      current_time = datetime.datetime.now().strftime("%Y_%m_%d")
      random_suffix = random_id()
      mf_system_schema = f"mf_test_{current_time}_{random_suffix}"
      default_source_schema = mf_system_schema
  
      use_persistent_source_schema = bool(
          request.config.getoption(
              USE_PERSISTENT_SOURCE_SCHEMA_CLI_FLAG, default=False
          )
      )
      if use_persistent_source_schema:
          source_table_snapshots_hash = SqlTableSnapshotHash.create_from_hashes(
              tuple(
                  table_snapshot.snapshot_hash
                  for table_snapshot in source_table_snapshot_repository.table_snapshots
              )
          )
          # Since the source schema is used in the SQL output snapshots (but replaced with *'s), make the name of the
          # schema to be the same length so that the SQL output snapshots don't change.
          persistent_source_schema_prefix = "mf_test_src_"
  
          # Ensure that there are a reasonable number of hash digits in the schema so that we don't get collisions.
          available_hash_digits = max(
              len(default_source_schema)
              - len(persistent_source_schema_prefix),
              0,
          )
          if available_hash_digits < 8:
              raise RuntimeError(
                  f"The generated name for the persistent source schema would have {available_hash_digits} data hash "
                  f"digits, which is not enough to have low collisions. default_source_schema: {default_source_schema} "
                  f"source_schema_prefix: {persistent_source_schema_prefix}"
              )
  
          # Use as many digits of the hash without exceeding the length of the default source schema.
          mf_source_schema = f"{persistent_source_schema_prefix}{source_table_snapshots_hash.str_value}"[
              : len(default_source_schema)
          ]
  
          if len(mf_source_schema) != len(default_source_schema):
              raise RuntimeError(
                  f"The persistent source schema should be the same length as the default source schema to keep the "
                  f"generated SQL snapshots consistent between non-persistent and persistent modes. persistent source "
                  f"schema: {mf_source_schema} default source schema: {default_source_schema}"
              )
  
          logger.info(
              f"Since the flag {USE_PERSISTENT_SOURCE_SCHEMA_CLI_FLAG} was specified, this session will use the "
              f"persistent source schema {mf_source_schema}. If the required source tables do not exist in this "
              f"schema, they will be created. However, the source schema (and the associated tables) will not "
              f"be dropped at the end of the testing session."
          )
      else:
          mf_source_schema = default_source_schema
  
      return MetricFlowTestSessionState(
          sql_engine_url=engine_url,
          sql_engine_password=engine_password,
          mf_system_schema=mf_system_schema,
          mf_source_schema=mf_source_schema,
          display_plans=bool(
              request.config.getoption(
                  DISPLAY_PLANS_CLI_FLAG, default=False
              )
          ),
          overwrite_snapshots=bool(
              request.config.getoption(
                  OVERWRITE_SNAPSHOTS_CLI_FLAG, default=False
              )
          ),
          plans_displayed=0,
          max_plans_displayed=6,
          use_persistent_source_schema=bool(
              request.config.getoption(
                  USE_PERSISTENT_SOURCE_SCHEMA_CLI_FLAG, default=False
              )
          ),
      )
  
imports: from __future__ import annotations
  import datetime
  import logging
  import os
  import warnings
  from dataclasses import dataclass
  from pathlib import Path
  import _pytest.config
  import pytest
  import sqlalchemy.util
  from _pytest.fixtures import FixtureRequest
  from sqlalchemy.engine import make_url
  from metricflow.random_id import random_id
  from metricflow.test.fixtures.sql_clients.common_client import SqlDialect
  from metricflow.test.table_snapshot.table_snapshots import SqlTableSnapshotHash, SqlTableSnapshotRepository
label_desc: Use image node_id 4 for calling a global function with example usage: mf_test_session_state(request, disable_sql_alchemy_deprecation_warning, source_table_snapshot_repository) and returns: MetricFlowTestSessionState