Remove test-specific error handling from production code

Address reviewer feedback: Remove try-except blocks that were catching
ImportError/AttributeError for test environments. Tests should use mocking
instead of having special cases in production code.

- Remove try-except blocks around XCom.get_value() calls in get_link()
- Add proper XCom.get_value() mocking to all tests that call get_link()
- Ensure tests properly mock XCom behavior instead of relying on error handling

Author: hwang-cadent

providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py

@@ -81,12 +81,7 @@ def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
             assert isinstance(operator, TriggerDagRunOperator)
 
         # Try to get the resolved dag_id from XCom first (for dynamic dag_ids)
-        trigger_dag_id = None
-        try:
-            trigger_dag_id = XCom.get_value(ti_key=ti_key, key=XCOM_DAG_ID)
-        except (ImportError, AttributeError):
-            # Fallback for test environments or when supervisor comms is not available
-            pass
+        trigger_dag_id = XCom.get_value(ti_key=ti_key, key=XCOM_DAG_ID)
 
         # Fallback to operator attribute and rendered fields if not in XCom
         if not trigger_dag_id:
@@ -99,12 +94,7 @@ def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
 
         # Fetch the correct dag_run_id for the triggerED dag which is
         # stored in xcom during execution of the triggerING task.
-        triggered_dag_run_id = None
-        try:
-            triggered_dag_run_id = XCom.get_value(ti_key=ti_key, key=XCOM_RUN_ID)
-        except (ImportError, AttributeError):
-            # Fallback for test environments or when supervisor comms is not available
-            pass
+        triggered_dag_run_id = XCom.get_value(ti_key=ti_key, key=XCOM_RUN_ID)
 
         if AIRFLOW_V_3_0_PLUS:
             from airflow.utils.helpers import build_airflow_dagrun_url

providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py

@@ -441,8 +441,11 @@ def test_trigger_dagrun_pushes_dag_id_to_xcom(self, dag_maker, session):
         run_id_xcom = triggering_ti.xcom_pull(key=XCOM_RUN_ID)
         assert run_id_xcom == "test_run_id"
 
-    def test_extra_operator_link(self, dag_maker, session):
+    @mock.patch("airflow.providers.standard.operators.trigger_dagrun.XCom.get_value")
+    def test_extra_operator_link(self, mock_xcom_get_value, dag_maker, session):
         """Asserts whether the correct extra links url will be created."""
+        from airflow.providers.standard.operators.trigger_dagrun import XCOM_RUN_ID
+        
         with dag_maker(TEST_DAG_ID, default_args={"start_date": DEFAULT_DATE}, serialized=True):
             task = TriggerDagRunOperator(
                 task_id="test_task", trigger_dag_id=TRIGGERED_DAG_ID, trigger_run_id="test_run_id"
@@ -452,6 +455,14 @@ def test_extra_operator_link(self, dag_maker, session):
 
         triggering_ti = session.query(TaskInstance).filter_by(task_id=task.task_id, dag_id=task.dag_id).one()
 
+        # Mock XCom.get_value to return None for dag_id but return run_id for XCOM_RUN_ID
+        def mock_get_value(ti_key, key):
+            if key == XCOM_RUN_ID:
+                return "test_run_id"
+            return None
+        
+        mock_xcom_get_value.side_effect = mock_get_value
+
         with mock.patch("airflow.utils.helpers.build_airflow_url_with_query") as mock_build_url:
             # This is equivalent of a task run calling this and pushing to xcom
             task.operator_extra_links[0].get_link(operator=task, ti_key=triggering_ti.key)
@@ -463,7 +474,8 @@ def test_extra_operator_link(self, dag_maker, session):
         }
         assert expected_args in args
 
-    def test_extra_operator_link_with_dynamic_dag_id(self, dag_maker, session):
+    @mock.patch("airflow.providers.standard.operators.trigger_dagrun.XCom.get_value")
+    def test_extra_operator_link_with_dynamic_dag_id(self, mock_xcom_get_value, dag_maker, session):
         """Test that operator link works correctly when dag_id is dynamically resolved from XCom."""
         from airflow.providers.standard.operators.trigger_dagrun import XCOM_DAG_ID, XCOM_RUN_ID
 
@@ -479,8 +491,15 @@ def test_extra_operator_link_with_dynamic_dag_id(self, dag_maker, session):
 
         triggering_ti = session.query(TaskInstance).filter_by(task_id=task.task_id, dag_id=task.dag_id).one()
 
-        # Simulate that the operator has pushed a dynamic dag_id to XCom
-        triggering_ti.xcom_push(key=XCOM_DAG_ID, value="dynamic_dag_id")
+        # Mock XCom.get_value to return our test values
+        def mock_get_value(ti_key, key):
+            if key == XCOM_DAG_ID:
+                return "dynamic_dag_id"
+            elif key == XCOM_RUN_ID:
+                return "test_run_id"
+            return None
+        
+        mock_xcom_get_value.side_effect = mock_get_value
 
         with mock.patch("airflow.utils.helpers.build_airflow_url_with_query") as mock_build_url:
             task.operator_extra_links[0].get_link(operator=task, ti_key=triggering_ti.key)