[BEAM-3713] Add pytest for unit tests (apache#9756)

This is apache#7949 without IT support.

- Runs unit tests using pytest
  - tox: `tox -e py27-gcp-pytest,py36-pytest,etc.`
  - gradle: `../../gradlew pythonPreCommitPytest`
  - github PR phrase: `run python_pytest precommit`
- Tests run in parallel (still single-threaded but on multiple worker
processes).
  - no_xdist marker used for tests that don't work the xdist plugin.
- Allows specifying test module in tox cmd line. Example:
```sh
tox -e py27-pytest apache_beam.transforms.window_test
```

.gitignore

@@ -44,6 +44,7 @@ sdks/python/README.md
 sdks/python/apache_beam/portability/api/*pb2*.*
 sdks/python/apache_beam/portability/api/*.yaml
 sdks/python/nosetests*.xml
+sdks/python/pytest*.xml
 sdks/python/postcommit_requirements.txt
 
 # Ignore IntelliJ files.

.test-infra/jenkins/PrecommitJobBuilder.groovy

@@ -38,14 +38,19 @@ class PrecommitJobBuilder {
   /** If defined, set of path expressions used to trigger the job on commit. */
   List<String> triggerPathPatterns = []
 
+  /** Whether to trigger on new PR commits. Useful to set to false when testing new jobs. */
+  boolean commitTriggering = true
+
   /**
    * Define a set of pre-commit jobs.
    *
    * @param additionalCustomization Job DSL closure with additional customization to apply to the job.
    */
   void build(Closure additionalCustomization = {}) {
     defineCronJob additionalCustomization
-    defineCommitJob additionalCustomization
+    if (commitTriggering) {
+      defineCommitJob additionalCustomization
+    }
     definePhraseJob additionalCustomization
   }
 

.test-infra/jenkins/job_PreCommit_Python.groovy

@@ -35,3 +35,18 @@ builder.build {
     archiveJunit('**/nosetests*.xml')
   }
 }
+
+// Temporary job for testing pytest-based testing.
+// TODO(BEAM-3713): Remove this job once nose tests are replaced.
+PrecommitJobBuilder builderPytest = new PrecommitJobBuilder(
+    scope: this,
+    nameBase: 'Python_pytest',
+    gradleTask: ':pythonPreCommitPytest',
+    commitTriggering: false,
+)
+builderPytest.build {
+  // Publish all test results to Jenkins.
+  publishers {
+    archiveJunit('**/pytest*.xml')
+  }
+}

build.gradle

@@ -208,6 +208,14 @@ task pythonPreCommit() {
   // have caught. Note that the same tests will still run in postcommit.
 }
 
+// TODO(BEAM-3713): Temporary task for testing pytest.
+task pythonPreCommitPytest() {
+  dependsOn ":sdks:python:test-suites:tox:py2:preCommitPy2Pytest"
+  dependsOn ":sdks:python:test-suites:tox:py35:preCommitPy35Pytest"
+  dependsOn ":sdks:python:test-suites:tox:py36:preCommitPy36Pytest"
+  dependsOn ":sdks:python:test-suites:tox:py37:preCommitPy37Pytest"
+}
+
 task pythonLintPreCommit() {
   dependsOn ":sdks:python:test-suites:tox:py2:lint"
   dependsOn ":sdks:python:test-suites:tox:py37:lint"

sdks/python/apache_beam/coders/coders_test_common.py

@@ -24,6 +24,8 @@
 import unittest
 from builtins import range
 
+import pytest
+
 from apache_beam.coders import proto2_coder_test_messages_pb2 as test_message
 from apache_beam.coders import coders
 from apache_beam.internal import pickler
@@ -47,6 +49,9 @@ def decode(self, encoded):
     return int(encoded) - 1
 
 
+# These tests need to all be run in the same process due to the asserts
+# in tearDownClass.
+@pytest.mark.no_xdist
 class CodersTest(unittest.TestCase):
 
   # These class methods ensure that we test each defined coder in both

sdks/python/apache_beam/io/gcp/tests/bigquery_matcher_test.py

@@ -24,6 +24,7 @@
 import unittest
 
 import mock
+import pytest
 from hamcrest import assert_that as hc_assert_that
 
 from apache_beam.io.gcp import bigquery_tools
@@ -114,6 +115,7 @@ def test_bigquery_table_matcher_query_error_retry(self, mock_bigquery):
     self.assertEqual(bq_verifier.MAX_RETRIES + 1, mock_query.call_count)
 
 
+@pytest.mark.no_xdist  # xdist somehow makes the test do real requests.
 @unittest.skipIf(bigquery is None, 'Bigquery dependencies are not installed.')
 @mock.patch.object(
     bq_verifier.BigqueryFullResultStreamingMatcher,

sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py

@@ -29,6 +29,7 @@
 # patches unittest.TestCase to be python3 compatible
 import future.tests.base  # pylint: disable=unused-import
 import mock
+import pytest
 
 import apache_beam as beam
 import apache_beam.transforms as ptransform
@@ -254,6 +255,8 @@ def test_biqquery_read_streaming_fail(self):
                                 r'source is not currently available'):
       p.run()
 
+  # TODO(BEAM-8095): Segfaults in Python 3.7 with xdist.
+  @pytest.mark.no_xdist
   def test_remote_runner_display_data(self):
     remote_runner = DataflowRunner()
     p = Pipeline(remote_runner,

sdks/python/apache_beam/runners/portability/stager_test.py

@@ -26,6 +26,7 @@
 import unittest
 
 import mock
+import pytest
 
 from apache_beam.io.filesystems import FileSystems
 from apache_beam.options.pipeline_options import DebugOptions
@@ -160,6 +161,8 @@ def test_no_main_session(self):
                      self.stager.stage_job_resources(
                          options, staging_location=staging_dir)[1])
 
+  # xdist adds unpicklable modules to the main session.
+  @pytest.mark.no_xdist
   def test_with_main_session(self):
     staging_dir = self.make_temp_dir()
     options = PipelineOptions()

sdks/python/apache_beam/transforms/ptransform.py

@@ -700,8 +700,8 @@ def __init__(self, fn, *args, **kwargs):
     # Ensure fn and side inputs are picklable for remote execution.
     try:
       self.fn = pickler.loads(pickler.dumps(self.fn))
-    except RuntimeError:
-      raise RuntimeError('Unable to pickle fn %s' % self.fn)
+    except RuntimeError as e:
+      raise RuntimeError('Unable to pickle fn %s: %s' % (self.fn, e))
 
     self.args = pickler.loads(pickler.dumps(self.args))
     self.kwargs = pickler.loads(pickler.dumps(self.kwargs))

sdks/python/apache_beam/transforms/window_test.py

@@ -54,9 +54,6 @@ def context(element, timestamp):
   return WindowFn.AssignContext(timestamp, element)
 
 
-sort_values = Map(lambda k_vs: (k_vs[0], sorted(k_vs[1])))
-
-
 class ReifyWindowsFn(core.DoFn):
   def process(self, element, window=core.DoFn.WindowParam):
     key, values = element
@@ -186,6 +183,7 @@ def test_sliding_windows(self):
   def test_sessions(self):
     with TestPipeline() as p:
       pcoll = self.timestamped_key_values(p, 'key', 1, 2, 3, 20, 35, 27)
+      sort_values = Map(lambda k_vs: (k_vs[0], sorted(k_vs[1])))
       result = (pcoll
                 | 'w' >> WindowInto(Sessions(10))
                 | GroupByKey()

sdks/python/conftest.py

@@ -0,0 +1,32 @@
+#
+# 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.
+#
+"""Pytest configuration and custom hooks."""
+
+from __future__ import absolute_import
+
+import sys
+
+MAX_SUPPORTED_PYTHON_VERSION = (3, 8)
+
+# See pytest.ini for main collection rules.
+collect_ignore_glob = []
+if sys.version_info < (3,):
+  collect_ignore_glob.append('*_py3*.py')
+else:
+  for minor in range(sys.version_info.minor + 1,
+                     MAX_SUPPORTED_PYTHON_VERSION[1] + 1):
+    collect_ignore_glob.append('*_py3%d.py' % minor)

sdks/python/pytest.ini

@@ -0,0 +1,32 @@
+#
+# 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.
+#
+
+[pytest]
+junit_family = xunit2
+
+# Disable class-name-based test discovery.
+python_classes =
+# Disable function-name-based test discovery.
+python_functions =
+# Discover tests using filenames.
+# See conftest.py for extra collection rules.
+python_files = test_*.py *_test.py *_test_py3*.py
+
+markers =
+    # Tests using this marker conflict with the xdist plugin in some way, such
+    # as enabling save_main_session.
+    no_xdist: run without pytest-xdist plugin

sdks/python/scripts/run_pytest.sh

@@ -0,0 +1,49 @@
+#!/bin/bash
+#
+#    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.
+#
+# Utility script for tox.ini for running unit tests.
+#
+# Runs tests in parallel, except those not compatible with xdist. Combines
+# exit statuses of runs, special-casing 5, which says that no tests were
+# selected.
+#
+# $1 - suite base name
+# $2 - additional arguments to pass to pytest
+
+envname=${1?First argument required: suite base name}
+posargs=$2
+
+# Run with pytest-xdist and without.
+python setup.py pytest --addopts="-o junit_suite_name=${envname} \
+  --junitxml=pytest_${envname}.xml -m 'not no_xdist' -n 6 --pyargs ${posargs}"
+status1=$?
+python setup.py pytest --addopts="-o junit_suite_name=${envname}_no_xdist \
+  --junitxml=pytest_${envname}_no_xdist.xml -m 'no_xdist' --pyargs ${posargs}"
+status2=$?
+
+# Exit with error if no tests were run (status code 5).
+if [[ $status1 == 5 && $status2 == 5 ]]; then
+  exit $status1
+fi
+
+# Exit with error if one of the statuses has an error that's not 5.
+if [[ $status1 && $status1 != 5 ]]; then
+  exit $status1
+fi
+if [[ $status2 && $status2 != 5 ]]; then
+  exit $status2
+fi

sdks/python/setup.py

@@ -149,6 +149,8 @@ def get_version():
     'pyyaml>=3.12,<6.0.0',
     'requests_mock>=1.7,<2.0',
     'tenacity>=5.0.2,<6.0',
+    'pytest>=4.4.0,<5.0',
+    'pytest-xdist>=1.29.0,<2',
     ]
 
 GCP_REQUIREMENTS = [
@@ -226,6 +228,7 @@ def run(self):
     install_requires=REQUIRED_PACKAGES,
     python_requires=python_requires,
     test_suite='nose.collector',
+    setup_requires=['pytest_runner'],
     tests_require=REQUIRED_TEST_PACKAGES,
     extras_require={
         'docs': ['Sphinx>=1.5.2,<2.0'],

sdks/python/test-suites/tox/py2/build.gradle

@@ -45,6 +45,14 @@ test.dependsOn testPy2Cython
 // project.
 testPy2Cython.mustRunAfter testPython2, testPy2Gcp
 
+// TODO(BEAM-3713): Temporary pytest tasks that should eventually replace
+//  nose-based test tasks.
+toxTask "testPy2GcpPytest", "py27-gcp-pytest"
+toxTask "testPython2Pytest", "py27-pytest"
+toxTask "testPy2CythonPytest", "py27-cython-pytest"
+// Ensure that cython tests run exclusively to other tests.
+testPy2CythonPytest.mustRunAfter testPython2Pytest, testPy2GcpPytest
+
 toxTask "docs", "docs"
 assemble.dependsOn docs
 
@@ -56,3 +64,11 @@ task preCommitPy2() {
   dependsOn "testPython2"
   dependsOn "testPy2Gcp"
 }
+
+task preCommitPy2Pytest {
+  dependsOn "docs"
+  dependsOn "testPy2CythonPytest"
+  dependsOn "testPython2Pytest"
+  dependsOn "testPy2GcpPytest"
+  dependsOn "lint"
+}

sdks/python/test-suites/tox/py35/build.gradle

@@ -39,8 +39,23 @@ test.dependsOn testPy35Cython
 // project.
 testPy35Cython.mustRunAfter testPython35, testPy35Gcp
 
+// TODO(BEAM-3713): Temporary pytest tasks that should eventually replace
+//  nose-based test tasks.
+toxTask "testPy35GcpPytest", "py35-gcp-pytest"
+toxTask "testPython35Pytest", "py35-pytest"
+toxTask "testPy35CythonPytest", "py35-cython-pytest"
+// Ensure that cython tests run exclusively to other tests.
+testPy35CythonPytest.mustRunAfter testPython35Pytest, testPy35GcpPytest
+
 task preCommitPy35() {
     dependsOn "testPython35"
     dependsOn "testPy35Gcp"
     dependsOn "testPy35Cython"
 }
+
+task preCommitPy35Pytest {
+  dependsOn "testPython35Pytest"
+  dependsOn "testPy35GcpPytest"
+  dependsOn "testPy35CythonPytest"
+  dependsOn "lint"
+}

sdks/python/test-suites/tox/py36/build.gradle

@@ -40,8 +40,22 @@ test.dependsOn testPy36Cython
 // project.
 testPy36Cython.mustRunAfter testPython36, testPy36Gcp
 
+// TODO(BEAM-3713): Temporary pytest tasks that should eventually replace
+//  nose-based test tasks.
+toxTask "testPy36GcpPytest", "py36-gcp-pytest"
+toxTask "testPython36Pytest", "py36-pytest"
+toxTask "testPy36CythonPytest", "py36-cython-pytest"
+// Ensure that cython tests run exclusively to other tests.
+testPy36CythonPytest.mustRunAfter testPython36Pytest, testPy36GcpPytest
+
 task preCommitPy36() {
     dependsOn "testPython36"
     dependsOn "testPy36Gcp"
     dependsOn "testPy36Cython"
 }
+
+task preCommitPy36Pytest {
+  dependsOn "testPython36Pytest"
+  dependsOn "testPy36GcpPytest"
+  dependsOn "testPy36CythonPytest"
+}

sdks/python/test-suites/tox/py37/build.gradle

@@ -49,9 +49,23 @@ test.dependsOn testPy37Interactive
 // project.
 testPy37Cython.mustRunAfter testPython37, testPy37Gcp
 
+// TODO(BEAM-3713): Temporary pytest tasks that should eventually replace
+//  nose-based test tasks.
+toxTask "testPy37GcpPytest", "py37-gcp-pytest"
+toxTask "testPython37Pytest", "py37-pytest"
+toxTask "testPy37CythonPytest", "py37-cython-pytest"
+// Ensure that cython tests run exclusively to other tests.
+testPy37CythonPytest.mustRunAfter testPython37Pytest, testPy37GcpPytest
+
 task preCommitPy37() {
     dependsOn "testPython37"
     dependsOn "testPy37Gcp"
     dependsOn "testPy37Cython"
     dependsOn "testPy37Interactive"
 }
+
+task preCommitPy37Pytest {
+  dependsOn "testPython37Pytest"
+  dependsOn "testPy37GcpPytest"
+  dependsOn "testPy37CythonPytest"
+}

sdks/python/test_config.py

@@ -20,6 +20,7 @@
 This module contains nose plugin hooks that configures Beam tests which
 includes ValidatesRunner test and E2E integration test.
 
+TODO(BEAM-3713): Remove this module once nose is removed.
 """
 
 from __future__ import absolute_import