Move some tests to quarantine (apache#8511)

.github/workflows/ci.yml

@@ -256,6 +256,39 @@ jobs:
       - name: "Tests"
         run: ./scripts/ci/ci_run_airflow_testing.sh
 
+  tests-quarantined:
+    timeout-minutes: 30
+    name: "${{matrix.test-type}}:Pg${{matrix.postgres-version}},Py${{matrix.python-version}}"
+    runs-on: ubuntu-latest
+    continue-on-error: true
+    needs: [static-checks-1, static-checks-2, pyfiles]
+    strategy:
+      matrix:
+        python-version: [3.6]
+        postgres-version: [9.6]
+        test-type: [Quarantined]
+      fail-fast: false
+    env:
+      BACKEND: postgres
+      PYTHON_MAJOR_MINOR_VERSION: ${{ matrix.python-version }}
+      POSTGRES_VERSION: ${{ matrix.postgres-version }}
+      RUN_TESTS: "true"
+      CI_JOB_TYPE: "Tests"
+      TEST_TYPE: ${{ matrix.test-type }}
+    # For pull requests only run tests when python files changed
+    if: needs.pyfiles.outputs.count != '0' || github.event_name != 'pull_request'
+    steps:
+      - uses: actions/checkout@master
+      - uses: actions/setup-python@v1
+        with:
+          python-version: '3.x'
+      - name: "Free space"
+        run: ./scripts/ci/ci_free_space_on_ci.sh
+      - name: "Build CI image ${{ matrix.python-version }}"
+        run: ./scripts/ci/ci_prepare_image_on_ci.sh
+      - name: "Tests"
+        run: ./scripts/ci/ci_run_airflow_testing.sh
+
   requirements:
     timeout-minutes: 20
     name: "Requirements: Py${{ matrix.python-version }}"

TESTING.rst

@@ -313,6 +313,17 @@ Some of the tests rung for a long time. Such tests are marked with ``@pytest.mar
 Those tests are skipped by default. You can enable them with ``--include-long-running`` flag. You
 can also decide to only run tests with ``-m long-running`` flags to run only those tests.
 
+Quarantined tests
+-----------------
+
+Some of our tests are quarantined. This means that this test will be run in isolation and that it will be
+re-run several times. Also when quarantined tests fail, the whole test suite will not fail. The quarantined
+tests are usually flaky tests that need some attention and fix.
+
+Those tests are marked with ``@pytest.mark.quarantined`` annotation.
+Those tests are skipped by default. You can enable them with ``--include-quarantined`` flag. You
+can also decide to only run tests with ``-m quarantined`` flag to run only those tests.
+
 Running Tests with Kubernetes
 -----------------------------
 

requirements/requirements-python3.6.txt

@@ -37,6 +37,7 @@ alembic==1.4.2
 amqp==2.5.2
 analytics-python==1.2.9
 ansiwrap==0.8.4
+apipkg==1.5
 apispec==1.3.3
 appdirs==1.4.3
 argcomplete==1.11.1
@@ -108,6 +109,7 @@ elasticsearch-dbapi==0.1.0
 elasticsearch-dsl==7.1.0
 elasticsearch==7.5.1
 entrypoints==0.3
+execnet==1.7.1
 facebook-business==6.0.3
 fastavro==0.23.2
 filelock==3.0.12
@@ -266,7 +268,11 @@ pypd==1.1.0
 pyrsistent==0.16.0
 pysftp==0.2.9
 pytest-cov==2.8.1
+pytest-forked==1.1.3
 pytest-instafail==0.4.1.post0
+pytest-rerunfailures==9.0
+pytest-timeout==1.3.4
+pytest-xdist==1.31.0
 pytest==5.4.1
 python-daemon==2.1.2
 python-dateutil==2.8.1

requirements/requirements-python3.7.txt

@@ -37,6 +37,7 @@ alembic==1.4.2
 amqp==2.5.2
 analytics-python==1.2.9
 ansiwrap==0.8.4
+apipkg==1.5
 apispec==1.3.3
 appdirs==1.4.3
 argcomplete==1.11.1
@@ -108,6 +109,7 @@ elasticsearch-dbapi==0.1.0
 elasticsearch-dsl==7.1.0
 elasticsearch==7.5.1
 entrypoints==0.3
+execnet==1.7.1
 facebook-business==6.0.3
 fastavro==0.23.2
 filelock==3.0.12
@@ -264,7 +266,11 @@ pypd==1.1.0
 pyrsistent==0.16.0
 pysftp==0.2.9
 pytest-cov==2.8.1
+pytest-forked==1.1.3
 pytest-instafail==0.4.1.post0
+pytest-rerunfailures==9.0
+pytest-timeout==1.3.4
+pytest-xdist==1.31.0
 pytest==5.4.1
 python-daemon==2.1.2
 python-dateutil==2.8.1

requirements/setup-3.6.md5

@@ -1 +1 @@
-0f80651f705442f63ed62481b95d421f  /opt/airflow/setup.py
+90fcd0e330052de1268264634e0d4035  /opt/airflow/setup.py

requirements/setup-3.7.md5

@@ -1 +1 @@
-0f80651f705442f63ed62481b95d421f  /opt/airflow/setup.py
+90fcd0e330052de1268264634e0d4035  /opt/airflow/setup.py

scripts/ci/ci_run_airflow_testing.sh

@@ -76,6 +76,10 @@ ENABLED_INTEGRATIONS=${ENABLED_INTEGRATIONS:=""}
 if [[ ${TEST_TYPE:=} == "Integration" ]]; then
     export ENABLED_INTEGRATIONS="${AVAILABLE_INTEGRATIONS}"
     export RUN_INTEGRATION_TESTS="${AVAILABLE_INTEGRATIONS}"
+elif [[ ${TEST_TYPE:=} == "Long" ]]; then
+    export ONLY_RUN_LONG_RUNNING_TESTS="true"
+elif [[ ${TEST_TYPE:=} == "Quarantined" ]]; then
+    export ONLY_RUN_QUARANTINED_TESTS="true"
 fi
 
 for _INT in ${ENABLED_INTEGRATIONS}

scripts/ci/docker-compose/base.yml

@@ -46,6 +46,8 @@ services:
       - ENABLE_KIND_CLUSTER
       - ENABLED_INTEGRATIONS
       - RUN_INTEGRATION_TESTS
+      - ONLY_RUN_LONG_RUNNING_TESTS
+      - ONLY_RUN_QUARANTINED_TESTS
       - BREEZE
       - INSTALL_AIRFLOW_VERSION
       - DB_RESET

setup.py

@@ -446,6 +446,9 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
     'pytest',
     'pytest-cov',
     'pytest-instafail',
+    'pytest-rerunfailures',
+    'pytest-timeout',
+    'pytest-xdist',
     'pywinrm',
     'qds-sdk>=1.9.6',
     'requests_mock',

tests/cli/commands/test_dag_command.py

@@ -24,6 +24,7 @@
 from datetime import datetime, time, timedelta
 
 import mock
+import pytest
 import pytz
 
 from airflow import settings
@@ -237,6 +238,7 @@ def test_cli_backfill_depends_on_past_backwards(self, mock_run):
             verbose=False,
         )
 
+    @pytest.mark.quarantined
     def test_next_execution(self):
         # A scaffolding function
         def reset_dr_db(dag_id):
@@ -360,6 +362,7 @@ def test_pause(self):
         dag_command.dag_unpause(args)
         self.assertIn(self.dagbag.dags['example_bash_operator'].get_is_paused(), [False, 0])
 
+    @pytest.mark.quarantined
     def test_trigger_dag(self):
         dag_command.dag_trigger(self.parser.parse_args([
             'dags', 'trigger', 'example_bash_operator',

tests/cli/commands/test_task_command.py

@@ -22,6 +22,7 @@
 from datetime import datetime, timedelta
 from unittest import mock
 
+import pytest
 from parameterized import parameterized
 from tabulate import tabulate
 
@@ -219,6 +220,7 @@ def test_parentdag_downstream_clear(self):
             '--exclude-parentdag'])
         task_command.task_clear(args)
 
+    @pytest.mark.quarantined
     def test_local_run(self):
         args = self.parser.parse_args([
             'tasks',
@@ -253,6 +255,7 @@ def setUp(self):
 
         self.parser = cli_parser.get_parser()
 
+    @pytest.mark.quarantined
     def test_run_ignores_all_dependencies(self):
         """
         Test that run respects ignore_all_dependencies

tests/cli/commands/test_webserver_command.py

@@ -23,6 +23,7 @@
 from unittest import mock
 
 import psutil
+import pytest
 
 from airflow import settings
 from airflow.cli import cli_parser
@@ -86,6 +87,7 @@ def test_cli_webserver_debug(self):
         proc.wait()
 
 
+@pytest.mark.quarantined
 class TestCliWebServer(unittest.TestCase):
     @classmethod
     def setUpClass(cls):

tests/conftest.py

@@ -98,6 +98,11 @@ def pytest_addoption(parser):
         action="store_true",
         help="Includes long running tests (marked with long_running marker). They are skipped by default.",
     )
+    group.addoption(
+        "--include-quarantined",
+        action="store_true",
+        help="Includes quarantined tests (marked with quarantined marker). They are skipped by default.",
+    )
 
 
 def initial_db_init():
@@ -179,6 +184,9 @@ def pytest_configure(config):
     config.addinivalue_line(
         "markers", "long_running: mark test that run for a long time (many minutes)"
     )
+    config.addinivalue_line(
+        "markers", "quarantined: mark test that are in quarantine (i.e. flaky, need to be isolated and fixed)"
+    )
     config.addinivalue_line(
         "markers", "credential_file(name): mark tests that require credential file in CREDENTIALS_DIR"
     )
@@ -245,6 +253,13 @@ def skip_long_running_test(item):
                     format(item=item))
 
 
+def skip_quarantined_test(item):
+    for _ in item.iter_markers(name="quarantined"):
+        pytest.skip("The test is skipped because it has quarantined marker. "
+                    "And --include-quarantined flag is passed to pytest. {item}".
+                    format(item=item))
+
+
 def skip_if_integration_disabled(marker, item):
     integration_name = marker.args[0]
     environment_variable_name = "INTEGRATION_" + integration_name.upper()
@@ -304,6 +319,7 @@ def pytest_runtest_setup(item):
     selected_systems_list = item.config.getoption("--system")
 
     include_long_running = item.config.getoption("--include-long-running")
+    include_quarantined = item.config.getoption("--include-quarantined")
 
     for marker in item.iter_markers(name="integration"):
         skip_if_integration_disabled(marker, item)
@@ -325,5 +341,7 @@ def pytest_runtest_setup(item):
         skip_if_not_marked_with_runtime(selected_runtime, item)
     if not include_long_running:
         skip_long_running_test(item)
+    if not include_quarantined:
+        skip_quarantined_test(item)
     skip_if_credential_file_missing(item)
     skip_if_airflow_2_test(item)

tests/executors/test_dask_executor.py

@@ -20,6 +20,8 @@
 from datetime import timedelta
 from unittest import mock
 
+import pytest
+
 from airflow.configuration import conf
 from airflow.jobs.backfill_job import BackfillJob
 from airflow.models import DagBag
@@ -85,6 +87,7 @@ def test_dask_executor_functions(self):
         executor = DaskExecutor(cluster_address=self.cluster.scheduler_address)
         self.assert_tasks_on_executor(executor)
 
+    @pytest.mark.quarantined
     def test_backfill_integration(self):
         """
         Test that DaskExecutor can be used to backfill example dags

tests/jobs/test_backfill_job.py

@@ -52,6 +52,7 @@
 DEFAULT_DATE = timezone.datetime(2016, 1, 1)
 
 
+@pytest.mark.quarantined
 class TestBackfillJob(unittest.TestCase):
 
     def _get_dummy_dag(self, dag_id, pool=Pool.DEFAULT_POOL_NAME, task_concurrency=None):

tests/jobs/test_scheduler_job.py

@@ -1990,6 +1990,7 @@ def test_execute_task_instances_limit(self):
             ti.refresh_from_db()
             self.assertEqual(State.QUEUED, ti.state)
 
+    @pytest.mark.quarantined
     @pytest.mark.xfail(condition=True, reason="The test is flaky with nondeterministic result")
     def test_change_state_for_tis_without_dagrun(self):
         dag1 = DAG(dag_id='test_change_state_for_tis_without_dagrun', start_date=DEFAULT_DATE)
@@ -2589,6 +2590,7 @@ def do_schedule(mock_dagbag, mock_collect_dags):
         ti.refresh_from_db()
         self.assertEqual(State.SUCCESS, ti.state)
 
+    @pytest.mark.quarantined
     def test_retry_still_in_executor(self):
         """
         Checks if the scheduler does not put a task in limbo, when a task is retried
@@ -2676,6 +2678,7 @@ def run_with_error(ti, ignore_ti_state=False):
         ti.refresh_from_db()
         self.assertEqual(ti.state, State.SUCCESS)
 
+    @pytest.mark.quarantined
     @pytest.mark.xfail(condition=True, reason="This test is failing!")
     def test_retry_handling_job(self):
         """

tests/sensors/test_external_task_sensor.py

@@ -526,6 +526,7 @@ def dag_bag_multiple():
     yield dag_bag
 
 
+@pytest.mark.quarantined
 @pytest.mark.backend("postgres", "mysql")
 def test_clear_multiple_external_task_marker(dag_bag_multiple):
     """

tests/sensors/test_timeout_sensor.py

@@ -19,6 +19,8 @@
 import unittest
 from datetime import timedelta
 
+import pytest
+
 from airflow.exceptions import AirflowSensorTimeout, AirflowSkipException
 from airflow.models.dag import DAG
 from airflow.sensors.base_sensor_operator import BaseSensorOperator
@@ -72,6 +74,7 @@ def setUp(self):
         }
         self.dag = DAG(TEST_DAG_ID, default_args=args)
 
+    @pytest.mark.quarantined
     def test_timeout(self):
         op = TimeoutTestSensor(
             task_id='test_timeout',

tests/test_impersonation.py

@@ -26,6 +26,8 @@
 import unittest.mock
 from copy import deepcopy
 
+import pytest
+
 from airflow import models
 from airflow.jobs.backfill_job import BackfillJob
 from airflow.utils.db import add_default_pool_if_not_exists
@@ -107,6 +109,7 @@ def create_user():
             )
 
 
+@pytest.mark.quarantined
 class TestImpersonation(unittest.TestCase):
 
     def setUp(self):
@@ -183,6 +186,7 @@ def test_impersonation_subdag(self):
         )
 
 
+@pytest.mark.quarantined
 class TestImpersonationWithCustomPythonPath(unittest.TestCase):
 
     @mock_custom_module_path(TEST_UTILS_FOLDER)

tests/utils/test_process_utils.py

@@ -31,6 +31,7 @@
 from unittest import mock
 
 import psutil
+import pytest
 
 from airflow.exceptions import AirflowException
 from airflow.utils import process_utils
@@ -124,6 +125,7 @@ def my_sleep_subprocess_with_signals():
     sleep(100)
 
 
+@pytest.mark.quarantined
 class TestKillChildProcessesByPids(unittest.TestCase):
     def test_should_kill_process(self):
         before_num_process = subprocess.check_output(["ps", "-ax", "-o", "pid="]).decode().count("\n")

tests/www/test_views.py

@@ -2156,6 +2156,7 @@ def test_user_defined_filter_and_macros_raise_error(self, get_dag_function):
         )
 
 
+@pytest.mark.quarantined
 class TestTriggerDag(TestBase):
 
     def setUp(self):