Skip to content

Commit fdd2a4c

Browse files
committed
Fix unit tests for TestGetXComByKeys
1 parent 25f6850 commit fdd2a4c

1 file changed

Lines changed: 24 additions & 14 deletions

File tree

  • airflow-core/tests/unit/api_fastapi/execution_api/versions/head

‎airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py‎

Lines changed: 24 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -64,11 +64,10 @@ def _(
6464
dag_id: str = Path(),
6565
run_id: str = Path(),
6666
task_id: str = Path(),
67-
xcom_key: str = Path(alias="key"),
6867
token=CurrentTIToken,
6968
):
7069
with create_session() as session:
71-
has_xcom_access(dag_id, run_id, task_id, xcom_key, request, session, token)
70+
has_xcom_access(dag_id, run_id, task_id, request, session, token)
7271
raise HTTPException(
7372
status_code=status.HTTP_403_FORBIDDEN,
7473
detail={
@@ -825,39 +824,50 @@ def test_empty_key_list_returns_empty_list(self, client, create_task_instance, s
825824
assert response.status_code == 200
826825
assert response.json() == []
827826

828-
def test_map_index_filter(self, client, create_task_instance, session):
827+
def test_map_index_filter(self, client, dag_maker, session):
829828
"""Only XCom rows matching the requested map_index are returned."""
830-
ti = create_task_instance()
829+
830+
class MyOperator(EmptyOperator):
831+
def __init__(self, *, x, **kwargs):
832+
super().__init__(**kwargs)
833+
self.x = x
834+
835+
with dag_maker(dag_id="dag"):
836+
MyOperator.partial(task_id="task").expand(x=["a", "b"])
837+
dag_run = dag_maker.create_dagrun(run_id="run")
838+
tis = {ti.map_index: ti for ti in dag_run.task_instances}
839+
mapped_ti = tis[0]
840+
831841
session.add(
832842
XComModel(
833843
key="return_value_0",
834844
value="for_index_0",
835-
dag_run_id=ti.dag_run.id,
836-
run_id=ti.run_id,
837-
task_id=ti.task_id,
838-
dag_id=ti.dag_id,
845+
dag_run_id=dag_run.id,
846+
run_id=dag_run.run_id,
847+
task_id=mapped_ti.task_id,
848+
dag_id=mapped_ti.dag_id,
839849
map_index=0,
840850
)
841851
)
842852
session.add(
843853
XComModel(
844854
key="return_value_0",
845855
value="for_default",
846-
dag_run_id=ti.dag_run.id,
847-
run_id=ti.run_id,
848-
task_id=ti.task_id,
849-
dag_id=ti.dag_id,
856+
dag_run_id=dag_run.id,
857+
run_id=dag_run.run_id,
858+
task_id=mapped_ti.task_id,
859+
dag_id=mapped_ti.dag_id,
850860
map_index=-1,
851861
)
852862
)
853863
session.commit()
854864

855865
response_default = client.post(
856-
f"/execution/xcoms/{ti.dag_id}/{ti.run_id}/{ti.task_id}/keys",
866+
f"/execution/xcoms/{mapped_ti.dag_id}/{dag_run.run_id}/{mapped_ti.task_id}/keys",
857867
json={"keys": ["return_value_0"]},
858868
)
859869
response_indexed = client.post(
860-
f"/execution/xcoms/{ti.dag_id}/{ti.run_id}/{ti.task_id}/keys",
870+
f"/execution/xcoms/{mapped_ti.dag_id}/{dag_run.run_id}/{mapped_ti.task_id}/keys",
861871
params={"map_index": 0},
862872
json={"keys": ["return_value_0"]},
863873
)

0 commit comments

Comments
 (0)