docstr: Verify that _cleanup_table actually deletes the rows it should.
  
  TaskInstance represents the "normal" case.  DagRun is the odd case where we want
  to keep the last non-externally-triggered DagRun record even if it should be
  deleted according to the provided timestamp.
  
  We also verify that the "on delete cascade" behavior is as expected.  Some tables
  have foreign keys defined so for example if we delete a dag run, all its associated
  task instances should be purged as well.  But if we delete task instances the
  associated dag runs should remain.
docstr_tok: ["Verify","that","_cleanup_table","actually","deletes","the","rows","it","should",".","TaskInstance","represents","the","``","normal","''","case",".","DagRun","is","the","odd","case","where","we","want","to","keep","the","last","non-externally-triggered","DagRun","record","even","if","it","should","be","deleted","according","to","the","provided","timestamp",".","We","also","verify","that","the","``","on","delete","cascade","''","behavior","is","as","expected",".","Some","tables","have","foreign","keys","defined","so","for","example","if","we","delete","a","dag","run",",","all","its","associated","task","instances","should","be","purged","as","well",".","But","if","we","delete","task","instances","the","associated","dag","runs","should","remain","."]
code: def test__cleanup_table(
      self,
      table_name,
      date_add_kwargs,
      expected_to_delete,
      external_trigger,
  ):
      """
      Verify that _cleanup_table actually deletes the rows it should.
  
      TaskInstance represents the "normal" case.  DagRun is the odd case where we want
      to keep the last non-externally-triggered DagRun record even if it should be
      deleted according to the provided timestamp.
  
      We also verify that the "on delete cascade" behavior is as expected.  Some tables
      have foreign keys defined so for example if we delete a dag run, all its associated
      task instances should be purged as well.  But if we delete task instances the
      associated dag runs should remain.
  
      """
      base_date = pendulum.DateTime(
          2022, 1, 1, tzinfo=pendulum.timezone("UTC")
      )
      num_tis = 10
      create_tis(
          base_date=base_date,
          num_tis=num_tis,
          external_trigger=external_trigger,
      )
      with create_session() as session:
          clean_before_date = base_date.add(**date_add_kwargs)
          _cleanup_table(
              **config_dict[table_name].__dict__,
              clean_before_timestamp=clean_before_date,
              dry_run=False,
              session=session,
              table_names=["dag_run", "task_instance"],
          )
          model = config_dict[table_name].orm_model
          expected_remaining = num_tis - expected_to_delete
          assert len(session.query(model).all()) == expected_remaining
          if model.name == "task_instance":
              assert len(session.query(DagRun).all()) == num_tis
          elif model.name == "dag_run":
              assert (
                  len(session.query(TaskInstance).all())
                  == expected_remaining
              )
          else:
              raise Exception("unexpected")
  
imports: from __future__ import annotations
  from contextlib import suppress
  from datetime import datetime
  from importlib import import_module
  from io import StringIO
  from pathlib import Path
  from unittest.mock import MagicMock, mock_open, patch
  from uuid import uuid4
  import pendulum
  import pytest
  from sqlalchemy import text
  from sqlalchemy.exc import OperationalError
  from sqlalchemy.ext.declarative import DeclarativeMeta
  from airflow.exceptions import AirflowException
  from airflow.models import DagModel, DagRun, TaskInstance
  from airflow.operators.python import PythonOperator
  from airflow.utils.db_cleanup import ARCHIVE_TABLE_PREFIX, CreateTableAs, _build_query, _cleanup_table, _confirm_drop_archives, _dump_table_to_file, _get_archived_table_names, config_dict, drop_archived_tables, export_archived_records, run_cleanup
  from airflow.utils.session import create_session
  from tests.test_utils.db import clear_db_dags, clear_db_datasets, clear_db_runs, drop_tables_with_prefix
label_desc: Use image node_id 8 for calling the TestDBCleanup obj's underlying member method code with example usage: obj.test__cleanup_table(table_name, date_add_kwargs, expected_to_delete, external_trigger) without return types