Disable remaining integration tests for public unit test builds (dags…

…ter-io#15762)

I've played whack-a-mole to skip the remaining parts of our "unit tests"
that rely on third-party integrations.

.buildkite/dagster-buildkite/dagster_buildkite/pipelines/dagster_oss_main.py

@@ -22,7 +22,7 @@ def build_dagster_oss_main_steps() -> List[BuildkiteStep]:
     # overridden by setting the `INTERNAL_BRANCH` environment variable or passing
     # `[INTERNAL_BRANCH=<branch>]` in the commit message. Master/release branches
     # always run on the matching internal branch.
-    if not oss_contribution:
+    if not oss_contribution and not os.getenv("CI_DISABLE_INTEGRATION_TESTS"):
         if branch_name == "master" or is_release_branch(branch_name):
             pipeline_name = "internal"
             trigger_branch = branch_name  # build on matching internal release branch

.buildkite/dagster-buildkite/dagster_buildkite/steps/dagster.py

@@ -46,12 +46,14 @@ def build_dagster_steps() -> List[BuildkiteStep]:
     steps += build_helm_steps()
     steps += build_sql_schema_check_steps()
     steps += build_graphql_python_client_backcompat_steps()
-    steps += build_integration_steps()
+    if not os.getenv("CI_DISABLE_INTEGRATION_TESTS"):
+        steps += build_integration_steps()
 
     # Build images containing the dagster-test sample project. This is a dependency of certain
     # dagster core and extension lib tests. Run this after we build our library package steps
     # because need to know whether it's a dependency of any of them.
-    steps += build_test_project_steps()
+    if not os.getenv("CI_DISABLE_INTEGRATION_TESTS"):
+        steps += build_test_project_steps()
 
     return steps
 

.buildkite/dagster-buildkite/dagster_buildkite/steps/integration.py

@@ -237,15 +237,23 @@ def build_integration_suite_steps(
 
 
 def default_integration_suite_pytest_extra_cmds(version: str, _) -> List[str]:
-    return [
+    cmds = [
         'export AIRFLOW_HOME="/airflow"',
         "mkdir -p $${AIRFLOW_HOME}",
         "export DAGSTER_DOCKER_IMAGE_TAG=$${BUILDKITE_BUILD_ID}-" + version,
         'export DAGSTER_DOCKER_REPOSITORY="$${AWS_ACCOUNT_ID}.dkr.ecr.us-west-2.amazonaws.com"',
         "aws ecr get-login --no-include-email --region us-west-2 | sh",
-        r"aws s3 cp s3://\${BUILDKITE_SECRETS_BUCKET}/gcp-key-elementl-dev.json "
-        + GCP_CREDS_LOCAL_FILE,
-        "export GOOGLE_APPLICATION_CREDENTIALS=" + GCP_CREDS_LOCAL_FILE,
+    ]
+
+    # If integration tests are disabled, we won't have any gcp credentials to download.
+    if not os.getenv("CI_DISABLE_INTEGRATION_TESTS"):
+        cmds += [
+            r"aws s3 cp s3://\${BUILDKITE_SECRETS_BUCKET}/gcp-key-elementl-dev.json "
+            + GCP_CREDS_LOCAL_FILE,
+            "export GOOGLE_APPLICATION_CREDENTIALS=" + GCP_CREDS_LOCAL_FILE,
+        ]
+
+    cmds += [
         "pushd python_modules/libraries/dagster-celery",
         # Run the rabbitmq db. We are in docker running docker
         # so this will be a sibling container.
@@ -258,3 +266,5 @@ def default_integration_suite_pytest_extra_cmds(version: str, _) -> List[str]:
         ),
         "popd",
     ]
+
+    return cmds

.buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py

@@ -228,11 +228,15 @@ def k8s_extra_cmds(version: str, _) -> List[str]:
     ]
 
 
-gcp_extra_cmds = [
-    r"aws s3 cp s3://\${BUILDKITE_SECRETS_BUCKET}/gcp-key-elementl-dev.json "
-    + GCP_CREDS_LOCAL_FILE,
-    "export GOOGLE_APPLICATION_CREDENTIALS=" + GCP_CREDS_LOCAL_FILE,
-]
+gcp_extra_cmds = (
+    [
+        r"aws s3 cp s3://\${BUILDKITE_SECRETS_BUCKET}/gcp-key-elementl-dev.json "
+        + GCP_CREDS_LOCAL_FILE,
+        "export GOOGLE_APPLICATION_CREDENTIALS=" + GCP_CREDS_LOCAL_FILE,
+    ]
+    if not os.getenv("CI_DISABLE_INTEGRATION_TESTS")
+    else []
+)
 
 
 postgres_extra_cmds = [

.buildkite/dagster-buildkite/dagster_buildkite/steps/test_project.py

@@ -1,3 +1,4 @@
+import os
 from typing import List, Optional, Set
 
 from ..images.versions import (
@@ -94,8 +95,11 @@ def _test_project_step_key(version: AvailablePythonVersion) -> str:
 
 
 def test_project_depends_fn(version: AvailablePythonVersion, _) -> List[str]:
-    build_test_project_for.add(version)
-    return [_test_project_step_key(version)]
+    if not os.getenv("CI_DISABLE_INTEGRATION_TESTS"):
+        build_test_project_for.add(version)
+        return [_test_project_step_key(version)]
+    else:
+        return []
 
 
 def skip_if_version_not_needed(version: AvailablePythonVersion) -> Optional[str]:

python_modules/libraries/dagster-celery-docker/dagster_celery_docker_tests/test_execute_docker.py

@@ -4,6 +4,7 @@
 import os
 from contextlib import contextmanager
 
+import pytest
 from dagster._core.execution.api import execute_job
 from dagster._utils.merger import merge_dicts
 from dagster._utils.test.postgres_instance import postgres_instance_for_test
@@ -27,6 +28,7 @@ def celery_docker_postgres_instance(overrides=None):
         yield instance
 
 
+@pytest.mark.integration
 def test_execute_celery_docker_image_on_executor_config(aws_creds):
     docker_image = get_test_project_docker_image()
     docker_config = {
@@ -75,6 +77,7 @@ def test_execute_celery_docker_image_on_executor_config(aws_creds):
             assert result.output_for_node("get_environment") == "here!"
 
 
+@pytest.mark.integration
 def test_execute_celery_docker_image_on_job_config(aws_creds):
     docker_image = get_test_project_docker_image()
     docker_config = {

python_modules/libraries/dagster-gcp-pyspark/dagster_gcp_pyspark_tests/bigquery/test_type_handler.py

@@ -79,6 +79,7 @@ def temporary_bigquery_table(schema_name: str) -> Iterator[str]:
         ).result()
 
 
+@pytest.mark.integration
 def test_handle_output(spark):
     with patch("pyspark.sql.DataFrame.write") as mock_write:
         handler = BigQueryPySparkTypeHandler()
@@ -111,6 +112,7 @@ def test_handle_output(spark):
         assert len(mock_write.method_calls) == 1
 
 
+@pytest.mark.integration
 def test_load_input(spark):
     with patch("pyspark.sql.DataFrameReader.load") as mock_read:
         columns = ["col1", "col2"]

python_modules/libraries/dagster-gcp/dagster_gcp_tests/bigquery_tests/test_ops.py

@@ -44,6 +44,7 @@ def get_dataset():
     return "test_ds_" + str(uuid.uuid4()).replace("-", "_")
 
 
+@pytest.mark.integration
 def test_simple_queries():
     @job(resource_defs={"bigquery": bigquery_resource})
     def bq_test():
@@ -151,6 +152,7 @@ def test_config():
         assert error_message in result.errors[0].message
 
 
+@pytest.mark.integration
 def test_create_delete_dataset():
     client = bigquery.Client()
     dataset = get_dataset()

python_modules/libraries/dagster-gcp/dagster_gcp_tests/bigquery_tests/test_resource.py

@@ -9,6 +9,7 @@
 
 
 @pytest.mark.skipif(not IS_BUILDKITE, reason="Requires access to the BUILDKITE bigquery DB")
+@pytest.mark.integration
 def test_old_resource_authenticate_via_config():
     asset_info = dict()
 
@@ -52,6 +53,7 @@ def test_asset() -> int:
 
 
 @pytest.mark.skipif(not IS_BUILDKITE, reason="Requires access to the BUILDKITE bigquery DB")
+@pytest.mark.integration
 def test_pythonic_resource_authenticate_via_config():
     asset_info = dict()
 
@@ -94,6 +96,7 @@ def test_asset(bigquery: BigQueryResource) -> int:
 
 
 @pytest.mark.skipif(not IS_BUILDKITE, reason="Requires access to the BUILDKITE bigquery DB")
+@pytest.mark.integration
 def test_pythonic_resource_authenticate_via_env():
     @asset
     def test_asset(bigquery: BigQueryResource) -> int:

python_modules/libraries/dagster-gcp/dagster_gcp_tests/dataproc_tests/test_resources.py

@@ -4,6 +4,7 @@
 from unittest import mock
 
 import httplib2
+import pytest
 from dagster import RunConfig, _seven, job
 from dagster_gcp import (
     DataprocOpConfig,
@@ -75,6 +76,7 @@ def request(
             return response, content
 
 
+@pytest.mark.integration
 def test_dataproc_resource():
     """Tests dataproc cluster creation/deletion. Requests are captured by the responses library, so
     no actual HTTP requests are made here.
@@ -132,6 +134,7 @@ def test_dataproc():
         assert result.success
 
 
+@pytest.mark.integration
 def test_wait_for_job_with_timeout():
     """Test submitting a job with timeout of 0 second so that it always fails."""
     with mock.patch("httplib2.Http", new=HttpSnooper):
@@ -186,6 +189,7 @@ def test_dataproc():
             assert "Job run timed out" in str(e)
 
 
+@pytest.mark.integration
 def test_pydantic_dataproc_resource():
     """Tests pydantic dataproc cluster creation/deletion. Requests are captured by the responses library, so
     no actual HTTP requests are made here.
@@ -237,6 +241,7 @@ def test_dataproc():
         assert result.success
 
 
+@pytest.mark.integration
 def test_wait_for_job_with_timeout_pydantic():
     """Test submitting a job with timeout of 0 second so that it always fails."""
     with mock.patch("httplib2.Http", new=HttpSnooper):

python_modules/libraries/dagster-gcp/dagster_gcp_tests/gcs_tests/test_compute_log_manager.py

@@ -28,6 +28,7 @@
 ]
 
 
+@pytest.mark.integration
 def test_compute_log_manager(gcs_bucket):
     @job
     def simple():
@@ -118,6 +119,7 @@ def easy(context):
                 assert expected in stderr.data
 
 
+@pytest.mark.integration
 def test_compute_log_manager_with_envvar(gcs_bucket):
     @job
     def simple():
@@ -206,6 +208,7 @@ def easy(context):
                     assert expected in stderr.data
 
 
+@pytest.mark.integration
 def test_compute_log_manager_from_config(gcs_bucket):
     gcs_prefix = "foobar"
 
@@ -228,6 +231,7 @@ def test_compute_log_manager_from_config(gcs_bucket):
     assert isinstance(instance.compute_log_manager, GCSComputeLogManager)
 
 
+@pytest.mark.integration
 def test_prefix_filter(gcs_bucket):
     gcs_prefix = "foo/bar/"  # note the trailing slash
 
@@ -248,6 +252,7 @@ def test_prefix_filter(gcs_bucket):
         assert logs == "hello hello"
 
 
+@pytest.mark.integration
 def test_storage_download_url_fallback(gcs_bucket):
     with tempfile.TemporaryDirectory() as temp_dir:
         manager = GCSComputeLogManager(bucket=gcs_bucket, local_dir=temp_dir)
@@ -275,6 +280,7 @@ def _return_mocked_blob(*args, **kwargs):
             assert url.startswith("/logs")  # falls back to local storage url
 
 
+@pytest.mark.integration
 class TestGCSComputeLogManager(TestCapturedLogManager):
     __test__ = True
 

python_modules/libraries/dagster-gcp/dagster_gcp_tests/gcs_tests/test_gcs_file_manager.py

@@ -1,11 +1,13 @@
 from unittest import mock
 
+import pytest
 from dagster import configured, job, op
 from dagster_gcp.gcs.file_manager import GCSFileHandle, GCSFileManager
 from dagster_gcp.gcs.resources import gcs_file_manager
 from google.cloud import storage
 
 
+@pytest.mark.integration
 def test_gcs_file_manager_write():
     gcs_mock = mock.MagicMock()
     file_manager = GCSFileManager(storage.client.Client(), "some-bucket", "some-key")

python_modules/libraries/dagster-gcp/dagster_gcp_tests/gcs_tests/test_io_manager.py

@@ -1,5 +1,6 @@
 import pickle
 
+import pytest
 from dagster import (
     AssetsDefinition,
     DagsterInstance,
@@ -81,6 +82,7 @@ def basic_external_plan_execution():
     return basic_external_plan_execution
 
 
+@pytest.mark.integration
 def test_gcs_pickle_io_manager_execution(gcs_bucket):
     inty_job = define_inty_job()
 

python_modules/libraries/dagster-gcp/dagster_gcp_tests/gcs_tests/test_resources.py

@@ -1,9 +1,11 @@
+import pytest
 from dagster import build_op_context, op
 from dagster_gcp.gcs.resources import GCSResource, gcs_resource
 
 PROJECT_ID = "test-project1231"
 
 
+@pytest.mark.integration
 def test_gcs_resource():
     @op(required_resource_keys={"gcs"})
     def gcs_op(context):
@@ -16,6 +18,7 @@ def gcs_op(context):
     assert gcs_op(context)
 
 
+@pytest.mark.integration
 def test_pydantic_gcs_resource():
     @op
     def gcs_op(gcs: GCSResource):