docstr: Test that the dag file processor does not try to insert already existing item into the database
docstr_tok: ["Test","that","the","dag","file","processor","does","not","try","to","insert","already","existing","item","into","the","database"]
code: def test_dag_file_processor_sla_miss_doesnot_raise_integrity_error(
      self, mock_get_dagbag, mock_stats_incr, dag_maker
  ):
      """
      Test that the dag file processor does not try to insert already existing item into the database
      """
      session = settings.Session()
  
      # Create dag with a start of 2 days ago, but an sla of 1 day
      # ago so we'll already have an sla_miss on the books
      test_start_date = timezone.utcnow() - datetime.timedelta(days=2)
      with dag_maker(
          dag_id="test_sla_miss",
          default_args={
              "start_date": test_start_date,
              "sla": datetime.timedelta(days=1),
          },
      ) as dag:
          task = EmptyOperator(task_id="dummy")
  
      dag_maker.create_dagrun(
          execution_date=test_start_date, state=State.SUCCESS
      )
  
      # Create a TaskInstance for two days ago
      ti = TaskInstance(
          task=task, execution_date=test_start_date, state="success"
      )
      session.merge(ti)
      session.flush()
  
      mock_dagbag = mock.Mock()
      mock_dagbag.get_dag.return_value = dag
      mock_get_dagbag.return_value = mock_dagbag
  
      DagFileProcessor.manage_slas(
          dag_folder=dag.fileloc,
          dag_id="test_sla_miss",
          session=session,
      )
      sla_miss_count = (
          session.query(SlaMiss)
          .filter(
              SlaMiss.dag_id == dag.dag_id,
              SlaMiss.task_id == task.task_id,
          )
          .count()
      )
      assert sla_miss_count == 1
      mock_stats_incr.assert_called_with(
          "sla_missed",
          tags={"dag_id": "test_sla_miss", "task_id": "dummy"},
      )
      # Now call manage_slas and see that it runs without errors
      # because of existing SlaMiss above.
      # Since this is run often, it's possible that it runs before another
      # ti is successful thereby trying to insert a duplicate record.
      DagFileProcessor.manage_slas(
          dag_folder=dag.fileloc,
          dag_id="test_sla_miss",
          session=session,
      )
  
imports: from __future__ import annotations
  import datetime
  import os
  from unittest import mock
  from unittest.mock import MagicMock, patch
  from zipfile import ZipFile
  import pytest
  from airflow import PY311, settings
  from airflow.callbacks.callback_requests import TaskCallbackRequest
  from airflow.configuration import TEST_DAGS_FOLDER, conf
  from airflow.dag_processing.manager import DagFileProcessorAgent
  from airflow.dag_processing.processor import DagFileProcessor, DagFileProcessorProcess
  from airflow.models import DagBag, DagModel, SlaMiss, TaskInstance, errors
  from airflow.models.serialized_dag import SerializedDagModel
  from airflow.models.taskinstance import SimpleTaskInstance
  from airflow.operators.empty import EmptyOperator
  from airflow.utils import timezone
  from airflow.utils.session import create_session
  from airflow.utils.state import State
  from airflow.utils.types import DagRunType
  from tests.test_utils.config import conf_vars, env_vars
  from tests.test_utils.db import clear_db_dags, clear_db_import_errors, clear_db_jobs, clear_db_pools, clear_db_runs, clear_db_serialized_dags, clear_db_sla_miss
  from tests.test_utils.mock_executor import MockExecutor
label_desc: Use image node_id 9 for calling the TestDagFileProcessor obj's underlying member method code with example usage: obj.test_dag_file_processor_sla_miss_doesnot_raise_integrity_error(mock_get_dagbag, mock_stats_incr, dag_maker) without return types