code: def assert_snapshot_text_equal(
      request: FixtureRequest,
      mf_test_session_state: MetricFlowTestSessionState,
      group_id: str,
      snapshot_id: str,
      snapshot_text: str,
      snapshot_file_extension: str,
      exclude_line_regex: Optional[str] = None,
      incomparable_strings_replacement_function: Optional[
          Callable[[str], str]
      ] = None,
      additional_sub_directories_for_snapshots: Tuple[str, ...] = (),
  ) -> None:
      """Similar to assert_plan_snapshot_text_equal(), but with more controls on how the snapshot paths are generated."""
      file_path = (
          snapshot_path_prefix(
              request=request,
              snapshot_group=group_id,
              snapshot_id=snapshot_id,
              additional_sub_directories=additional_sub_directories_for_snapshots,
          )
          + snapshot_file_extension
      )
  
      if incomparable_strings_replacement_function is not None:
          snapshot_text = incomparable_strings_replacement_function(
              snapshot_text
          )
  
      # Add a new line at the end of the file so that PRs don't show the "no newline" symbol on Github.
      if len(snapshot_text) > 1 and snapshot_text[-1] != "\n":
          snapshot_text = snapshot_text + "\n"
  
      # If we are in overwrite mode, make a new plan:
      if mf_test_session_state.overwrite_snapshots:
          # Create parent directory for the plan text files.
          os.makedirs(os.path.dirname(file_path), exist_ok=True)
          with open(file_path, "w") as snapshot_text_file:
              snapshot_text_file.write(snapshot_text)
  
      # Throw an exception if the plan is not there.
      if not os.path.exists(file_path):
          raise FileNotFoundError(
              f"Could not find snapshot file at path {file_path}. Re-run with --overwrite-snapshots and check git status "
              "to see what's new."
          )
  
      if mf_test_session_state.display_plans:
          if not mf_test_session_state.overwrite_snapshots:
              logger.warning(
                  f"Not overwriting snapshots, so displaying existing snapshot at {file_path}"
              )
  
          if (
              mf_test_session_state.plans_displayed
              >= mf_test_session_state.max_plans_displayed
          ):
              raise RuntimeError(
                  f"Can't display snapshot - hit limit of "
                  f"{mf_test_session_state.max_plans_displayed} "
                  f"plans displayed."
              )
          webbrowser.open("file://" + file_path)
          mf_test_session_state.plans_displayed += 1
  
      # Read the existing plan from the file and compare with the actual plan
      with open(file_path, "r") as snapshot_text_file:
          expected_snapshot_text = snapshot_text_file.read()
  
          if exclude_line_regex:
              # Filter out lines that should be ignored.
              expected_snapshot_text = _exclude_lines_matching_regex(
                  file_contents=expected_snapshot_text,
                  exclude_line_regex=exclude_line_regex,
              )
              snapshot_text = _exclude_lines_matching_regex(
                  file_contents=snapshot_text,
                  exclude_line_regex=exclude_line_regex,
              )
          # pytest should show a detailed diff with "assert actual_modified == expected_modified", but it's not, so doing
          # this instead.
          if snapshot_text != expected_snapshot_text:
              differ = difflib.Differ()
              diff = differ.compare(
                  expected_snapshot_text.splitlines(),
                  snapshot_text.splitlines(),
              )
              assert False, (
                  f"Snapshot from {file_path} does not match. Diff from expected to actual:\n"
                  + "\n".join(diff)
              )
  
imports: from __future__ import annotations
  import difflib
  import logging
  import os
  import re
  import webbrowser
  from typing import Any, Callable, List, Optional, Tuple, TypeVar
  import tabulate
  from _pytest.fixtures import FixtureRequest
  from dbt_semantic_interfaces.pretty_print import pformat_big_objects
  from metricflow.dag.mf_dag import MetricFlowDag
  from metricflow.dataflow.dataflow_plan import DataflowPlan
  from metricflow.dataflow.dataflow_plan_to_text import dataflow_plan_as_text
  from metricflow.execution.execution_plan import ExecutionPlan
  from metricflow.execution.execution_plan_to_text import execution_plan_to_text
  from metricflow.mf_logging.pretty_print import mf_pformat
  from metricflow.model.semantics.linkable_spec_resolver import LinkableElementSet
  from metricflow.naming.object_builder_scheme import ObjectBuilderNamingScheme
  from metricflow.protocols.sql_client import SqlClient
  from metricflow.specs.specs import InstanceSpecSet, LinkableSpecSet
  from metricflow.test.fixtures.setup_fixtures import MetricFlowTestSessionState, check_sql_engine_snapshot_marker
label_desc: Use image node_id 6 for calling a global function with example usage: assert_snapshot_text_equal(request, mf_test_session_state, group_id, snapshot_id, snapshot_text, snapshot_file_extension, exclude_line_regex, incomparable_strings_replacement_function, additional_sub_directories_for_snapshots) without return types