@pytest.mark.need_serialized_dag
def test_bigquery_operator_extra_serialized_field_when_multiple_queries(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', 'SELECT * FROM test_table2'])
    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', 'SELECT * FROM test_table2'])
    assert (serialized_dag['dag']['tasks'][0]['_operator_extra_links'] == [{'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink': {'index': 0}}, {'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink': {'index': 1}}])
    assert isinstance(list(simple_task.operator_extra_links)[0], BigQueryConsoleIndexableLink)
    job_id = ['123', '45']
    ti.xcom_push(key='job_id', value=job_id)
    assert ({'BigQuery Console #1', 'BigQuery Console #2'} == simple_task.operator_extra_link_dict.keys())
    assert ('https://console.cloud.google.com/bigquery?j=123' == simple_task.get_extra_links(ti, 'BigQuery Console #1'))
    assert ('https://console.cloud.google.com/bigquery?j=45' == simple_task.get_extra_links(ti, 'BigQuery Console #2'))
@pytest.mark.need_serialized_dag
def test_bigquery_operator_extra_serialized_field_when_multiple_queries(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', 'SELECT * FROM test_table2'])
    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', 'SELECT * FROM test_table2'])
    assert (serialized_dag['dag']['tasks'][0]['_operator_extra_links'] == [{'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink': {'index': 0}}, {'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink': {'index': 1}}])
    assert isinstance(list(simple_task.operator_extra_links)[0], BigQueryConsoleIndexableLink)
    job_id = ['123', '45']
    ti.xcom_push(key='job_id', value=job_id)
    assert ({'BigQuery Console #1', 'BigQuery Console #2'} == simple_task.operator_extra_link_dict.keys())
    assert ('https://console.cloud.google.com/bigquery?j=123' == simple_task.get_extra_links(ti, 'BigQuery Console #1'))
    assert ('https://console.cloud.google.com/bigquery?j=45' == simple_task.get_extra_links(ti, 'BigQuery Console #2'))