diff --git a/google/cloud/managed_spark_connect/session.py b/google/cloud/managed_spark_connect/session.py index 5f07ead1..61f5256d 100644 --- a/google/cloud/managed_spark_connect/session.py +++ b/google/cloud/managed_spark_connect/session.py @@ -52,6 +52,7 @@ Aborted, FailedPrecondition, InvalidArgument, + MethodNotImplemented, NotFound, PermissionDenied, ) @@ -427,6 +428,118 @@ def _is_valid_session_id(session_id: str) -> bool: return bool(re.match(pattern, session_id)) +_SESSION_TEMPLATE_RESOURCE_PATTERN = re.compile( + r"^projects/[^/]+/(?:locations|regions)/[^/]+/sessionTemplates/[a-zA-Z0-9_-]+$" +) +_SESSION_TEMPLATE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$") + + +def _is_valid_session_template_ref(template: str) -> bool: + if "/" in template: + return bool(_SESSION_TEMPLATE_RESOURCE_PATTERN.match(template)) + return bool(_SESSION_TEMPLATE_ID_PATTERN.match(template)) + + +def _extract_api_error_details( + exc: BaseException, +) -> Tuple[List[str], List[str]]: + """Return (reasons, debug_details) from a google.api_core GoogleAPICallError.""" + reasons: List[str] = [] + debug_details: List[str] = [] + for item in getattr(exc, "details", None) or (): + reason = getattr(item, "reason", None) + if isinstance(reason, str) and reason and reason not in reasons: + reasons.append(reason) + detail = getattr(item, "detail", None) + if ( + isinstance(detail, str) + and detail + and not detail.startswith("[ORIGINAL ERROR]") + and detail not in debug_details + ): + debug_details.append(detail) + for violation in getattr(item, "field_violations", None) or (): + field = getattr(violation, "field", "") + desc = getattr(violation, "description", "") + if desc: + entry = f"{field}: {desc}" if field else desc + if entry not in debug_details: + debug_details.append(entry) + return reasons, debug_details + + +def _format_create_session_error( + exc: BaseException, + project_id: Optional[str], + region: Optional[str], + api_endpoint: Optional[str], + session_template: Optional[str] = None, +) -> str: + prefix = "Error while creating Managed Spark Session" + raw_msg = (getattr(exc, "message", None) or str(exc)).strip() + if isinstance(exc, MethodNotImplemented) and "status: 404" in raw_msg: + default_endpoint = ( + f"{region}-dataproc.googleapis.com" if region else None + ) + if region and ( + api_endpoint is None or api_endpoint == default_endpoint + ): + return ( + f"{prefix}: Regional endpoint '{default_endpoint}' " + f"(derived from location '{region}') returned HTTP 404; " + f"check that '{region}' is a valid Managed Spark region." + ) + endpoint_desc = f" '{api_endpoint}'" if api_endpoint else "" + region_desc = f" (location '{region}')" if region else "" + return ( + f"{prefix}: Endpoint{endpoint_desc}{region_desc} " + f"returned HTTP 404: {raw_msg}" + ) + + base_msg = raw_msg + reasons, debug_details = _extract_api_error_details(exc) + extra_parts: List[str] = [] + + for detail in debug_details: + if detail not in base_msg: + extra_parts.append(detail) + + if not debug_details: + if "CONSUMER_INVALID" in reasons and project_id: + note = f"Project '{project_id}' was not found or is invalid." + if note not in base_msg: + extra_parts.append(note) + elif "RESOURCE_PROJECT_INVALID" in reasons and project_id: + note = f"Invalid project ID '{project_id}'." + if note not in base_msg: + extra_parts.append(note) + elif "LOCATION_POLICY_VIOLATED" in reasons and region: + note = f"Invalid location '{region}'." + if note not in base_msg: + extra_parts.append(note) + + if ( + isinstance(exc, InvalidArgument) + and session_template + and not _is_valid_session_template_ref(session_template) + and "session_template" not in base_msg + and "Session Template" not in base_msg + and "session template" not in base_msg + ): + extra_parts.append( + f"Invalid session_template '{session_template}'; expected a " + "template ID or 'projects/{project}/locations/{location}/" + "sessionTemplates/{template}'." + ) + + if base_msg and extra_parts and base_msg[-1] not in ".!?:;": + base_msg = f"{base_msg}." + parts = [p for p in (base_msg, *extra_parts) if p] + if not parts: + return prefix + return f"{prefix}: {' '.join(parts)}" + + def _timestamp_str(value: Any) -> Optional[str]: if value is None: return None @@ -941,12 +1054,24 @@ def create_session_pbar(): logger.error( f"Exception while writing active session to file {file_path}, {e}" ) - except (InvalidArgument, PermissionDenied) as e: + except ( + InvalidArgument, + PermissionDenied, + MethodNotImplemented, + ) as e: ManagedSparkSession._active_s8s_session_id = None ManagedSparkSession._active_session_uses_custom_id = False raise ManagedSparkConnectException( - f"Error while creating Managed Spark Session: {e.message}" - ) + _format_create_session_error( + e, + self._project_id, + self._region, + _get_client_option( + self._client_options, "api_endpoint" + ), + session_config.session_template, + ) + ) from e except DefaultCredentialsError as e: ManagedSparkSession._active_s8s_session_id = None ManagedSparkSession._active_session_uses_custom_id = False diff --git a/tests/integration/test_session.py b/tests/integration/test_session.py index 055b8a7f..67ea6b4f 100644 --- a/tests/integration/test_session.py +++ b/tests/integration/test_session.py @@ -14,6 +14,11 @@ import datetime import uuid +from google.api_core.exceptions import ( + InvalidArgument, + MethodNotImplemented, + PermissionDenied, +) from google.cloud.dataproc_v1 import ( CreateSessionTemplateRequest, DeleteSessionRequest, @@ -25,6 +30,9 @@ TerminateSessionRequest, ) from google.cloud.managed_spark_connect import ManagedSparkSession +from google.cloud.managed_spark_connect.exceptions import ( + ManagedSparkConnectException, +) from pyspark.errors.exceptions import connect as connect_exceptions from pyspark.sql.types import StringType import pytest @@ -516,3 +524,114 @@ def test_serverless_session_termination_and_deletion_lifecycle( ) except Exception: pass + + +def test_create_spark_session_with_invalid_project_id( + test_project, test_region, os_environment, isolated_session_state +): + """Test that a nonexistent project ID raises a clear ManagedSparkConnectException.""" + invalid_project = f"{test_project}-nonexistent-{uuid.uuid4().hex[:8]}" + + with pytest.raises(ManagedSparkConnectException) as exc_info: + ( + ManagedSparkSession.builder.projectId(invalid_project) + .location(test_region) + .getOrCreate() + ) + + assert isinstance(exc_info.value.__cause__, PermissionDenied) + msg = str(exc_info.value) + assert msg.startswith("Error while creating Managed Spark Session:") + assert f"Permission denied on resource project {invalid_project}." in msg + assert "not found" in msg + assert ManagedSparkSession._active_s8s_session_id is None + + +def test_create_spark_session_with_invalid_region( + test_project, os_environment, isolated_session_state +): + """Test that an invalid region raises a clear ManagedSparkConnectException explaining the 404.""" + invalid_region = "us-east-4" + + with pytest.raises(ManagedSparkConnectException) as exc_info: + ( + ManagedSparkSession.builder.projectId(test_project) + .location(invalid_region) + .getOrCreate() + ) + + assert isinstance(exc_info.value.__cause__, MethodNotImplemented) + msg = str(exc_info.value) + assert msg == ( + "Error while creating Managed Spark Session: Regional endpoint" + " 'us-east-4-dataproc.googleapis.com' (derived from location" + " 'us-east-4') returned HTTP 404; check that 'us-east-4' is a valid" + " Managed Spark region." + ) + assert ManagedSparkSession._active_s8s_session_id is None + + +def test_create_spark_session_with_invalid_session_template( + test_project, test_region, os_environment, isolated_session_state +): + """Test that a nonexistent session template raises a clear ManagedSparkConnectException.""" + template_id = f"nonexistent-template-{uuid.uuid4().hex[:8]}" + nonexistent_template = ( + f"projects/{test_project}/locations/{test_region}/sessionTemplates/" + f"{template_id}" + ) + session_config = Session() + session_config.session_template = nonexistent_template + + with pytest.raises(ManagedSparkConnectException) as exc_info: + ( + ManagedSparkSession.builder.projectId(test_project) + .location(test_region) + .sessionConfig(session_config) + .getOrCreate() + ) + + assert isinstance(exc_info.value.__cause__, InvalidArgument) + msg = str(exc_info.value) + assert msg.startswith("Error while creating Managed Spark Session:") + assert f"'{template_id}' does not exist" in msg + assert ManagedSparkSession._active_s8s_session_id is None + + bare_template_config = Session() + bare_template_config.session_template = template_id + + with pytest.raises(ManagedSparkConnectException) as bare_exc_info: + ( + ManagedSparkSession.builder.projectId(test_project) + .location(test_region) + .sessionConfig(bare_template_config) + .getOrCreate() + ) + + assert isinstance(bare_exc_info.value.__cause__, InvalidArgument) + bare_msg = str(bare_exc_info.value) + assert bare_msg.startswith("Error while creating Managed Spark Session:") + assert f"'{template_id}' does not exist" in bare_msg + assert ManagedSparkSession._active_s8s_session_id is None + + mismatched_region_config = Session() + mismatched_region_config.session_template = ( + f"projects/{test_project}/locations/us-east-4/sessionTemplates/" + f"{template_id}" + ) + + with pytest.raises(ManagedSparkConnectException) as mismatch_exc_info: + ( + ManagedSparkSession.builder.projectId(test_project) + .location(test_region) + .sessionConfig(mismatched_region_config) + .getOrCreate() + ) + + assert isinstance(mismatch_exc_info.value.__cause__, InvalidArgument) + mismatch_msg = str(mismatch_exc_info.value) + assert mismatch_msg.startswith( + "Error while creating Managed Spark Session:" + ) + assert "us-east-4" in mismatch_msg + assert ManagedSparkSession._active_s8s_session_id is None diff --git a/tests/unit/test_init.py b/tests/unit/test_init.py index 12a1dad9..59c89084 100644 --- a/tests/unit/test_init.py +++ b/tests/unit/test_init.py @@ -1251,7 +1251,9 @@ def _boom(): print(json.dumps({ "type": f"{type(e).__module__}.{type(e).__qualname__}", "message": str(e), + "msg": getattr(e, "msg", None), "is_import_error": isinstance(e, ImportError), + "is_runtime_error": isinstance(e, RuntimeError), "name": getattr(e, "name", None), })) """ @@ -1329,7 +1331,9 @@ def test_missing_pyspark_with_missing_or_broken_ipython_recommends_client( def test_missing_pyspark_is_an_import_error_named_pyspark(self): error = self._run("plain", "absent") self.assertTrue(error["is_import_error"]) + self.assertFalse(error["is_runtime_error"]) self.assertEqual(error["name"], "pyspark") + self.assertEqual(error["msg"], error["message"]) def test_leftover_pyspark_namespace_dir_recommends_client(self): error = self._run("plain", "namespace") @@ -1355,7 +1359,9 @@ def test_old_pyspark_recommends_upgrading_via_spark_extra(self): # Caught by the same `except ImportError:` guards as a # missing pyspark. self.assertTrue(error["is_import_error"]) + self.assertFalse(error["is_runtime_error"]) self.assertEqual(error["name"], "pyspark") + self.assertEqual(error["msg"], error["message"]) self.assertEqual( error["message"], "PySpark 3.5.1 is installed, but google-cloud-spark-connect " diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 565340bc..74a82f70 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -22,10 +22,12 @@ Aborted, FailedPrecondition, InvalidArgument, + MethodNotImplemented, NotFound, PermissionDenied, ) from google.auth.exceptions import DefaultCredentialsError +from google.rpc import error_details_pb2 from google.cloud.managed_spark_connect import ManagedSparkSession from google.cloud.managed_spark_connect import ide from google.cloud.managed_spark_connect import session as session_module @@ -856,12 +858,302 @@ def test_create_spark_session_with_invalid_argument( mock_credentials.return_value = (cred, "") with self.assertRaises(ManagedSparkConnectException) as e: ManagedSparkSession.builder.sessionConfig(Session()).getOrCreate() - self.assertEqual( - e.exception.error_message, - "Error while creating Managed Spark Session: " - "400 Network does not have permissions", + self.assertEqual( + e.exception.message, + "Error while creating Managed Spark Session: " + "Network does not have permissions", + ) + self.assertIsInstance(e.exception.__cause__, InvalidArgument) + + @mock.patch("google.auth.default") + @mock.patch("google.cloud.dataproc_v1.SessionControllerClient") + def test_create_spark_session_with_typo_project_id( + self, + mock_session_controller_client, + mock_credentials, + ): + cred = mock.MagicMock() + cred.token = "token" + mock_credentials.return_value = (cred, "") + mock_session_controller_client.return_value.create_session.side_effect = PermissionDenied( + "Permission denied on resource project typo-project.", + details=[ + error_details_pb2.ErrorInfo( + reason="CONSUMER_INVALID", + domain="googleapis.com", + ), + error_details_pb2.DebugInfo( + detail="[ORIGINAL ERROR] generic::permission_denied: ..." + ), + error_details_pb2.DebugInfo( + detail="Project 'projects/typo-project' not found or deleted." + ), + ], + ) + + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("typo-project") + .location("us-central1") + .getOrCreate() + ) + + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Permission denied on resource project typo-project. " + "Project 'projects/typo-project' not found or deleted.", + ) + self.assertIsInstance(ctx.exception.__cause__, PermissionDenied) + + @mock.patch("google.auth.default") + @mock.patch("google.cloud.dataproc_v1.SessionControllerClient") + def test_create_spark_session_with_typo_region( + self, + mock_session_controller_client, + mock_credentials, + ): + cred = mock.MagicMock() + cred.token = "token" + mock_credentials.return_value = (cred, "") + mock_session_controller_client.return_value.create_session.side_effect = MethodNotImplemented( + "Received http2 header with status: 404. This usually indicates " + "that the 'api_endpoint' configuration in ClientOptions is " + "incorrect, contains a typo, or is an invalid regional endpoint " + "for this service." + ) + + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("test-project") + .location("us-east-4") + .getOrCreate() ) + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Regional endpoint 'us-east-4-dataproc.googleapis.com' " + "(derived from location 'us-east-4') returned HTTP 404; " + "check that 'us-east-4' is a valid Managed Spark region.", + ) + self.assertIsInstance(ctx.exception.__cause__, MethodNotImplemented) + + @mock.patch("google.auth.default") + @mock.patch("google.cloud.dataproc_v1.SessionControllerClient") + def test_create_spark_session_with_custom_endpoint_http_404( + self, + mock_session_controller_client, + mock_credentials, + ): + cred = mock.MagicMock() + cred.token = "token" + mock_credentials.return_value = (cred, "") + raw_msg = "Received http2 header with status: 404." + mock_session_controller_client.return_value.create_session.side_effect = MethodNotImplemented( + raw_msg + ) + + with mock.patch.dict( + os.environ, + { + "GOOGLE_CLOUD_DATAPROC_API_ENDPOINT": ( + "custom-dataproc.googleapis.com" + ) + }, + ): + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("test-project") + .location("us-central1") + .getOrCreate() + ) + + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Endpoint 'custom-dataproc.googleapis.com' " + "(location 'us-central1') returned HTTP 404: " + f"{raw_msg}", + ) + self.assertIsInstance(ctx.exception.__cause__, MethodNotImplemented) + + @mock.patch("google.auth.default") + @mock.patch("google.cloud.dataproc_v1.SessionControllerClient") + def test_create_spark_session_error_detail_fallbacks_and_field_violations( + self, + mock_session_controller_client, + mock_credentials, + ): + cred = mock.MagicMock() + cred.token = "token" + mock_credentials.return_value = (cred, "") + + # 1. CONSUMER_INVALID without DebugInfo detail + mock_session_controller_client.return_value.create_session.side_effect = PermissionDenied( + "Permission denied on resource project bad-proj", + details=[ + error_details_pb2.ErrorInfo( + reason="CONSUMER_INVALID", + domain="googleapis.com", + ) + ], + ) + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("bad-proj") + .location("us-central1") + .getOrCreate() + ) + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Permission denied on resource project bad-proj. " + "Project 'bad-proj' was not found or is invalid.", + ) + + # 2. RESOURCE_PROJECT_INVALID without DebugInfo detail + mock_session_controller_client.return_value.create_session.side_effect = InvalidArgument( + "Invalid resource field value in the request.", + details=[ + error_details_pb2.ErrorInfo( + reason="RESOURCE_PROJECT_INVALID", + domain="googleapis.com", + ) + ], + ) + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("BAD_PROJ!") + .location("us-central1") + .getOrCreate() + ) + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Invalid resource field value in the request. " + "Invalid project ID 'BAD_PROJ!'.", + ) + + # 3. LOCATION_POLICY_VIOLATED without DebugInfo detail + mock_session_controller_client.return_value.create_session.side_effect = PermissionDenied( + "Permission denied on 'locations/us-east-4' (or it may not exist).", + details=[ + error_details_pb2.ErrorInfo( + reason="LOCATION_POLICY_VIOLATED", + domain="googleapis.com", + ) + ], + ) + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("test-project") + .location("us-east-4") + .getOrCreate() + ) + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Permission denied on 'locations/us-east-4' (or it may not exist). " + "Invalid location 'us-east-4'.", + ) + + # 4. BadRequest.FieldViolation details + mock_session_controller_client.return_value.create_session.side_effect = InvalidArgument( + "Request contains an invalid argument.", + details=[ + error_details_pb2.BadRequest( + field_violations=[ + error_details_pb2.BadRequest.FieldViolation( + field="session.session_template", + description="Template name is malformed.", + ) + ] + ) + ], + ) + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("test-project") + .location("us-central1") + .getOrCreate() + ) + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Request contains an invalid argument. " + "session.session_template: Template name is malformed.", + ) + + @mock.patch("google.auth.default") + @mock.patch("google.cloud.dataproc_v1.SessionControllerClient") + def test_create_spark_session_with_typo_session_template( + self, + mock_session_controller_client, + mock_credentials, + ): + cred = mock.MagicMock() + cred.token = "token" + mock_credentials.return_value = (cred, "") + mock_session_controller_client.return_value.create_session.side_effect = InvalidArgument( + "Dataproc Session Template 'interactive-analysis-typo' does not exist" + ) + + session_config = Session() + session_config.session_template = ( + "projects/test-project/locations/us-central1/" + "sessionTemplates/interactive-analysis-typo" + ) + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("test-project") + .location("us-central1") + .sessionConfig(session_config) + .getOrCreate() + ) + + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Dataproc Session Template 'interactive-analysis-typo' does not exist", + ) + self.assertIsInstance(ctx.exception.__cause__, InvalidArgument) + + @mock.patch("google.auth.default") + @mock.patch("google.cloud.dataproc_v1.SessionControllerClient") + def test_create_spark_session_with_malformed_session_template_uri( + self, + mock_session_controller_client, + mock_credentials, + ): + cred = mock.MagicMock() + cred.token = "token" + mock_credentials.return_value = (cred, "") + mock_session_controller_client.return_value.create_session.side_effect = InvalidArgument( + "Request contains an invalid argument." + ) + + bad_template = ( + "projects/test-project/locations/us-central1/sessionTemplate/foo" + ) + with self.assertRaises(ManagedSparkConnectException) as ctx: + ( + ManagedSparkSession.builder.projectId("test-project") + .location("us-central1") + .sessionTemplate(bad_template) + .getOrCreate() + ) + + self.assertEqual( + ctx.exception.message, + "Error while creating Managed Spark Session: " + "Request contains an invalid argument. " + f"Invalid session_template '{bad_template}'; expected a " + "template ID or 'projects/{project}/locations/{location}/" + "sessionTemplates/{template}'.", + ) + self.assertIsInstance(ctx.exception.__cause__, InvalidArgument) + @mock.patch("google.auth.default") @mock.patch("google.cloud.dataproc_v1.SessionControllerClient") @mock.patch(