[AIRFLOW-328][AIRFLOW-371] Remove redundant default configuration & f…

…ix unit test configuration

AIRFLOW-328
https://issues.apache.org/jira/browse/AIRFLOW-328
Previously, Airflow had both a default template for airflow.cfg AND a
dictionary of default values. Frequently, these get out of sync (an
option in one has a different value than in the other, or isn’t present
in the other). This commit removes the default dict and uses the
airflow.cfg template to provide defaults. The ConfigParser first reads
the template, loading all the options it contains, and then reads the
user’s actual airflow.cfg to overwrite the default values with any new
ones.

AIRFLOW-371
https://issues.apache.org/jira/browse/AIRFLOW-371
Calling test_mode() didn't actually change Airflow's configuration! This actually wasn't an issue in unit tests because the unit test run script was hardcoded to point at the unittest.cfg file, but it needed to be fixed.

[AIRFLOW-328] Remove redundant default configuration

Previously, Airflow had both a default template
for airflow.cfg AND a dictionary of default
values. Frequently, these get out of sync (an
option in one has a different value than in the
other, or isn’t present in the other). This commit
removes the default dict and uses the airflow.cfg
template to provide defaults. The ConfigParser
first reads the template, loading all the options
it contains, and then reads the user’s actual
airflow.cfg to overwrite the default values with
any new ones.

[AIRFLOW-371] Make test_mode() functional

Previously, calling test_mode() didn’t actually
do anything.

This PR renames it to load_test_config() (to
avoid confusion, ht @r39132).

In addition, manually entering test_mode after
Airflow launches might be too late — some
options have already been loaded (DAGS_FOLDER,
etc.). This makes it so setting
tests/unit_test_mode OR the equivalent env var
(AIRFLOW__TESTS__UNIT_TEST_MODE) will load the
test config immediately, prior to loading the
rest of Airflow.

Closes apache#1677 from jlowin/Simplify-config

docs/configuration.rst

@@ -4,6 +4,8 @@ Configuration
 Setting up the sandbox in the :doc:`start` section was easy;
 building a production-grade environment requires a bit more work!
 
+.. _setting-options:
+
 Setting Configuration Options
 '''''''''''''''''''''''''''''
 
@@ -228,3 +230,20 @@ integrated with upstart
 .. code-block:: bash
 
     initctl airflow-webserver status
+
+Test Mode
+'''''''''
+Airflow has a fixed set of "test mode" configuration options. You can load these
+at any time by calling ``airflow.configuration.load_test_config()`` (note this
+operation is not reversible!). However, some options (like the DAG_FOLDER) are
+loaded before you have a chance to call load_test_config(). In order to eagerly load
+the test configuration, set test_mode in airflow.cfg:
+
+.. code-block:: bash
+
+  [tests]
+  unit_test_mode = True
+
+Due to Airflow's automatic environment variable expansion (see :ref:`setting-options`),
+you can also set the env var ``AIRFLOW__CORE__UNIT_TEST_MODE`` to temporarily overwrite
+airflow.cfg.

run_unit_tests.sh

@@ -16,7 +16,7 @@
 
 # environment
 export AIRFLOW_HOME=${AIRFLOW_HOME:=~/airflow}
-export AIRFLOW_CONFIG=$AIRFLOW_HOME/unittests.cfg
+export AIRFLOW__CORE__UNIT_TEST_MODE=True
 
 # configuration test
 export AIRFLOW__TESTSECTION__TESTKEY=testvalue

tests/configuration.py

@@ -19,11 +19,10 @@
 from airflow import configuration
 from airflow.configuration import conf
 
-configuration.test_mode()
-
 class ConfTest(unittest.TestCase):
+
     def setup(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
     def test_env_var_config(self):
         opt = conf.get('testsection', 'testkey')
@@ -38,15 +37,10 @@ def test_conf_as_dict(self):
         # test env vars
         self.assertEqual(cfg_dict['testsection']['testkey'], '< hidden >')
 
-        # test defaults
-        conf.remove_option('core', 'load_examples')
-        cfg_dict = conf.as_dict()
-        self.assertEqual(cfg_dict['core']['load_examples'], 'True')
-
         # test display_source
         cfg_dict = conf.as_dict(display_source=True)
-        self.assertEqual(cfg_dict['core']['unit_test_mode'][1], 'airflow.cfg')
-        self.assertEqual(cfg_dict['core']['load_examples'][1], 'default')
+        self.assertEqual(
+            cfg_dict['core']['load_examples'][1], 'airflow config')
         self.assertEqual(
             cfg_dict['testsection']['testkey'], ('< hidden >', 'env var'))
 

tests/contrib/hooks/aws_hook.py

@@ -29,7 +29,7 @@
 class TestAwsHook(unittest.TestCase):
     @mock_emr
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
     @unittest.skipIf(mock_emr is None, 'mock_emr package not present')
     @mock_emr

tests/contrib/hooks/emr_hook.py

@@ -29,7 +29,7 @@
 class TestEmrHook(unittest.TestCase):
     @mock_emr
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
     @unittest.skipIf(mock_emr is None, 'mock_emr package not present')
     @mock_emr

tests/contrib/operators/emr_add_steps_operator.py

@@ -28,7 +28,7 @@
 
 class TestEmrAddStepsOperator(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
         # Mock out the emr_client (moto has incorrect response)
         mock_emr_client = MagicMock()

tests/contrib/operators/emr_create_job_flow_operator.py

@@ -28,7 +28,7 @@
 
 class TestEmrCreateJobFlowOperator(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
         # Mock out the emr_client (moto has incorrect response)
         mock_emr_client = MagicMock()

tests/contrib/operators/emr_terminate_job_flow_operator.py

@@ -27,7 +27,7 @@
 
 class TestEmrTerminateJobFlowOperator(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
         # Mock out the emr_client (moto has incorrect response)
         mock_emr_client = MagicMock()

tests/contrib/operators/fs_operator.py

@@ -23,7 +23,7 @@
 
 TEST_DAG_ID = 'unit_tests'
 DEFAULT_DATE = datetime(2015, 1, 1)
-configuration.test_mode()
+configuration.load_test_config()
 
 
 def reset(dag_id=TEST_DAG_ID):
@@ -37,7 +37,7 @@ def reset(dag_id=TEST_DAG_ID):
 
 class FileSensorTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         from airflow.contrib.hooks.fs_hook import FSHook
         hook = FSHook()
         args = {

tests/contrib/operators/hipchat_operator.py

@@ -31,7 +31,7 @@
 
 class HipChatOperatorTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
     @unittest.skipIf(mock is None, 'mock package not present')
     @mock.patch('requests.request')

tests/contrib/operators/ssh_execute_operator.py

@@ -23,7 +23,7 @@
 
 TEST_DAG_ID = 'unit_tests'
 DEFAULT_DATE = datetime(2015, 1, 1)
-configuration.test_mode()
+configuration.load_test_config()
 
 
 def reset(dag_id=TEST_DAG_ID):
@@ -38,7 +38,7 @@ def reset(dag_id=TEST_DAG_ID):
 
 class SSHExecuteOperatorTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         from airflow.contrib.hooks.ssh_hook import SSHHook
         hook = SSHHook()
         hook.no_host_key_check = True

tests/contrib/sensors/emr_base_sensor.py

@@ -21,7 +21,7 @@
 
 class TestEmrBaseSensor(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
     def test_subclasses_that_implment_required_methods_and_constants_succeed_when_response_is_good(self):
         class EmrBaseSensorSubclass(EmrBaseSensor):

tests/contrib/sensors/emr_job_flow_sensor.py

@@ -87,7 +87,7 @@
 
 class TestEmrJobFlowSensor(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
         # Mock out the emr_client (moto has incorrect response)
         self.mock_emr_client = MagicMock()

tests/contrib/sensors/emr_step_sensor.py

@@ -82,7 +82,7 @@
 
 class TestEmrStepSensor(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
         # Mock out the emr_client (moto has incorrect response)
         self.mock_emr_client = MagicMock()

tests/core.py

@@ -35,7 +35,7 @@
 from airflow.executors import SequentialExecutor, LocalExecutor
 from airflow.models import Variable
 
-configuration.test_mode()
+configuration.load_test_config()
 from airflow import jobs, models, DAG, utils, macros, settings, exceptions
 from airflow.models import BaseOperator
 from airflow.operators.bash_operator import BashOperator
@@ -116,7 +116,7 @@ class CoreTest(unittest.TestCase):
                               "num_runs": 1}
 
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         self.dagbag = models.DagBag(
             dag_folder=DEV_NULL, include_examples=True)
         self.args = {'owner': 'airflow', 'start_date': DEFAULT_DATE}
@@ -836,7 +836,7 @@ def test_task_fail_duration(self):
 
 class CliTests(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         app = application.create_app()
         app.config['TESTING'] = True
         self.parser = cli.CLIFactory.get_parser()
@@ -1003,7 +1003,7 @@ def test_variables(self):
 
 class WebUiTests(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         configuration.conf.set("webserver", "authenticate", "False")
         app = application.create_app()
         app.config['TESTING'] = True
@@ -1226,7 +1226,7 @@ def test_unauthorized_password_auth(self):
         self.assertEqual(response.status_code, 302)
 
     def tearDown(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         session = Session()
         session.query(models.User).delete()
         session.commit()
@@ -1310,7 +1310,7 @@ def test_with_filters(self):
         assert 'Connections' in response.data.decode('utf-8')
 
     def tearDown(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         session = Session()
         session.query(models.User).delete()
         session.commit()
@@ -1346,7 +1346,7 @@ def test_group_belonging(self):
             assert set(auth.ldap_groups) == set(users[user])
 
     def tearDown(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         configuration.conf.set("webserver", "authenticate", "False")
 
 
@@ -1365,7 +1365,7 @@ def prepare_request(self, request):
 
 class HttpOpSensorTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         args = {'owner': 'airflow', 'start_date': DEFAULT_DATE_ISO}
         dag = DAG(TEST_DAG_ID, default_args=args)
         self.dag = dag
@@ -1419,7 +1419,7 @@ def check_for_path(self, hdfs_path):
 
 class ConnectionTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         utils.db.initdb()
         os.environ['AIRFLOW_CONN_TEST_URI'] = (
             'postgres://username:[email protected]:5432/the_database')
@@ -1475,7 +1475,7 @@ def test_env_var_priority(self):
 
 class WebHDFSHookTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
 
     def test_simple_init(self):
         from airflow.hooks.webhdfs_hook import WebHDFSHook
@@ -1498,7 +1498,7 @@ def test_init_proxy_user(self):
                  "Skipping test because S3Hook is not installed")
 class S3HookTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         self.s3_test_url = "s3://test/this/is/not/a-real-key.txt"
 
     def test_parse_s3_url(self):
@@ -1523,7 +1523,7 @@ def test_parse_s3_url(self):
 
 class SSHHookTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         from airflow.contrib.hooks.ssh_hook import SSHHook
         self.hook = SSHHook()
         self.hook.no_host_key_check = True

tests/jobs.py

@@ -37,7 +37,7 @@
 from tests.executor.test_executor import TestExecutor
 
 from airflow import configuration
-configuration.test_mode()
+configuration.load_test_config()
 
 try:
     from unittest import mock

tests/operators/hive_operator.py

@@ -20,7 +20,7 @@
 import six
 
 from airflow import DAG, configuration, operators, utils
-configuration.test_mode()
+configuration.load_test_config()
 
 import os
 import unittest
@@ -39,7 +39,7 @@
 
     class HiveServer2Test(unittest.TestCase):
         def setUp(self):
-            configuration.test_mode()
+            configuration.load_test_config()
 
         def test_select_conn(self):
             from airflow.hooks.hive_hooks import HiveServer2Hook
@@ -71,7 +71,7 @@ def test_to_csv(self):
     class HivePrestoTest(unittest.TestCase):
 
         def setUp(self):
-            configuration.test_mode()
+            configuration.load_test_config()
             args = {'owner': 'airflow', 'start_date': DEFAULT_DATE}
             dag = DAG('test_dag_id', default_args=args)
             self.dag = dag

tests/operators/operators.py

@@ -21,7 +21,7 @@
 
 from airflow import DAG, configuration, operators, utils
 from airflow.utils.tests import skipUnlessImported
-configuration.test_mode()
+configuration.load_test_config()
 
 import os
 import unittest
@@ -36,7 +36,7 @@
 @skipUnlessImported('airflow.operators.mysql_operator', 'MySqlOperator')
 class MySqlTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         args = {
             'owner': 'airflow',
             'mysql_conn_id': 'airflow_db',
@@ -99,7 +99,7 @@ def test_sql_sensor(self):
 @skipUnlessImported('airflow.operators.postgres_operator', 'PostgresOperator')
 class PostgresTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         args = {'owner': 'airflow', 'start_date': DEFAULT_DATE}
         dag = DAG(TEST_DAG_ID, default_args=args)
         self.dag = dag
@@ -166,7 +166,7 @@ class TransferTests(unittest.TestCase):
     cluster = None
 
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         args = {'owner': 'airflow', 'start_date': DEFAULT_DATE}
         dag = DAG(TEST_DAG_ID, default_args=args)
         self.dag = dag

tests/operators/sensors.py

@@ -26,7 +26,7 @@
 from airflow.exceptions import (AirflowException,
                                 AirflowSensorTimeout,
                                 AirflowSkipException)
-configuration.test_mode()
+configuration.load_test_config()
 
 DEFAULT_DATE = datetime(2015, 1, 1)
 TEST_DAG_ID = 'unit_test_dag'
@@ -69,7 +69,7 @@ def execute(self, context):
 
 class SensorTimeoutTest(unittest.TestCase):
     def setUp(self):
-        configuration.test_mode()
+        configuration.load_test_config()
         args = {
             'owner': 'airflow',
             'start_date': DEFAULT_DATE