Fix TriggerDagRunOperator with deferrable parameter

[dylanbstorey]

Fixes a bug in the TriggerDagRunOperator when operating where deferrable=True.

The cause of the bug was a misunderstanding in which execution date was being brought through with the context + a bad mocked value in the original tests.

The suggested remedy (here) is to have the Trigger instance return a serialized version of itself and then use that to validate the exit status of the external dag.

Functional test below

from datetime import datetime
from airflow import DAG
from airflow.utils.state import TaskInstanceState, State

from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.operators.python import PythonOperator
from airflow.decorators import task

with DAG('driver_dags',
         schedule_interval=None,
         start_date=datetime(2023, 2, 7),
         catchup=False,
         ) as driver_dag:

    driver = TriggerDagRunOperator(
        task_id='driver',
        trigger_dag_id='down_stream',
        reset_dag_run=True,
        wait_for_completion=True,
        poke_interval=20,
        allowed_states=[TaskInstanceState.SUCCESS],
        do_xcom_push=True,
        deferrable=True,
    )


def _dummy_task():
    from time import sleep
    """ Create dummy object"""
    sleep(15)
    return


with DAG('down_stream',  schedule_interval=None,
         start_date=datetime(2023, 1, 23), is_paused_upon_creation=False, catchup=False) as downstream:

    empty = PythonOperator(
        task_id="empty", python_callable=_dummy_task,
        )

[josh-fell]

Looks good to me. A small question on potentially modifying the unit tests slightly to match the desired exception message.

[dylanbstorey]

Finished completing the PR after the UI rebase/merge issues last week, re requesting review and approval from previous individuals to just verify everything one last time.
Local tests and functional tests with an example DAG appear good.

[hussein-awala]

This solution is based on the result of self.serialize(), I wonder if we can add a test for this method (without mocking it) to make sure that when someone change it, a test for trigger dagrun blocks him until update the code which use it. WDYT?