[AIRFLOW-4016] Clear runs for BackfillJobTest (apache#4839)

* Clear runs for BackfillJobTest

* Fixing import

* Fixing flake8

tests/test_jobs.py

@@ -59,6 +59,8 @@
 from airflow.utils.net import get_hostname
 from airflow.utils.state import State
 from airflow.utils.timeout import timeout
+from tests.test_utils.db import clear_db_runs, clear_db_pools, clear_db_dags, \
+    clear_db_sla_miss, clear_db_errors
 from tests.core import TEST_DAG_FOLDER
 from tests.executors.test_executor import TestExecutor
 from tests.compat import mock
@@ -126,10 +128,8 @@ def abort():
 class BackfillJobTest(unittest.TestCase):
 
     def setUp(self):
-        with create_session() as session:
-            session.query(models.DagRun).delete()
-            session.query(models.Pool).delete()
-            session.query(models.TaskInstance).delete()
+        clear_db_runs()
+        clear_db_pools()
 
         self.parser = cli.CLIFactory.get_parser()
         self.dagbag = DagBag(include_examples=True)
@@ -1225,7 +1225,7 @@ def test_backfill_run_backwards(self):
 
 class LocalTaskJobTest(unittest.TestCase):
     def setUp(self):
-        pass
+        clear_db_runs()
 
     def test_localtaskjob_essential_attr(self):
         """
@@ -1385,14 +1385,11 @@ def test_localtaskjob_double_trigger(self):
 class SchedulerJobTest(unittest.TestCase):
 
     def setUp(self):
-        with create_session() as session:
-            session.query(models.DagRun).delete()
-            session.query(models.TaskInstance).delete()
-            session.query(models.Pool).delete()
-            session.query(models.DagModel).delete()
-            session.query(models.SlaMiss).delete()
-            session.query(errors.ImportError).delete()
-            session.commit()
+        clear_db_runs()
+        clear_db_pools()
+        clear_db_dags()
+        clear_db_sla_miss()
+        clear_db_errors()
 
     @classmethod
     def setUpClass(cls):

tests/test_utils/db.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from airflow.models import DagRun, TaskInstance, Pool, DagModel, errors
+from airflow.models.slamiss import SlaMiss
+from airflow.utils.db import create_session
+
+
+def clear_db_runs():
+    with create_session() as session:
+        session.query(DagRun).delete()
+        session.query(TaskInstance).delete()
+
+
+def clear_db_dags():
+    with create_session() as session:
+        session.query(DagModel).delete()
+
+
+def clear_db_sla_miss():
+    with create_session() as session:
+        session.query(SlaMiss).delete()
+
+
+def clear_db_errors():
+    with create_session() as session:
+        session.query(errors.ImportError).delete()
+
+
+def clear_db_pools():
+    with create_session() as session:
+        session.query(Pool).delete()