[AIRFLOW-821] Fix py3 compatibility

iteritems() does not exist in py3.

Closes apache#2039 from bolkedebruin/AIRFLOW-821

airflow/jobs.py

@@ -28,6 +28,7 @@
 import multiprocessing
 import os
 import signal
+import six
 import sys
 import threading
 import time
@@ -691,7 +692,7 @@ def update_import_errors(session, dagbag):
             ).delete()
 
         # Add the errors of the processed files
-        for filename, stacktrace in dagbag.import_errors.iteritems():
+        for filename, stacktrace in six.iteritems(dagbag.import_errors):
             session.add(models.ImportError(
                 filename=filename,
                 stacktrace=stacktrace))

tests/jobs.py

@@ -51,8 +51,6 @@
     except ImportError:
         mock = None
 
-IS_PYTHON_3_TRAVIS = sys.version_info >= (3, 0) and "TRAVIS" in os.environ
-
 DEV_NULL = '/dev/null'
 DEFAULT_DATE = datetime.datetime(2016, 1, 1)
 
@@ -1244,8 +1242,6 @@ def test_dag_catchup_option(self):
         # The DR should be scheduled BEFORE now
         self.assertLess(dr.execution_date, datetime.datetime.now())
 
-    @unittest.skipIf(IS_PYTHON_3_TRAVIS,
-                     "Fails in Python 3 on Travis but not reproducible locally")
     def test_add_unparseable_file_before_sched_start_creates_import_error(self):
         try:
             dags_folder = mkdtemp()
@@ -1266,8 +1262,6 @@ def test_add_unparseable_file_before_sched_start_creates_import_error(self):
         self.assertEqual(import_error.stacktrace,
                          "invalid syntax ({}, line 1)".format(TEMP_DAG_FILENAME))
 
-    @unittest.skipIf(IS_PYTHON_3_TRAVIS,
-                     "Fails in Python 3 on Travis but not reproducible locally")
     def test_add_unparseable_file_after_sched_start_creates_import_error(self):
         try:
             dags_folder = mkdtemp()
@@ -1306,8 +1300,6 @@ def test_no_import_errors_with_parseable_dag(self):
 
         self.assertEqual(len(import_errors), 0)
 
-    @unittest.skipIf(IS_PYTHON_3_TRAVIS,
-                     "Fails in Python 3 on Travis but not reproducible locally")
     def test_new_import_error_replaces_old(self):
         try:
             dags_folder = mkdtemp()