test: share integration session, use spark.master=local, and align CI extras

[dborowitz]

Summary

Refactors the integration test suite and CI workflows to substantially reduce Dataproc Serverless session churn, VM footprint, and vulnerability to regional GCE stockouts (503 Failed to create main node pool), while expanding CI coverage across packaging extras ([client], [spark], and with/without [interactive]).

Metric	Before	After	Change
Collected integration test items	44 (42 active, 2 skipped)	11 (10 active, 1 skipped)	-75%
Remote Dataproc Serverless clusters created	42	5 (1 multi-node + 4 single-node)	-88%
Total GCE VMs provisioned per run	~130 VMs	7 VMs (3 + 4×1)	-94.6%
Leaked clusters per run	4 (ml-pipeline-session-* × 2 + terminated-recreate × 2)	0 (all wrapped in finally + 10m idle TTL)	-100%

Key Changes

P0: Shared Session Fixture, Dead Parameterization Removal, and Leak Prevention

• Shared multi-node session fixture (tests/integration/conftest.py): Provides a session-scoped default multi-node ManagedSparkSession (_shared_connect_session / connect_session) shared across all non-destructive tests (test_create_spark_session_with_default_notebook_behavior, test_reuse_s8s_spark_session, test_sql_functions, test_sql_udf, test_sparksql_magic_with_managed_spark_session, test_dpip_install_success, test_dpip_install_failure), and an isolated_session_state fixture for tests that manage their own session lifecycle.
• Remove dead image_version=[None, "3.0"] fixture: Previously parameterized every test in test_session.py and test_magics.py without ever passing image_version to the session builder, doubling the entire integration suite.
• Safety TTLs & leak cleanup: Sets MANAGED_SPARK_CONNECT_TTL_SECONDS=1800 (30m) and MANAGED_SPARK_CONNECT_IDLE_TTL_SECONDS=600 (10m, the minimum allowed by ServerlessValidator) by default in conftest.py, and folds test_stop_named_session_with_terminate_true + finally: cleanup into test_session_reuse_with_custom_id so named sessions are never leaked for 24 hours.

P1: Single-Node Mode (spark.master=local) & Test Consolidation

• Use spark.master=local on isolated control-plane/lifecycle tests: Provisions a single 4-vCPU driver VM and 0 executor node pools for test_create_spark_session_with_session_template_and_user_provided_session_config, test_session_reuse_with_custom_id, and test_serverless_session_termination_and_deletion_lifecycle, while keeping the shared connect_session multi-node to test out-of-the-box defaults, remote UDFs, and %dpip artifact distribution.
• Right-size executor config override test: Changes .config("spark.executor.cores", "7") ... .config("spark.executor.cores", "16") to "8" $\rightarrow$ "4" with spark.master=local, and verifies SessionTemplate default runtime version ("3.0") via get_session_template rather than spinning up a separate cluster.
• Consolidate termination & deletion lifecycle tests: Combines test_stop_managed_session_with_terminate_false, test_get_or_create_spark_session_with_terminated_serverless_session, test_stop_spark_session_with_terminated_serverless_session, and test_stop_spark_session_with_deleted_serverless_session into test_serverless_session_termination_and_deletion_lifecycle (2 single-node sessions instead of 4 multi-node sessions).
• Drop local-only unit test duplicates from tests/integration/: Removes pure local CLI/regex validation tests that are already covered in tests/unit/.

P2: Realistic CI Packaging Lanes

• Install .[client,interactive] in integration-tests.yaml: Tests against the lightweight pyspark-client + pyspark-connect runtime recommended in the README, ensures %%sparksql is not skipped in integration CI, and fixes the pip cache key to hash setup.py and requirements-test.txt.
• Verify both without and with [interactive] in tests.yaml: Installs .[${{ matrix.spark-extra }}] first to run TestWithoutInteractiveExtra (expanded to test ManagedSparkSession.builder.getOrCreate(), get_tqdm_bar(), _repr_html_(), and stop() when IPython/sqlparse/traitlets/colabsqlviz/ipywidgets are absent), then installs .[${{ matrix.spark-extra }},interactive] for the full unit and local suites.
• Run test_create_local_spark_session in tests/local/test_sparksql_local.py: Moves the DATAPROC_WORKLOAD_TYPE=batch local classic Spark test into tests/local/ so it runs on every PR in the spark lane under Java 17/21, and isolates local Spark Connect proxy/environment state.

[gemini-code-assist]

Code Review

This pull request refactors the integration tests by consolidating common fixtures into tests/integration/conftest.py, reducing duplication across test files, and adding robust lifecycle and SQL-related tests. It also improves local Spark session tests and unit tests for environments without interactive dependencies. The review feedback highlights two important improvements: replacing os.environ.clear() with a safer environment restoration approach to prevent unsetting critical system variables, and using getattr when accessing SparkSession._instantiatedSession to avoid potential AttributeError across different PySpark versions.

[gemini-code-assist]

Clearing os.environ completely with os.environ.clear() can be dangerous in concurrent test environments or when other threads/libraries are running. It temporarily unsets critical system environment variables (like PATH, HOME, TMPDIR, etc.), which can cause unexpected failures in concurrent processes or subprocesses. A safer approach is to only delete the keys that were added during the test and restore the original values for modified keys.

Suggested change

	os.environ.clear()
	os.environ.update(original_environment)
	for key in list(os.environ.keys()):
	if key not in original_environment:
	del os.environ[key]
	os.environ.update(original_environment)

[gemini-code-assist]

Directly accessing SparkSession._instantiatedSession can raise an AttributeError if the attribute is not present on the SparkSession class (for example, in different PySpark versions or if the class is mocked/patched in certain test environments). Using getattr is a safer and more robust way to retrieve and stop the instantiated session.

Suggested change

	if SparkSession._instantiatedSession is not None:
	SparkSession._instantiatedSession.stop()
	instantiated = getattr(SparkSession, "_instantiatedSession", None)
	if instantiated is not None:
	instantiated.stop()

[dborowitz]

TIL CI only runs on PRs from main repo branches, not forks (and this even applied to the other PRs I submitted this week, yikes, but good thing it ran on the 2.0.0 PR). I will reopen.