Repository: great-expectations/great_expectations
Pull request #7019 (2023-02-01 14:14:03)

[FEATURE] Experimental `DataAsset` `test_connection`
Changes proposed in this pull request:
- GREAT-1417
- Add `schema_name` parameter to `TableAsset`
- Add tests for `Datasource` `test_connection` errors and their associated `assets`

### Definition of Done
- [X] My code follows the Great Expectations [style guide](https://docs.greatexpectations.io/docs/contributing/style_guides/code_style)
- [X] I have performed a [self-review](https://docs.greatexpectations.io/docs/contributing/contributing_checklist) of my own code
- [X] I have commented my code, particularly in hard-to-understand areas
- [X] I have made corresponding changes to the documentation
- [X] I have added [unit tests](https://docs.greatexpectations.io/docs/contributing/contributing_test#writing-unit-and-integration-tests) where applicable and made sure that new and existing tests are passing.
- [X] I have run any local integration tests and made sure that nothing is broken.

netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *niobium-lead-7998* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 9b43d132038819a70f608805910a5b6467248089 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/niobium-lead-7998/deploys/63da73ad5c711800086a812a |
codesee-maps[bot]: <details>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=7019&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations">Click on the image</a> for a new way to code review</b>
</summary>

  - Make big changes easier — review code in small groups of related files
  - Know where to start — see the whole change at a glance
  - Take a code tour — explore the change with an interactive tour
  - Make comments and review — all fully sync’ed with github

    [Try it now!](https://app.codesee.io/r/reviews?pr=7019&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)
</details>

[![Review these changes using an interactive CodeSee Map](https://s3.us-east-2.amazonaws.com/maps.codesee.io/images/github/great-expectations/great_expectations/7019/0628f7e7/9b43d132038819a70f608805910a5b6467248089.svg)](https://app.codesee.io/r/reviews?pr=7019&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)

#### Legend
![CodeSee Map Legend](https://app.codesee.io/Legend_v2.png)

Patch:
diff --git a/great_expectations/experimental/datasources/filesystem_data_asset.py b/great_expectations/experimental/datasources/filesystem_data_asset.py
index 27b453045..a94e7b5ce 100644
--- a/great_expectations/experimental/datasources/filesystem_data_asset.py
+++ b/great_expectations/experimental/datasources/filesystem_data_asset.py
@@ -70,6 +70,11 @@ to use as its "include" directive for File-Path style DataAsset processing."""
         Raises:
             TestConnectionError: If the connection test fails.
         """
+        if not self.base_directory.exists():
+            raise TestConnectionError(
+                f"Path: {self.base_directory.resolve()} does not exist."
+            )
+
         success = False
         for filepath in self.base_directory.iterdir():
             if self.regex.match(filepath.name):
@@ -78,7 +83,7 @@ to use as its "include" directive for File-Path style DataAsset processing."""
                 break
         if not success:
             raise TestConnectionError(
-                f"No file at path: {self.base_directory} matched the regex: {self.regex}."
+                f"No file at path: {self.base_directory} matched the regex: {self.regex.pattern}"
             )
 
     def _fully_specified_batch_requests_with_path(
diff --git a/great_expectations/experimental/datasources/interfaces.py b/great_expectations/experimental/datasources/interfaces.py
index f7cacc794..79f1d6fa1 100644
--- a/great_expectations/experimental/datasources/interfaces.py
+++ b/great_expectations/experimental/datasources/interfaces.py
@@ -161,7 +161,7 @@ class DataAsset(ExperimentalBaseModel):
         """Test the connection for the DataAsset.
 
         Raises:
-            TestConnectionError
+            TestConnectionError: If the connection test fails.
         """
         raise NotImplementedError(
             """One needs to implement "test_connection" on a DataAsset subclass."""
@@ -327,7 +327,7 @@ class Datasource(
     # class attrs
     asset_types: ClassVar[List[Type[DataAsset]]] = []
     # Datasource instance attrs but these will be fed into the `execution_engine` constructor
-    _excluded_eng_args: ClassVar[Set[str]] = {
+    _EXCLUDED_EXEC_ENG_ARGS: ClassVar[Set[str]] = {
         "name",
         "type",
         "execution_engine",
@@ -344,7 +344,7 @@ class Datasource(
 
     # private attrs
     _cached_execution_engine_kwargs: Dict[str, Any] = pydantic.PrivateAttr({})
-    _execution_engine: ExecutionEngine | None = pydantic.PrivateAttr(None)
+    _execution_engine: Union[ExecutionEngine, None] = pydantic.PrivateAttr(None)
 
     @pydantic.validator("assets", each_item=True)
     @classmethod
@@ -378,7 +378,9 @@ class Datasource(
         return self.execution_engine_override or self.execution_engine_type
 
     def get_execution_engine(self) -> ExecutionEngine:
-        current_execution_engine_kwargs = self.dict(exclude=self._excluded_eng_args)
+        current_execution_engine_kwargs = self.dict(
+            exclude=self._EXCLUDED_EXEC_ENG_ARGS
+        )
         if (
             current_execution_engine_kwargs != self._cached_execution_engine_kwargs
             or not self._execution_engine
diff --git a/great_expectations/experimental/datasources/schemas/PostgresDatasource.json b/great_expectations/experimental/datasources/schemas/PostgresDatasource.json
index f93b6e9d9..4443c8a82 100644
--- a/great_expectations/experimental/datasources/schemas/PostgresDatasource.json
+++ b/great_expectations/experimental/datasources/schemas/PostgresDatasource.json
@@ -124,6 +124,10 @@
                 "table_name": {
                     "title": "Table Name",
                     "type": "string"
+                },
+                "schema_name": {
+                    "title": "Schema Name",
+                    "type": "string"
                 }
             },
             "required": [
diff --git a/great_expectations/experimental/datasources/schemas/SQLDatasource.json b/great_expectations/experimental/datasources/schemas/SQLDatasource.json
index 5fd576cab..abecd5472 100644
--- a/great_expectations/experimental/datasources/schemas/SQLDatasource.json
+++ b/great_expectations/experimental/datasources/schemas/SQLDatasource.json
@@ -121,6 +121,10 @@
                 "table_name": {
                     "title": "Table Name",
                     "type": "string"
+                },
+                "schema_name": {
+                    "title": "Schema Name",
+                    "type": "string"
                 }
             },
             "required": [
diff --git a/great_expectations/experimental/datasources/schemas/TableAsset.json b/great_expectations/experimental/datasources/schemas/TableAsset.json
index 2b57c6a21..f55bd65d6 100644
--- a/great_expectations/experimental/datasources/schemas/TableAsset.json
+++ b/great_expectations/experimental/datasources/schemas/TableAsset.json
@@ -28,6 +28,10 @@
         "table_name": {
             "title": "Table Name",
             "type": "string"
+        },
+        "schema_name": {
+            "title": "Schema Name",
+            "type": "string"
         }
     },
     "required": [
diff --git a/great_expectations/experimental/datasources/spark_datasource.py b/great_expectations/experimental/datasources/spark_datasource.py
index 6c70ee996..ca355c4f0 100644
--- a/great_expectations/experimental/datasources/spark_datasource.py
+++ b/great_expectations/experimental/datasources/spark_datasource.py
@@ -66,7 +66,7 @@ class SparkDatasource(Datasource):
             test_assets: If assets have been passed to the SparkDatasource, whether to test them as well.
 
         Raises:
-            TestConnectionError
+            TestConnectionError: If the connection test fails.
         """
         # Only self.assets can be tested for PandasDatasource
         if self.assets and test_assets:
diff --git a/great_expectations/experimental/datasources/sql_datasource.py b/great_expectations/experimental/datasources/sql_datasource.py
index cd8ee9ef3..cae86427a 100644
--- a/great_expectations/experimental/datasources/sql_datasource.py
+++ b/great_expectations/experimental/datasources/sql_datasource.py
@@ -159,14 +159,6 @@ class _SQLAsset(DataAsset):
     column_splitter: Optional[SqlYearMonthSplitter] = None
     name: str
 
-    def test_connection(self) -> None:
-        """Test the connection for the _SQLAsset.
-
-        Raises:
-            TestConnectionError
-        """
-        assert isinstance(self.datasource, SQLDatasource)
-
     def batch_request_options_template(
         self,
     ) -> BatchRequestOptions:
@@ -269,6 +261,7 @@ class _SQLAsset(DataAsset):
 
         batch_list: List[Batch] = []
         column_splitter = self.column_splitter
+        batch_spec_kwargs: dict[str, str | dict | None]
         for request in self._fully_specified_batch_requests(batch_request):
             batch_metadata = copy.deepcopy(request.options)
             batch_spec_kwargs = self._create_batch_spec_kwargs()
@@ -347,6 +340,9 @@ class QueryAsset(_SQLAsset):
     type: Literal["query"] = "query"  # type: ignore[assignment]
     query: str
 
+    def test_connection(self) -> None:
+        pass
+
     @pydantic.validator("query")
     def query_must_start_with_select(cls, v: str):
         query = v.lstrip()
@@ -376,29 +372,38 @@ class TableAsset(_SQLAsset):
     # Instance fields
     type: Literal["table"] = "table"  # type: ignore[assignment]
     table_name: str
+    schema_name: Optional[str] = None
 
     def test_connection(self) -> None:
         """Test the connection for the TableAsset.
 
         Raises:
-            TestConnectionError
+            TestConnectionError: If the connection test fails.
         """
         assert isinstance(self.datasource, SQLDatasource)
-        schema: Optional[str] = None
-        table_name: str
-        if "." in self.table_name:
-            schema, table_name = self.table_name.split(".")
-        else:
-            table_name = self.table_name
         engine: sqlalchemy.engine.Engine = self.datasource.get_engine()
-        exists = sqlalchemy.inspect(engine).has_table(
-            table_name=table_name,
-            schema=schema,
+        inspector: sqlalchemy.engine.Inspector = sqlalchemy.inspect(engine)
+
+        table_str = (
+            f"{self.schema_name}.{self.table_name}"
+            if self.schema_name
+            else self.table_name
+        )
+
+        if self.schema_name and self.schema_name not in inspector.get_schema_names():
+            raise TestConnectionError(
+                f'Attempt to connect to table: "{table_str}" failed because the schema '
+                f'"{self.schema_name}" does not exist.'
+            )
+
+        table_exists = sqlalchemy.inspect(engine).has_table(
+            table_name=self.table_name,
+            schema=self.schema_name,
         )
-        if not exists:
+        if not table_exists:
             raise TestConnectionError(
-                f"Attempt to connect to table: {self.table_name} failed because the table "
-                "does not exist."
+                f'Attempt to connect to table: "{table_str}" failed because the table '
+                f'"{self.table_name}" does not exist.'
             )
 
     def as_selectable(self) -> sqlalchemy.sql.Selectable:
@@ -415,6 +420,7 @@ class TableAsset(_SQLAsset):
             "type": "table",
             "data_asset_name": self.name,
             "table_name": self.table_name,
+            "schema_name": self.schema_name,
             "batch_identifiers": {},
         }
 
@@ -442,7 +448,8 @@ class SQLDatasource(Datasource):
     assets: Dict[str, Union[TableAsset, QueryAsset]] = {}
 
     # private attrs
-    _engine: sqlalchemy.engine.Engine = pydantic.PrivateAttr()
+    _cached_connection_string: str = pydantic.PrivateAttr("")
+    _engine: Union[sqlalchemy.engine.Engine, None] = pydantic.PrivateAttr(None)
 
     @property
     def execution_engine_type(self) -> Type[ExecutionEngine]:
@@ -451,30 +458,25 @@ class SQLDatasource(Datasource):
 
         return SqlAlchemyExecutionEngine
 
-    class Config:
-        arbitrary_types_allowed = True
-
-    def __init__(self, **kwargs) -> None:
-        super().__init__(**kwargs)
-        # validate that SQL Alchemy was successfully imported and attempt to create an engine
-        if SQLALCHEMY_IMPORTED:
-            if "connection_string" in kwargs and kwargs["connection_string"]:
+    def get_engine(self) -> sqlalchemy.engine.Engine:
+        if self.connection_string != self._cached_connection_string or not self._engine:
+            # validate that SQL Alchemy was successfully imported and attempt to create an engine
+            if SQLALCHEMY_IMPORTED:
                 try:
-                    self._engine = sqlalchemy.create_engine(kwargs["connection_string"])
+                    self._engine = sqlalchemy.create_engine(self.connection_string)
                 except Exception as e:
-                    # connection_string has passed pydantic validation,
-                    # but still fails to create a sqlalchemy engine
+                    # connection_string has passed pydantic validation, but still fails to create a sqlalchemy engine
+                    # one possible case is a missing plugin (e.g. psycopg2)
                     raise SQLDatasourceError(
                         "Unable to create a SQLAlchemy engine from "
-                        f"connection_string: {kwargs['connection_string']} due to the "
+                        f"connection_string: {self.connection_string} due to the "
                         f"following exception: {str(e)}"
-                    )
-        else:
-            raise SQLDatasourceError(
-                "Unable to create SQLDatasource due to missing sqlalchemy dependency."
-            )
-
-    def get_engine(self) -> sqlalchemy.engine.Engine:
+                    ) from e
+                self._cached_connection_string = self.connection_string
+            else:
+                raise SQLDatasourceError(
+                    "Unable to create SQLDatasource due to missing sqlalchemy dependency."
+                )
         return self._engine
 
     def test_connection(self, test_assets: bool = True) -> None:
@@ -484,7 +486,7 @@ class SQLDatasource(Datasource):
             test_assets: If assets have been passed to the SQLDatasource, whether to test them as well.
 
         Raises:
-            TestConnectionError
+            TestConnectionError: If the connection test fails.
         """
         try:
             engine: sqlalchemy.engine.Engine = self.get_engine()
@@ -493,15 +495,17 @@ class SQLDatasource(Datasource):
             raise TestConnectionError(
                 "Attempt to connect to datasource failed with the following error message: "
                 f"{str(e)}"
-            )
+            ) from e
         if self.assets and test_assets:
             for asset in self.assets.values():
+                asset._datasource = self
                 asset.test_connection()
 
     def add_table_asset(
         self,
         name: str,
         table_name: str,
+        schema_name: Optional[str] = None,
         order_by: Optional[BatchSortersDefinition] = None,
     ) -> TableAsset:
         """Adds a table asset to this datasource.
@@ -509,6 +513,7 @@ class SQLDatasource(Datasource):
         Args:
             name: The name of this table asset.
             table_name: The table where the data resides.
+            schema_name: The schema that holds the table.
             order_by: A list of BatchSorters or BatchSorter strings.
 
         Returns:
@@ -517,6 +522,7 @@ class SQLDatasource(Datasource):
         asset = TableAsset(
             name=name,
             table_name=table_name,
+            schema_name=schema_name,
             order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
             # see DataAsset._parse_order_by_sorter()
         )
diff --git a/great_expectations/experimental/datasources/sqlite_datasource.py b/great_expectations/experimental/datasources/sqlite_datasource.py
index 99781094b..134186dc2 100644
--- a/great_expectations/experimental/datasources/sqlite_datasource.py
+++ b/great_expectations/experimental/datasources/sqlite_datasource.py
@@ -157,6 +157,7 @@ class SqliteDatasource(SQLDatasource):
         self,
         name: str,
         table_name: str,
+        schema_name: Optional[str] = None,
         order_by: Optional[BatchSortersDefinition] = None,
     ) -> SqliteTableAsset:
         """Adds a sqlite table asset to this sqlite datasource.
@@ -164,6 +165,7 @@ class SqliteDatasource(SQLDatasource):
         Args:
             name: The name of this table asset.
             table_name: The table where the data resides.
+            schema_name: The schema that holds the table.
             order_by: A list of BatchSorters or BatchSorter strings.
 
         Returns:
@@ -172,6 +174,7 @@ class SqliteDatasource(SQLDatasource):
         asset = SqliteTableAsset(
             name=name,
             table_name=table_name,
+            schema_name=schema_name,
             order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
             # see TableAsset._parse_order_by_sorter()
         )


Test patch:
diff --git a/tests/experimental/datasources/conftest.py b/tests/experimental/datasources/conftest.py
index d78e5cee0..a0839d003 100644
--- a/tests/experimental/datasources/conftest.py
+++ b/tests/experimental/datasources/conftest.py
@@ -52,7 +52,15 @@ class _MockConnection:
         return [(DEFAULT_MIN_DT, DEFAULT_MAX_DT)]
 
 
-class _MockSaEngine:
+class MockSaInspector:
+    def get_schema_names(self):
+        ...
+
+    def has_table(self):
+        ...
+
+
+class MockSaEngine:
     def __init__(self, dialect: Dialect):
         self.dialect = dialect
 
@@ -78,7 +86,7 @@ def sqlachemy_execution_engine_mock_cls(
             # We should likely let the user pass in an engine. In a SqlAlchemyExecutionEngine used in
             # non-mocked code the engine property is of the type:
             # from sqlalchemy.engine import Engine as SaEngine
-            self.engine = _MockSaEngine(dialect=Dialect(dialect))
+            self.engine = MockSaEngine(dialect=Dialect(dialect))
 
         def get_batch_data_and_markers(  # type: ignore[override]
             self, batch_spec: SqlAlchemyDatasourceBatchSpec
diff --git a/tests/experimental/datasources/test_config.py b/tests/experimental/datasources/test_config.py
index 1e237bc3e..da5e5ebd5 100644
--- a/tests/experimental/datasources/test_config.py
+++ b/tests/experimental/datasources/test_config.py
@@ -4,7 +4,6 @@ import pathlib
 import re
 from pprint import pformat as pf
 from typing import Callable, List
-from unittest import mock
 
 import pydantic
 import pytest
@@ -215,8 +214,7 @@ class TestExcludeUnsetAssetFields:
     ],
 )
 def test_load_config(inject_engine_lookup_double, load_method: Callable, input_):
-    with mock.patch("sqlalchemy.create_engine"):
-        loaded: GxConfig = load_method(input_)
+    loaded: GxConfig = load_method(input_)
     pp(loaded)
     assert loaded
 
@@ -335,8 +333,7 @@ def test_catch_bad_asset_configs(
     print(f"  Config\n{pf(config)}\n")
 
     with pytest.raises(pydantic.ValidationError) as exc_info:
-        with mock.patch("sqlalchemy.create_engine"):
-            GxConfig.parse_obj({"xdatasources": config})
+        GxConfig.parse_obj({"xdatasources": config})
 
     print(f"\n{exc_info.typename}:{exc_info.value}")
 
@@ -386,8 +383,7 @@ def test_general_column_splitter_errors(
 @pytest.fixture
 @functools.lru_cache(maxsize=1)
 def from_dict_gx_config() -> GxConfig:
-    with mock.patch("sqlalchemy.create_engine"):
-        gx_config = GxConfig.parse_obj(PG_COMPLEX_CONFIG_DICT)
+    gx_config = GxConfig.parse_obj(PG_COMPLEX_CONFIG_DICT)
     assert gx_config
     return gx_config
 
@@ -395,8 +391,7 @@ def from_dict_gx_config() -> GxConfig:
 @pytest.fixture
 @functools.lru_cache(maxsize=1)
 def from_json_gx_config() -> GxConfig:
-    with mock.patch("sqlalchemy.create_engine"):
-        gx_config = GxConfig.parse_raw(PG_COMPLEX_CONFIG_JSON)
+    gx_config = GxConfig.parse_raw(PG_COMPLEX_CONFIG_JSON)
     assert gx_config
     return gx_config
 
@@ -404,8 +399,7 @@ def from_json_gx_config() -> GxConfig:
 @pytest.fixture
 @functools.lru_cache(maxsize=1)
 def from_yaml_gx_config() -> GxConfig:
-    with mock.patch("sqlalchemy.create_engine"):
-        gx_config = GxConfig.parse_yaml(PG_CONFIG_YAML_STR)
+    gx_config = GxConfig.parse_yaml(PG_CONFIG_YAML_STR)
     assert gx_config
     return gx_config
 
@@ -416,8 +410,7 @@ def test_dict_config_round_trip(
     dumped: dict = from_dict_gx_config.dict()
     print(f"  Dumped Dict ->\n\n{pf(dumped)}\n")
 
-    with mock.patch("sqlalchemy.create_engine"):
-        re_loaded: GxConfig = GxConfig.parse_obj(dumped)
+    re_loaded: GxConfig = GxConfig.parse_obj(dumped)
     pp(re_loaded)
     assert re_loaded
 
@@ -430,8 +423,7 @@ def test_json_config_round_trip(
     dumped: str = from_json_gx_config.json(indent=2)
     print(f"  Dumped JSON ->\n\n{dumped}\n")
 
-    with mock.patch("sqlalchemy.create_engine"):
-        re_loaded: GxConfig = GxConfig.parse_raw(dumped)
+    re_loaded: GxConfig = GxConfig.parse_raw(dumped)
     pp(re_loaded)
     assert re_loaded
 
@@ -444,8 +436,7 @@ def test_yaml_config_round_trip(
     dumped: str = from_yaml_gx_config.yaml()
     print(f"  Dumped YAML ->\n\n{dumped}\n")
 
-    with mock.patch("sqlalchemy.create_engine"):
-        re_loaded: GxConfig = GxConfig.parse_yaml(dumped)
+    re_loaded: GxConfig = GxConfig.parse_yaml(dumped)
     pp(re_loaded)
     assert re_loaded
 
@@ -464,8 +455,7 @@ def test_yaml_file_config_round_trip(
 
     print(f"  yaml_file -> \n\n{yaml_file.read_text()}")
 
-    with mock.patch("sqlalchemy.create_engine"):
-        re_loaded: GxConfig = GxConfig.parse_yaml(yaml_file)
+    re_loaded: GxConfig = GxConfig.parse_yaml(yaml_file)
     pp(re_loaded)
     assert re_loaded
 
diff --git a/tests/experimental/datasources/test_pandas_datasource.py b/tests/experimental/datasources/test_pandas_datasource.py
index 88657e207..3743b52d1 100644
--- a/tests/experimental/datasources/test_pandas_datasource.py
+++ b/tests/experimental/datasources/test_pandas_datasource.py
@@ -13,6 +13,7 @@ from pytest import MonkeyPatch, param
 
 import great_expectations.exceptions as ge_exceptions
 import great_expectations.execution_engine.pandas_execution_engine
+from great_expectations.experimental.datasources.interfaces import TestConnectionError
 from great_expectations.experimental.datasources.pandas_datasource import (
     CSVAsset,
     JSONAsset,
@@ -434,3 +435,56 @@ def test_pandas_sorter(
             metadata = batches[batch_index].metadata
             assert metadata[key1] == range1
             assert metadata[key2] == range2
+
+
+def bad_base_directory_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
+    base_directory = pathlib.Path("/this/path/is/not/here")
+    regex = re.compile(r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
+    test_connection_error = TestConnectionError(
+        f"Path: {base_directory.resolve()} does not exist."
+    )
+    return base_directory, regex, test_connection_error
+
+
+def bad_regex_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
+    relative_path = pathlib.Path(
+        "..", "..", "test_sets", "taxi_yellow_tripdata_samples"
+    )
+    base_directory = (
+        pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
+    )
+    regex = re.compile(r"green_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
+    test_connection_error = TestConnectionError(
+        f"No file at path: {base_directory.resolve()} matched the regex: {regex.pattern}"
+    )
+    return base_directory, regex, test_connection_error
+
+
+@pytest.fixture(params=[bad_base_directory_config, bad_regex_config])
+def datasource_test_connection_error_messages(
+    pandas_datasource: PandasDatasource, request
+) -> tuple[PandasDatasource, TestConnectionError]:
+    base_directory, regex, test_connection_error = request.param()
+    csv_asset = CSVAsset(  # type: ignore[call-arg]
+        name="csv_asset",
+        base_directory=base_directory,
+        regex=regex,
+    )
+    pandas_datasource.assets = {"csv_asset": csv_asset}
+    return pandas_datasource, test_connection_error
+
+
+@pytest.mark.unit
+def test_test_connection_failures(
+    datasource_test_connection_error_messages: tuple[
+        PandasDatasource, TestConnectionError
+    ]
+):
+    (
+        pandas_datasource,
+        test_connection_error,
+    ) = datasource_test_connection_error_messages
+
+    with pytest.raises(type(test_connection_error)) as e:
+        pandas_datasource.test_connection()
+    assert str(e.value) == str(test_connection_error)
diff --git a/tests/experimental/datasources/test_postgres_datasource.py b/tests/experimental/datasources/test_postgres_datasource.py
index b0501b7fa..68bf24f19 100644
--- a/tests/experimental/datasources/test_postgres_datasource.py
+++ b/tests/experimental/datasources/test_postgres_datasource.py
@@ -1,7 +1,8 @@
+from __future__ import annotations
+
 from contextlib import contextmanager
 from pprint import pprint
 from typing import Callable, ContextManager, Generator, Optional, Tuple
-from unittest import mock
 
 import pytest
 from pydantic import ValidationError
@@ -14,6 +15,7 @@ from great_expectations.experimental.datasources.interfaces import (
     BatchRequest,
     BatchRequestOptions,
     BatchSorter,
+    TestConnectionError,
 )
 from great_expectations.experimental.datasources.postgres_datasource import (
     PostgresDatasource,
@@ -25,6 +27,9 @@ from great_expectations.experimental.datasources.sql_datasource import (
 from tests.experimental.datasources.conftest import (
     DEFAULT_MAX_DT,
     DEFAULT_MIN_DT,
+    Dialect,
+    MockSaEngine,
+    MockSaInspector,
     sqlachemy_execution_engine_mock_cls,
 )
 
@@ -46,12 +51,10 @@ def _source(
     original_override = PostgresDatasource.execution_engine_override
     try:
         PostgresDatasource.execution_engine_override = execution_eng_cls
-        with mock.patch("sqlalchemy.create_engine"):
-            postgres_datasource = PostgresDatasource(
-                name="my_datasource",
-                connection_string=connection_string,  # type: ignore[arg-type] # coerced
-            )
-        yield postgres_datasource
+        yield PostgresDatasource(
+            name="my_datasource",
+            connection_string=connection_string,  # type: ignore[arg-type] # coerced
+        )
     finally:
         PostgresDatasource.execution_engine_override = original_override
 
@@ -98,11 +101,13 @@ def assert_batch_request(
 
 
 @pytest.mark.unit
-def test_add_table_asset_with_splitter(create_source: CreateSourceFixture):
+def test_add_table_asset_with_splitter(mocker, create_source: CreateSourceFixture):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        with mock.patch("sqlalchemy.inspect"):
+        with mocker.patch("sqlalchemy.create_engine"), mocker.patch(
+            "sqlalchemy.engine"
+        ), mocker.patch("sqlalchemy.inspect"):
             asset = source.add_table_asset(name="my_asset", table_name="my_table")
         asset.add_year_and_month_splitter("my_column")
         assert len(source.assets) == 1
@@ -123,11 +128,13 @@ def test_add_table_asset_with_splitter(create_source: CreateSourceFixture):
 
 
 @pytest.mark.unit
-def test_add_table_asset_with_no_splitter(create_source: CreateSourceFixture):
+def test_add_table_asset_with_no_splitter(mocker, create_source: CreateSourceFixture):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        with mock.patch("sqlalchemy.inspect"):
+        with mocker.patch("sqlalchemy.create_engine"), mocker.patch(
+            "sqlalchemy.engine"
+        ), mocker.patch("sqlalchemy.inspect"):
             asset = source.add_table_asset(name="my_asset", table_name="my_table")
         assert len(source.assets) == 1
         assert asset == list(source.assets.values())[0]
@@ -213,6 +220,7 @@ def test_datasource_gets_batch_list_no_splitter(create_source):
             "batch_identifiers": {},
             "data_asset_name": "my_asset",
             "table_name": "my_table",
+            "schema_name": None,
             "type": "table",
         }
 
@@ -235,6 +243,7 @@ def assert_batch_specs_correct_with_year_month_splitter_defaults(batch_specs):
                 "type": "table",
                 "data_asset_name": "my_asset",
                 "table_name": "my_table",
+                "schema_name": None,
                 "batch_identifiers": {"my_col": {"year": year, "month": month}},
                 "splitter_method": "split_on_year_and_month",
                 "splitter_kwargs": {"column_name": "my_col"},
@@ -326,6 +335,7 @@ def test_datasource_gets_batch_list_splitter_with_partially_specified_batch_requ
                 "type": "table",
                 "data_asset_name": "my_asset",
                 "table_name": "my_table",
+                "schema_name": None,
                 "batch_identifiers": {"my_col": {"year": 2022, "month": month}},
                 "splitter_method": "split_on_year_and_month",
                 "splitter_kwargs": {"column_name": "my_col"},
@@ -350,6 +360,7 @@ def test_datasource_gets_batch_list_with_fully_specified_batch_request_options(
             "splitter_kwargs": {"column_name": "my_col"},
             "splitter_method": "split_on_year_and_month",
             "table_name": "my_table",
+            "schema_name": None,
             "type": "table",
         }
 
@@ -702,6 +713,137 @@ def test_validate_invalid_postgres_connection_string(
             pass
 
 
+def bad_connection_string_config() -> tuple[
+    str, str, str, str | None, TestConnectionError
+]:
+    connection_string = "postgresql+psycopg2://postgres:@localhost/bad_database"
+    table_name = "good_table"
+    schema_name = "good_schema"
+    mock_sqlalchemy_engine_connect_error_raised = (
+        '(psycopg2.OperationalError) connection to server at "localhost" (::1), port '
+        '5432 failed: FATAL:  database "bad_database" does not exist\n'
+        "\n"
+        "(Background on this error at: https://sqlalche.me/e/14/e3q8)"
+    )
+    test_connection_error = TestConnectionError(
+        "Attempt to connect to datasource failed with the following error message: "
+        f"{mock_sqlalchemy_engine_connect_error_raised}"
+    )
+    return (
+        connection_string,
+        table_name,
+        schema_name,
+        mock_sqlalchemy_engine_connect_error_raised,
+        test_connection_error,
+    )
+
+
+def bad_table_name_config() -> tuple[str, str, str, str | None, TestConnectionError]:
+    connection_string = "postgresql+psycopg2://postgres:@localhost/test_ci"
+    table_name = "bad_table"
+    schema_name = "good_schema"
+    mock_sqlalchemy_engine_connect_error_raised = None
+    test_connection_error = TestConnectionError(
+        'Attempt to connect to table: "good_schema.bad_table" failed because the table '
+        '"bad_table" does not exist.'
+    )
+    return (
+        connection_string,
+        table_name,
+        schema_name,
+        mock_sqlalchemy_engine_connect_error_raised,
+        test_connection_error,
+    )
+
+
+def bad_schema_name_config() -> tuple[str, str, str, str | None, TestConnectionError]:
+    connection_string = "postgresql+psycopg2://postgres:@localhost/test_ci"
+    table_name = "good_table"
+    schema_name = "bad_schema"
+    mock_sqlalchemy_engine_connect_error_raised = None
+    test_connection_error = TestConnectionError(
+        'Attempt to connect to table: "bad_schema.good_table" failed because the schema '
+        '"bad_schema" does not exist.'
+    )
+    return (
+        connection_string,
+        table_name,
+        schema_name,
+        mock_sqlalchemy_engine_connect_error_raised,
+        test_connection_error,
+    )
+
+
+@pytest.fixture(
+    params=[
+        bad_connection_string_config,
+        bad_table_name_config,
+        bad_schema_name_config,
+    ]
+)
+def datasource_test_connection_error_messages(
+    request,
+) -> tuple[PostgresDatasource, str | None, TestConnectionError]:
+    (
+        connection_string,
+        table_name,
+        schema_name,
+        mock_sqlalchemy_engine_connect_error_raised,
+        test_connection_error,
+    ) = request.param()
+    table_asset = TableAsset(
+        name="table_asset",
+        table_name=table_name,
+        schema_name=schema_name,
+    )
+    postgres_datasource = PostgresDatasource(
+        name="postgres_datasource",
+        connection_string=connection_string,
+        assets={"table_asset": table_asset},
+    )
+    return (
+        postgres_datasource,
+        mock_sqlalchemy_engine_connect_error_raised,
+        test_connection_error,
+    )
+
+
+@pytest.mark.unit
+def test_test_connection_failures(
+    mocker,
+    datasource_test_connection_error_messages: tuple[
+        PostgresDatasource, str | None, TestConnectionError
+    ],
+):
+    (
+        postgres_datasource,
+        mock_sqlalchemy_engine_connect_error_raised,
+        test_connection_error,
+    ) = datasource_test_connection_error_messages
+
+    create_engine = mocker.patch("sqlalchemy.create_engine")
+    create_engine.return_value = MockSaEngine(dialect=Dialect("postgresql"))
+    connect = mocker.patch(
+        "tests.experimental.datasources.conftest.MockSaEngine.connect"
+    )
+    if mock_sqlalchemy_engine_connect_error_raised:
+        connect.side_effect = Exception(mock_sqlalchemy_engine_connect_error_raised)
+    inspect = mocker.patch("sqlalchemy.inspect")
+    inspect.return_value = MockSaInspector()
+    get_schema_names = mocker.patch(
+        "tests.experimental.datasources.conftest.MockSaInspector.get_schema_names"
+    )
+    get_schema_names.return_value = ["good_schema"]
+    has_table = mocker.patch(
+        "tests.experimental.datasources.conftest.MockSaInspector.has_table"
+    )
+    has_table.return_value = False
+
+    with pytest.raises(type(test_connection_error)) as e:
+        postgres_datasource.test_connection()
+    assert str(e.value) == str(test_connection_error)
+
+
 @pytest.mark.unit
 def test_query_data_asset(create_source):
     query = "SELECT * FROM my_table"
diff --git a/tests/experimental/datasources/test_spark_datasource.py b/tests/experimental/datasources/test_spark_datasource.py
index bf812aca0..6e35fdd35 100644
--- a/tests/experimental/datasources/test_spark_datasource.py
+++ b/tests/experimental/datasources/test_spark_datasource.py
@@ -1,5 +1,8 @@
+from __future__ import annotations
+
 import logging
 import pathlib
+import re
 from dataclasses import dataclass
 from typing import List
 
@@ -9,6 +12,7 @@ import great_expectations.exceptions as ge_exceptions
 from great_expectations.alias_types import PathStr
 from great_expectations.experimental.datasources.interfaces import (
     BatchSortersDefinition,
+    TestConnectionError,
 )
 from great_expectations.experimental.datasources.spark_datasource import (
     CSVSparkAsset,
@@ -294,3 +298,56 @@ def test_spark_sorter(
             metadata = batches[batch_index].metadata
             assert metadata[key1] == range1
             assert metadata[key2] == range2
+
+
+def bad_base_directory_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
+    base_directory = pathlib.Path("/this/path/is/not/here")
+    regex = re.compile(r"yellow_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
+    test_connection_error = TestConnectionError(
+        f"Path: {base_directory.resolve()} does not exist."
+    )
+    return base_directory, regex, test_connection_error
+
+
+def bad_regex_config() -> tuple[pathlib.Path, re.Pattern, TestConnectionError]:
+    relative_path = pathlib.Path(
+        "..", "..", "test_sets", "taxi_yellow_tripdata_samples"
+    )
+    base_directory = (
+        pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
+    )
+    regex = re.compile(r"green_tripdata_sample_(?P<year>\d{4})-(?P<month>\d{2}).csv")
+    test_connection_error = TestConnectionError(
+        f"No file at path: {base_directory.resolve()} matched the regex: {regex.pattern}",
+    )
+    return base_directory, regex, test_connection_error
+
+
+@pytest.fixture(params=[bad_base_directory_config, bad_regex_config])
+def datasource_test_connection_error_messages(
+    spark_datasource: SparkDatasource, request
+) -> tuple[SparkDatasource, TestConnectionError]:
+    base_directory, regex, test_connection_error = request.param()
+    csv_spark_asset = CSVSparkAsset(
+        name="csv_spark_asset",
+        base_directory=base_directory,
+        regex=regex,
+    )
+    spark_datasource.assets = {"csv_spark_asset": csv_spark_asset}
+    return spark_datasource, test_connection_error
+
+
+@pytest.mark.unit
+def test_test_connection_failures(
+    datasource_test_connection_error_messages: tuple[
+        SparkDatasource, TestConnectionError
+    ]
+):
+    (
+        spark_datasource,
+        test_connection_error,
+    ) = datasource_test_connection_error_messages
+
+    with pytest.raises(type(test_connection_error)) as e:
+        spark_datasource.test_connection()
+    assert str(e.value) == str(test_connection_error)