[AIRFLOW-6957] Make retrieving Paused Dag ids a separate method

[kaxil]

Separating it out to a method makes it easy to test "just that" functionality.

---

Issue link: AIRFLOW-6957

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Commit message/PR title starts with [AIRFLOW-NNNN]. AIRFLOW-NNNN = JIRA ID^*
• Unit tests coverage for changes (not needed for documentation changes)
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

^* For document-only changes commit message can start with [AIRFLOW-XXXX].

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[mik-laj]

Can you also add a test that checks if the DagFileProcessor ignores these DAGs? This is the core, and it's worth one more test than not having it.

[kaxil]

Can you also add a test that checks if the DagFileProcessor ignores these DAGs? This is the core, and it's worth one more test than not having it.

We already have other cases covered:

airflow/tests/jobs/test_scheduler_job.py

Lines 847 to 854 in 847cb40

	def test_find_dags_to_run_skip_paused_dags(self):
	dagbag = DagBag(include_examples=False)
	dag = dagbag.get_dag('test_subdag_operator')
	dag_file_processor = DagFileProcessor(dag_ids=[], log=mock.MagicMock())
	dags = dag_file_processor._find_dags_to_process(dagbag.dags.values(), paused_dag_ids=[dag.dag_id])
	self.assertNotIn(dag, dags)

and

airflow/tests/jobs/test_scheduler_job.py

Lines 1162 to 1188 in 847cb40

	def test_execute_task_instances_is_paused_wont_execute(self):
	dag_id = 'SchedulerJobTest.test_execute_task_instances_is_paused_wont_execute'
	task_id_1 = 'dummy_task'
	dag = DAG(dag_id=dag_id, start_date=DEFAULT_DATE)
	task1 = DummyOperator(dag=dag, task_id=task_id_1)
	dagbag = self._make_simple_dag_bag([dag])
	dag_file_processor = DagFileProcessor(dag_ids=[], log=mock.MagicMock())
	scheduler = SchedulerJob()
	session = settings.Session()
	dr1 = dag_file_processor.create_dag_run(dag)
	ti1 = TaskInstance(task1, DEFAULT_DATE)
	ti1.state = State.SCHEDULED
	dr1.state = State.RUNNING
	dagmodel = DagModel()
	dagmodel.dag_id = dag_id
	dagmodel.is_paused = True
	session.merge(ti1)
	session.merge(dr1)
	session.add(dagmodel)
	session.commit()
	scheduler._execute_task_instances(dagbag)
	ti1.refresh_from_db()
	self.assertEqual(State.SCHEDULED, ti1.state)

The bug in the last PR was we were not checking how the results were returned from the query which this PR adds.

[codecov-io]

Codecov Report

❗ No coverage uploaded for pull request base (master@847cb40). Click here to learn what that means.
The diff coverage is 100%.

@@            Coverage Diff            @@
##             master    #7587   +/-   ##
=========================================
  Coverage          ?   86.58%           
=========================================
  Files             ?      896           
  Lines             ?    42624           
  Branches          ?        0           
=========================================
  Hits              ?    36905           
  Misses            ?     5719           
  Partials          ?        0

Impacted Files	Coverage Δ
airflow/jobs/scheduler_job.py	90.46% <100%> (ø)
airflow/models/dag.py	91.57% <100%> (ø)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 847cb40...f094e29. Read the comment docs.