@pytest.mark.need_serialized_dag
def test_bigquery_operator_extra_serialized_field_when_single_query(self, dag_maker, create_task_instance_of_operator):
    ti = create_task_instance_of_operator(BigQueryExecuteQueryOperator, dag_id=TEST_DAG_ID, execution_date=DEFAULT_DATE, task_id=TASK_ID, sql='SELECT * FROM test_table')
    serialized_dag = dag_maker.get_serialized_data()
    assert ('sql' in serialized_dag['dag']['tasks'][0])
    dag = SerializedDAG.from_dict(serialized_dag)
    simple_task = dag.task_dict[TASK_ID]
    assert (getattr(simple_task, 'sql') == 'SELECT * FROM test_table')
    assert (serialized_dag['dag']['tasks'][0]['_operator_extra_links'] == [{'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleLink': {}}])
    assert isinstance(list(simple_task.operator_extra_links)[0], BigQueryConsoleLink)
    ti.xcom_push('job_id', 12345)
    url = simple_task.get_extra_links(ti, BigQueryConsoleLink.name)
    assert (url == 'https://console.cloud.google.com/bigquery?j=12345')
@pytest.mark.need_serialized_dag
def test_bigquery_operator_extra_serialized_field_when_single_query(self, dag_maker, create_task_instance_of_operator):
    ti = create_task_instance_of_operator(BigQueryExecuteQueryOperator, dag_id=TEST_DAG_ID, execution_date=DEFAULT_DATE, task_id=TASK_ID, sql='SELECT * FROM test_table')
    serialized_dag = dag_maker.get_serialized_data()
    assert ('sql' in serialized_dag['dag']['tasks'][0])
    dag = SerializedDAG.from_dict(serialized_dag)
    simple_task = dag.task_dict[TASK_ID]
    assert (getattr(simple_task, 'sql') == 'SELECT * FROM test_table')
    assert (serialized_dag['dag']['tasks'][0]['_operator_extra_links'] == [{'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleLink': {}}])
    assert isinstance(list(simple_task.operator_extra_links)[0], BigQueryConsoleLink)
    ti.xcom_push('job_id', 12345)
    url = simple_task.get_extra_links(ti, BigQueryConsoleLink.name)
    assert (url == 'https://console.cloud.google.com/bigquery?j=12345')