You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/06/03 03:36:38 UTC
[GitHub] [airflow] Bowrna opened a new pull request, #23731: remove xfail test jobs in test_scheduler
Bowrna opened a new pull request, #23731:
URL: https://github.com/apache/airflow/pull/23731
relates: #23539
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragement file, named `{pr_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a diff in pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on code in PR #23731:
URL: https://github.com/apache/airflow/pull/23731#discussion_r888897153
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4099,42 +4099,44 @@ def test_catchup_works_correctly(self, dag_maker):
) > (timezone.utcnow() - timedelta(days=2))
-@pytest.mark.xfail(reason="Work out where this goes")
-def test_task_with_upstream_skip_process_task_instances():
+# @pytest.mark.xfail(reason="Work out where this goes")
Review Comment:
```suggestion
```
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4099,42 +4099,44 @@ def test_catchup_works_correctly(self, dag_maker):
) > (timezone.utcnow() - timedelta(days=2))
-@pytest.mark.xfail(reason="Work out where this goes")
-def test_task_with_upstream_skip_process_task_instances():
+# @pytest.mark.xfail(reason="Work out where this goes")
+@pytest.mark.need_serialized_dag
+def test_task_with_upstream_skip_process_task_instances(dag_maker, session):
"""
Test if _process_task_instances puts a task instance into SKIPPED state if any of its
upstream tasks are skipped according to TriggerRuleDep.
"""
Review Comment:
I think both the test name and the doc string need to be changed
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4099,42 +4099,44 @@ def test_catchup_works_correctly(self, dag_maker):
) > (timezone.utcnow() - timedelta(days=2))
-@pytest.mark.xfail(reason="Work out where this goes")
-def test_task_with_upstream_skip_process_task_instances():
+# @pytest.mark.xfail(reason="Work out where this goes")
+@pytest.mark.need_serialized_dag
+def test_task_with_upstream_skip_process_task_instances(dag_maker, session):
"""
Test if _process_task_instances puts a task instance into SKIPPED state if any of its
upstream tasks are skipped according to TriggerRuleDep.
"""
- clear_db_runs()
- with DAG(
- dag_id='test_task_with_upstream_skip_dag', start_date=DEFAULT_DATE, schedule_interval=None
- ) as dag:
+ with dag_maker(
+ dag_id='test_task_with_upstream_skip_process_task_instances',
+ start_date=DEFAULT_DATE,
+ session=session,
+ ):
dummy1 = EmptyOperator(task_id='dummy1')
dummy2 = EmptyOperator(task_id="dummy2")
dummy3 = EmptyOperator(task_id="dummy3")
[dummy1, dummy2] >> dummy3
-
# dag_file_processor = DagFileProcessor(dag_ids=[], log=mock.MagicMock())
- dag.clear()
- dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
+ dr = dag_maker.create_dagrun(state=State.RUNNING)
assert dr is not None
- with create_session() as session:
- tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
- # Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
- tis[dummy1.task_id].state = State.SKIPPED
- tis[dummy2.task_id].state = State.SUCCESS
- assert tis[dummy3.task_id].state == State.NONE
+ tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
+ # Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
+ tis[dummy1.task_id].state = State.SKIPPED
+ tis[dummy2.task_id].state = State.SUCCESS
+ assert tis[dummy3.task_id].state == State.NONE
+ session.flush()
+ # dr.refresh_from_db(session=session)
Review Comment:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a diff in pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
potiuk commented on code in PR #23731:
URL: https://github.com/apache/airflow/pull/23731#discussion_r877427392
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4120,22 +4120,40 @@ def test_task_with_upstream_skip_process_task_instances():
dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
assert dr is not None
+ # def get_ti_from_db(task):
+ # return (
+ # session.query(TaskInstance)
+ # .filter(
+ # TaskInstance.dag_id == dag.dag_id,
+ # TaskInstance.task_id == task.task_id,
+ # TaskInstance.run_id == dr.run_id,
+ # )
+ # .one()
+ # )
+
+ # with create_session() as session:
+ # get_ti_from_db(dummy1).state = State.SKIPPED
+ # get_ti_from_db(dummy2).state = State.SUCCESS
+ # session.flush()
+
with create_session() as session:
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
# Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
tis[dummy1.task_id].state = State.SKIPPED
tis[dummy2.task_id].state = State.SUCCESS
assert tis[dummy3.task_id].state == State.NONE
+ session.commit()
+ # dr.refresh_from_db(session=session)
# dag_runs = DagRun.find(dag_id='test_task_with_upstream_skip_dag')
Review Comment:
not really sure, I am afraid - those are pretty deep dive tests and they might require qutie a bit deeper knowledge (and time to look it up).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23731:
URL: https://github.com/apache/airflow/pull/23731#issuecomment-1144047637
> I'm perfectly fine with you taking this. I allready have few open PRs, that I'm struggling to get reviewed. I do not want another :)
@tanelk - doing my best to catch-up after the Summit, but if you think there is a delay with reviews feel free to ping me or any of the committers who already commented on your PRs :).
And you can watch my talk from the Summit which explains why sometimes those can heppen https://www.crowdcast.io/e/airflowsummit2022/53
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna closed pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
Bowrna closed pull request #23731: remove xfail test jobs in test_scheduler
URL: https://github.com/apache/airflow/pull/23731
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna closed pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
Bowrna closed pull request #23731: remove xfail test jobs in test_scheduler
URL: https://github.com/apache/airflow/pull/23731
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna commented on pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
Bowrna commented on PR #23731:
URL: https://github.com/apache/airflow/pull/23731#issuecomment-1146791349
Suggested review changes are done. If there is any other change that I have to add or work on, please tell me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna commented on a diff in pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
Bowrna commented on code in PR #23731:
URL: https://github.com/apache/airflow/pull/23731#discussion_r885193547
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4120,22 +4120,40 @@ def test_task_with_upstream_skip_process_task_instances():
dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
assert dr is not None
+ # def get_ti_from_db(task):
+ # return (
+ # session.query(TaskInstance)
+ # .filter(
+ # TaskInstance.dag_id == dag.dag_id,
+ # TaskInstance.task_id == task.task_id,
+ # TaskInstance.run_id == dr.run_id,
+ # )
+ # .one()
+ # )
+
+ # with create_session() as session:
+ # get_ti_from_db(dummy1).state = State.SKIPPED
+ # get_ti_from_db(dummy2).state = State.SUCCESS
+ # session.flush()
+
with create_session() as session:
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
# Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
tis[dummy1.task_id].state = State.SKIPPED
tis[dummy2.task_id].state = State.SUCCESS
assert tis[dummy3.task_id].state == State.NONE
+ session.commit()
+ # dr.refresh_from_db(session=session)
# dag_runs = DagRun.find(dag_id='test_task_with_upstream_skip_dag')
Review Comment:
I see. I think I picked up the wrong issue to fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tanelk commented on pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
tanelk commented on PR #23731:
URL: https://github.com/apache/airflow/pull/23731#issuecomment-1141678437
```diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py
index fd32e6dd7..fee991995 100644
--- a/tests/jobs/test_scheduler_job.py
+++ b/tests/jobs/test_scheduler_job.py
@@ -4099,37 +4099,35 @@ class TestSchedulerJob:
) > (timezone.utcnow() - timedelta(days=2))
-@pytest.mark.xfail(reason="Work out where this goes")
-def test_task_with_upstream_skip_process_task_instances():
- """
- Test if _process_task_instances puts a task instance into SKIPPED state if any of its
- upstream tasks are skipped according to TriggerRuleDep.
- """
- clear_db_runs()
- with DAG(
- dag_id='test_task_with_upstream_skip_dag', start_date=DEFAULT_DATE, schedule_interval=None
- ) as dag:
- dummy1 = EmptyOperator(task_id='dummy1')
- dummy2 = EmptyOperator(task_id="dummy2")
- dummy3 = EmptyOperator(task_id="dummy3")
- [dummy1, dummy2] >> dummy3
-
- # dag_file_processor = DagFileProcessor(dag_ids=[], log=mock.MagicMock())
- dag.clear()
- dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
- assert dr is not None
-
- with create_session() as session:
+ def test_task_with_upstream_skip_process_task_instances(self, dag_maker, session):
+ """
+ Test if _process_task_instances puts a task instance into SKIPPED state if any of its
+ upstream tasks are skipped according to TriggerRuleDep.
+ """
+ with dag_maker(
+ dag_id='test_task_with_upstream_skip_process_task_instances',
+ start_date=DEFAULT_DATE,
+ session=session,
+ ):
+ dummy1 = EmptyOperator(task_id='dummy1')
+ dummy2 = EmptyOperator(task_id="dummy2")
+ dummy3 = EmptyOperator(task_id="dummy3")
+ [dummy1, dummy2] >> dummy3
+
+ dr = dag_maker.create_dagrun(state=State.RUNNING)
+
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
# Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
tis[dummy1.task_id].state = State.SKIPPED
tis[dummy2.task_id].state = State.SUCCESS
assert tis[dummy3.task_id].state == State.NONE
+ session.flush()
- # dag_runs = DagRun.find(dag_id='test_task_with_upstream_skip_dag')
- # dag_file_processor._process_task_instances(dag, dag_runs=dag_runs)
+ self.scheduler_job = SchedulerJob(subdir=os.devnull)
+ self.scheduler_job._schedule_dag_run(dr, session)
+ session.flush()
- with create_session() as session:
+ session.refresh(dr)
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
assert tis[dummy1.task_id].state == State.SKIPPED
assert tis[dummy2.task_id].state == State.SUCCESS
```
The test is from pre 2.0 era. The method, that was supposed to set the TI to skiped (`dag_file_processor._process_task_instances`) was moved to scheduler job.
The test should pass now, but I'm not sure if it belongs here or perhaps this is covered by some other tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna commented on a diff in pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
Bowrna commented on code in PR #23731:
URL: https://github.com/apache/airflow/pull/23731#discussion_r878657728
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4120,22 +4120,40 @@ def test_task_with_upstream_skip_process_task_instances():
dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
assert dr is not None
+ # def get_ti_from_db(task):
+ # return (
+ # session.query(TaskInstance)
+ # .filter(
+ # TaskInstance.dag_id == dag.dag_id,
+ # TaskInstance.task_id == task.task_id,
+ # TaskInstance.run_id == dr.run_id,
+ # )
+ # .one()
+ # )
+
+ # with create_session() as session:
+ # get_ti_from_db(dummy1).state = State.SKIPPED
+ # get_ti_from_db(dummy2).state = State.SUCCESS
+ # session.flush()
+
with create_session() as session:
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
# Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
tis[dummy1.task_id].state = State.SKIPPED
tis[dummy2.task_id].state = State.SUCCESS
assert tis[dummy3.task_id].state == State.NONE
+ session.commit()
+ # dr.refresh_from_db(session=session)
# dag_runs = DagRun.find(dag_id='test_task_with_upstream_skip_dag')
Review Comment:
yes i took time to look but I am confused on this part. Let me try again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a diff in pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
potiuk commented on code in PR #23731:
URL: https://github.com/apache/airflow/pull/23731#discussion_r884768873
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4120,22 +4120,40 @@ def test_task_with_upstream_skip_process_task_instances():
dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
assert dr is not None
+ # def get_ti_from_db(task):
+ # return (
+ # session.query(TaskInstance)
+ # .filter(
+ # TaskInstance.dag_id == dag.dag_id,
+ # TaskInstance.task_id == task.task_id,
+ # TaskInstance.run_id == dr.run_id,
+ # )
+ # .one()
+ # )
+
+ # with create_session() as session:
+ # get_ti_from_db(dummy1).state = State.SKIPPED
+ # get_ti_from_db(dummy2).state = State.SUCCESS
+ # session.flush()
+
with create_session() as session:
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
# Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
tis[dummy1.task_id].state = State.SKIPPED
tis[dummy2.task_id].state = State.SUCCESS
assert tis[dummy3.task_id].state == State.NONE
+ session.commit()
+ # dr.refresh_from_db(session=session)
# dag_runs = DagRun.find(dag_id='test_task_with_upstream_skip_dag')
Review Comment:
You can try :) but likely in this case there are very few people who know it deeply enough to be able to help (and likely they will have to simply solve it rather than help :) - @ephraimbuddy , myself, @ashb, @uranusjr are the likely candidates :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23731:
URL: https://github.com/apache/airflow/pull/23731#issuecomment-1146814976
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy merged pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
ephraimbuddy merged PR #23731:
URL: https://github.com/apache/airflow/pull/23731
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna commented on pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
Bowrna commented on PR #23731:
URL: https://github.com/apache/airflow/pull/23731#issuecomment-1143387290
> The test is from pre 2.0 era. The method, that was supposed to set the TI to skiped (`dag_file_processor._process_task_instances`) was moved to scheduler job.
>
> The test should pass now, but I'm not sure if it belongs here or perhaps this is covered by some other tests
@tanelk Test that you suggested is working and it works as expected. The downstream task is set as Skipped. I think you should raise this PR as it is the changes you have suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna commented on a diff in pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
Bowrna commented on code in PR #23731:
URL: https://github.com/apache/airflow/pull/23731#discussion_r884522191
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4120,22 +4120,40 @@ def test_task_with_upstream_skip_process_task_instances():
dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
assert dr is not None
+ # def get_ti_from_db(task):
+ # return (
+ # session.query(TaskInstance)
+ # .filter(
+ # TaskInstance.dag_id == dag.dag_id,
+ # TaskInstance.task_id == task.task_id,
+ # TaskInstance.run_id == dr.run_id,
+ # )
+ # .one()
+ # )
+
+ # with create_session() as session:
+ # get_ti_from_db(dummy1).state = State.SKIPPED
+ # get_ti_from_db(dummy2).state = State.SUCCESS
+ # session.flush()
+
with create_session() as session:
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
# Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
tis[dummy1.task_id].state = State.SKIPPED
tis[dummy2.task_id].state = State.SUCCESS
assert tis[dummy3.task_id].state == State.NONE
+ session.commit()
+ # dr.refresh_from_db(session=session)
# dag_runs = DagRun.find(dag_id='test_task_with_upstream_skip_dag')
Review Comment:
@potiuk Could i post this in slack and see if i could get some help in fixing this issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tanelk commented on pull request #23731: remove xfail test jobs in test_scheduler
Posted by GitBox <gi...@apache.org>.
tanelk commented on PR #23731:
URL: https://github.com/apache/airflow/pull/23731#issuecomment-1143844588
> @tanelk Test that you suggested is working and it works as expected. The downstream task is set as Skipped. I think you should raise this PR as it is the changes you have suggested.
I'm perfectly fine with you taking this. I allready have few open PRs, that I'm struggling to get reviewed. I do not want another :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna commented on a diff in pull request #23731: committing to start discussion
Posted by GitBox <gi...@apache.org>.
Bowrna commented on code in PR #23731:
URL: https://github.com/apache/airflow/pull/23731#discussion_r873989382
##########
tests/jobs/test_scheduler_job.py:
##########
@@ -4120,22 +4120,40 @@ def test_task_with_upstream_skip_process_task_instances():
dr = dag.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING, execution_date=DEFAULT_DATE)
assert dr is not None
+ # def get_ti_from_db(task):
+ # return (
+ # session.query(TaskInstance)
+ # .filter(
+ # TaskInstance.dag_id == dag.dag_id,
+ # TaskInstance.task_id == task.task_id,
+ # TaskInstance.run_id == dr.run_id,
+ # )
+ # .one()
+ # )
+
+ # with create_session() as session:
+ # get_ti_from_db(dummy1).state = State.SKIPPED
+ # get_ti_from_db(dummy2).state = State.SUCCESS
+ # session.flush()
+
with create_session() as session:
tis = {ti.task_id: ti for ti in dr.get_task_instances(session=session)}
# Set dummy1 to skipped and dummy2 to success. dummy3 remains as none.
tis[dummy1.task_id].state = State.SKIPPED
tis[dummy2.task_id].state = State.SUCCESS
assert tis[dummy3.task_id].state == State.NONE
+ session.commit()
+ # dr.refresh_from_db(session=session)
# dag_runs = DagRun.find(dag_id='test_task_with_upstream_skip_dag')
Review Comment:
@potiuk
I tried to get rid of xfail test in `tests/jobs/test_scheduler_job.py::test_task_with_upstream_skip_process_task_instances` but in this function the comment says it expects dummy3 to be skipped. But it is returned as None. Can you tell me the expected behaviour?
I tried to check similar place in code where they try to skip task and find out the state of downstream task but I couldn't find many.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org