Repository: great-expectations/great_expectations
Pull request #6844 (2023-01-19 22:32:30)

[FEATURE] Experimental `Datasource` and `DataAsset` connection testing
Changes proposed in this pull request:
- GREAT-1416
- Add `test_connection` methods to `Datasource`s and `DataAsset`s
- Call `test_connection` when adding a `Datasource` or `DataAsset`

### 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 | bbc4c0070ef7b5ce090aacbc729f2d4183839daf |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/niobium-lead-7998/deploys/63c9c5004df8840009bc3a85 |
codesee-maps[bot]: <details>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=6844&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=6844&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/6844/d0850a75/ac694362a0491252e51c107a8975174f9398903c.svg)](https://app.codesee.io/r/reviews?pr=6844&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)

#### Legend
![CodeSee Map Legend](https://app.codesee.io/Legend_v2.png)
netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *niobium-lead-7998* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | ac694362a0491252e51c107a8975174f9398903c |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/niobium-lead-7998/deploys/63c9c509c9e0780008ab869f |

Patch:
diff --git a/great_expectations/data_context/data_context/abstract_data_context.py b/great_expectations/data_context/data_context/abstract_data_context.py
index 591d77cd7..14d80ea99 100644
--- a/great_expectations/data_context/data_context/abstract_data_context.py
+++ b/great_expectations/data_context/data_context/abstract_data_context.py
@@ -1917,17 +1917,9 @@ class AbstractDataContext(ConfigPeer, ABC):
             )
         )
 
-        # We get a single batch_definition so we can get the execution_engine here. All batches will share the same one
+        # We get a single BatchData so we can get the execution_engine here. All batches will share the same one
         # So the batch itself doesn't matter. But we use -1 because that will be the latest batch loaded.
-        execution_engine: ExecutionEngine
-        if hasattr(batch_list[-1], "execution_engine"):
-            # 'XBatch's are execution engine aware. We just checked for this attr so we ignore the following
-            # attr defined mypy error
-            execution_engine = batch_list[-1].execution_engine
-        else:
-            execution_engine = self.datasources[  # type: ignore[union-attr]
-                batch_list[-1].batch_definition.datasource_name
-            ].execution_engine
+        execution_engine: ExecutionEngine = batch_list[-1].data.execution_engine
 
         validator = Validator(
             execution_engine=execution_engine,
diff --git a/great_expectations/experimental/datasources/interfaces.py b/great_expectations/experimental/datasources/interfaces.py
index 87608d9c6..4968401d2 100644
--- a/great_expectations/experimental/datasources/interfaces.py
+++ b/great_expectations/experimental/datasources/interfaces.py
@@ -47,6 +47,11 @@ except ImportError:
     LOGGER.debug("No spark sql dataframe module available.")
     pyspark_sql_Row = None
 
+
+class TestConnectionError(Exception):
+    pass
+
+
 # BatchRequestOptions is a dict that is composed into a BatchRequest that specifies the
 # Batches one wants as returned. The keys represent dimensions one can slice the data along
 # and the values are the realized. If a value is None or unspecified, the batch_request
@@ -149,6 +154,16 @@ class DataAsset(ExperimentalBaseModel):
         assert isinstance(ds, Datasource)
         self._datasource = ds
 
+    def test_connection(self) -> None:
+        """Test the connection for the DataAsset.
+
+        Raises:
+            TestConnectionError
+        """
+        raise NotImplementedError(
+            "One needs to implement 'test_connection' on a DataAsset subclass"
+        )
+
     # Abstract Methods
     def batch_request_options_template(
         self,
@@ -323,23 +338,10 @@ class Datasource(
     type: str
     name: str
     assets: MutableMapping[str, DataAssetType] = {}
-    _execution_engine: ExecutionEngine = pydantic.PrivateAttr()
-
-    def __init__(self, **kwargs):
-        super().__init__(**kwargs)
-        engine_kwargs = {
-            k: v for (k, v) in kwargs.items() if k not in self._excluded_eng_args
-        }
-        self._execution_engine = self._execution_engine_type()(**engine_kwargs)
 
-    @property
-    def execution_engine(self) -> ExecutionEngine:
-        return self._execution_engine
-
-    class Config:
-        # TODO: revisit this (1 option - define __get_validator__ on ExecutionEngine)
-        # https://pydantic-docs.helpmanual.io/usage/types/#custom-data-types
-        arbitrary_types_allowed = True
+    # private attrs
+    _cached_execution_engine_kwargs: Dict[str, Any] = pydantic.PrivateAttr({})
+    _execution_engine: ExecutionEngine | None = pydantic.PrivateAttr(None)
 
     @pydantic.validator("assets", pre=True)
     def _load_asset_subtype(cls, v: Dict[str, dict]):
@@ -362,6 +364,18 @@ class Datasource(
         """Returns the execution engine to be used"""
         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)
+        if (
+            current_execution_engine_kwargs != self._cached_execution_engine_kwargs
+            or not self._execution_engine
+        ):
+            self._execution_engine = self._execution_engine_type()(
+                **current_execution_engine_kwargs
+            )
+            self._cached_execution_engine_kwargs = current_execution_engine_kwargs
+        return self._execution_engine
+
     def get_batch_list_from_batch_request(
         self, batch_request: BatchRequest
     ) -> List[Batch]:
@@ -396,6 +410,7 @@ class Datasource(
         # The setter for datasource is non-functional, so we access _datasource directly.
         # See the comment in DataAsset for more information.
         asset._datasource = self
+        asset.test_connection()
         self.assets[asset.name] = asset
         return asset
 
@@ -407,6 +422,19 @@ class Datasource(
             "One needs to implement 'execution_engine_type' on a Datasource subclass"
         )
 
+    def test_connection(self, test_assets: bool = True) -> None:
+        """Test the connection for the Datasource.
+
+        Args:
+            test_assets: If assets have been passed to the Datasource, an attempt can be made to test them as well.
+
+        Raises:
+            TestConnectionError
+        """
+        raise NotImplementedError(
+            "One needs to implement 'test_connection' on a Datasource subclass"
+        )
+
     # End Abstract Methods
 
 
diff --git a/great_expectations/experimental/datasources/metadatasource.py b/great_expectations/experimental/datasources/metadatasource.py
index 432e2bc96..bc471cd2d 100644
--- a/great_expectations/experimental/datasources/metadatasource.py
+++ b/great_expectations/experimental/datasources/metadatasource.py
@@ -49,7 +49,9 @@ class MetaDatasource(pydantic.main.ModelMetaclass):
         def _datasource_factory(name: str, **kwargs) -> Datasource:
             # TODO: update signature to match Datasource __init__ (ex update __signature__)
             LOGGER.debug(f"5. Adding '{name}' {cls_name}")
-            return cls(name=name, **kwargs)
+            datasource = cls(name=name, **kwargs)
+            datasource.test_connection()
+            return datasource
 
         _datasource_factory.__doc__ = cls.__doc__
 
diff --git a/great_expectations/experimental/datasources/pandas_datasource.py b/great_expectations/experimental/datasources/pandas_datasource.py
index ccdb38579..71ca11d9a 100644
--- a/great_expectations/experimental/datasources/pandas_datasource.py
+++ b/great_expectations/experimental/datasources/pandas_datasource.py
@@ -22,6 +22,7 @@ from great_expectations.experimental.datasources.interfaces import (
     BatchSortersDefinition,
     DataAsset,
     Datasource,
+    TestConnectionError,
 )
 
 if TYPE_CHECKING:
@@ -44,6 +45,23 @@ class _DataFrameAsset(DataAsset):
         default="batch_request_param_"
     )
 
+    def test_connection(self) -> None:
+        """Test the connection for the CSVAsset.
+
+        Raises:
+            TestConnectionError
+        """
+        success = False
+        for filepath in self.path.iterdir():
+            if self.regex.match(filepath.name):
+                # if one file in the path matches the regex, we consider this asset valid
+                success = True
+                break
+        if not success:
+            raise TestConnectionError(
+                f"No file at path: {self.path} matched the regex: {self.regex}."
+            )
+
     def _fully_specified_batch_requests_with_path(
         self, batch_request: BatchRequest
     ) -> List[Tuple[BatchRequest, pathlib.Path]]:
@@ -148,7 +166,8 @@ class _DataFrameAsset(DataAsset):
                     },
                 ),
             )
-            data, markers = self.datasource.execution_engine.get_batch_data_and_markers(
+            execution_engine: ExecutionEngine = self.datasource.get_execution_engine()
+            data, markers = execution_engine.get_batch_data_and_markers(
                 batch_spec=batch_spec
             )
 
@@ -170,8 +189,9 @@ class _DataFrameAsset(DataAsset):
             batch_metadata = copy.deepcopy(request.options)
             batch_metadata["path"] = path
 
-            # Some pydantic annotations are postponed due to circular imports. This will set the annotations before we
-            # instantiate the Batch class since we can import them above.
+            # Some pydantic annotations are postponed due to circular imports.
+            # Batch.update_forward_refs() will set the annotations before we
+            # instantiate the Batch class since we can import them in this scope.
             Batch.update_forward_refs()
             batch_list.append(
                 Batch(
@@ -228,6 +248,20 @@ class PandasDatasource(Datasource):
 
         return PandasExecutionEngine
 
+    def test_connection(self, test_assets: bool = True) -> None:
+        """Test the connection for the PandasDatasource.
+
+        Args:
+            test_assets: If assets have been passed to the PandasDatasource, whether to test them as well.
+
+        Raises:
+            TestConnectionError
+        """
+        # Only self.assets can be tested for PandasDatasource
+        if self.assets and test_assets:
+            for asset in self.assets.values():
+                asset.test_connection()  # type: ignore[union-attr]
+
     def add_csv_asset(
         self,
         name: str,
diff --git a/great_expectations/experimental/datasources/postgres_datasource.py b/great_expectations/experimental/datasources/postgres_datasource.py
index 605ba3650..91afb7456 100644
--- a/great_expectations/experimental/datasources/postgres_datasource.py
+++ b/great_expectations/experimental/datasources/postgres_datasource.py
@@ -8,9 +8,8 @@ class PostgresDatasource(SQLDatasource):
     """Adds a postgres datasource to the data context.
 
     Args:
-        name: The name of this postgres datasource
-        connection_string: The SQLAlchemy connection string used to connect to the postgres
-            database.
+        name: The name of this postgres datasource.
+        connection_string: The SQLAlchemy connection string used to connect to the postgres database.
             For example: "postgresql+psycopg2://postgres:@localhost/test_database"
         assets: An optional dictionary whose keys are TableAsset names and whose values
             are TableAsset objects.
diff --git a/great_expectations/experimental/datasources/sources.py b/great_expectations/experimental/datasources/sources.py
index 7ddde97a4..42125eec3 100644
--- a/great_expectations/experimental/datasources/sources.py
+++ b/great_expectations/experimental/datasources/sources.py
@@ -61,7 +61,7 @@ class _SourceFactories:
         An `.add_pandas()` pandas factory method will be added to `context.sources`.
 
         >>> class PandasDatasource(Datasource):
-        >>>     type: str = 'pandas'`
+        >>>     type: str = 'pandas'
         >>>     asset_types = [FileAsset]
         >>>     execution_engine: PandasExecutionEngine
         """
diff --git a/great_expectations/experimental/datasources/sql_datasource.py b/great_expectations/experimental/datasources/sql_datasource.py
index 081a59bea..510c034d6 100644
--- a/great_expectations/experimental/datasources/sql_datasource.py
+++ b/great_expectations/experimental/datasources/sql_datasource.py
@@ -27,11 +27,18 @@ from great_expectations.experimental.datasources.interfaces import (
     BatchSortersDefinition,
     DataAsset,
     Datasource,
+    TestConnectionError,
 )
 
-if TYPE_CHECKING:
+SQLALCHEMY_IMPORTED = False
+try:
     import sqlalchemy
 
+    SQLALCHEMY_IMPORTED = True
+except ImportError:
+    pass
+
+if TYPE_CHECKING:
     from great_expectations.execution_engine import ExecutionEngine
 
 
@@ -39,6 +46,10 @@ class SQLDatasourceError(Exception):
     pass
 
 
+class SQLDatasourceWarning(UserWarning):
+    pass
+
+
 @pydantic_dc.dataclass(frozen=True)
 class ColumnSplitter:
     column_name: str
@@ -98,15 +109,12 @@ def _query_for_year_and_month(
         [sqlalchemy.engine.base.Connection, str, str], DatetimeRange
     ],
 ) -> Dict[str, List]:
-    # We should make an assertion about the execution_engine earlier. Right now it is assigned to
-    # after construction. We may be able to use a hook in a property setter.
     from great_expectations.execution_engine import SqlAlchemyExecutionEngine
 
-    assert isinstance(
-        table_asset.datasource.execution_engine, SqlAlchemyExecutionEngine
-    )
+    execution_engine = table_asset.datasource.get_execution_engine()
+    assert isinstance(execution_engine, SqlAlchemyExecutionEngine)
 
-    with table_asset.datasource.execution_engine.engine.connect() as conn:
+    with execution_engine.engine.connect() as conn:
         datetimes: DatetimeRange = query_datetime_range(
             conn,
             table_asset.table_name,
@@ -136,6 +144,30 @@ class TableAsset(DataAsset):
     column_splitter: Optional[SqlYearMonthSplitter] = None
     name: str
 
+    def test_connection(self) -> None:
+        """Test the connection for the TableAsset.
+
+        Raises:
+            TestConnectionError
+        """
+        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,
+        )
+        if not exists:
+            raise TestConnectionError(
+                f"Attempt to connect to table: {self.table_name} failed because the table "
+                "does not exist."
+            )
+
     def batch_request_options_template(
         self,
     ) -> BatchRequestOptions:
@@ -258,7 +290,8 @@ class TableAsset(DataAsset):
                 )
             # Creating the batch_spec is our hook into the execution engine.
             batch_spec = SqlAlchemyDatasourceBatchSpec(**batch_spec_kwargs)
-            data, markers = self.datasource.execution_engine.get_batch_data_and_markers(
+            execution_engine: ExecutionEngine = self.datasource.get_execution_engine()
+            data, markers = execution_engine.get_batch_data_and_markers(
                 batch_spec=batch_spec
             )
 
@@ -277,8 +310,9 @@ class TableAsset(DataAsset):
                 batch_spec_passthrough=None,
             )
 
-            # Some pydantic annotations are postponed due to circular imports. This will set the annotations before we
-            # instantiate the Batch class since we can import them above.
+            # Some pydantic annotations are postponed due to circular imports.
+            # Batch.update_forward_refs() will set the annotations before we
+            # instantiate the Batch class since we can import them in this scope.
             Batch.update_forward_refs()
             batch_list.append(
                 Batch(
@@ -300,7 +334,7 @@ class SQLDatasource(Datasource):
     """Adds a generic SQL datasource to the data context.
 
     Args:
-        name: The name of this datasource
+        name: The name of this datasource.
         connection_string: The SQLAlchemy connection string used to connect to the database.
             For example: "postgresql+psycopg2://postgres:@localhost/test_database"
         assets: An optional dictionary whose keys are TableAsset names and whose values
@@ -316,6 +350,9 @@ class SQLDatasource(Datasource):
     connection_string: str
     assets: Dict[str, TableAsset] = {}
 
+    # private attrs
+    _engine: sqlalchemy.engine.Engine = pydantic.PrivateAttr()
+
     @property
     def execution_engine_type(self) -> Type[ExecutionEngine]:
         """Returns the default execution engine type."""
@@ -323,6 +360,53 @@ 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"]:
+                try:
+                    self._engine = sqlalchemy.create_engine(kwargs["connection_string"])
+                except Exception as e:
+                    # connection_string has passed pydantic validation,
+                    # but still fails to create a sqlalchemy engine
+                    raise SQLDatasourceError(
+                        "Unable to create a SQLAlchemy engine from "
+                        f"connection_string: {kwargs['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:
+        return self._engine
+
+    def test_connection(self, test_assets: bool = True) -> None:
+        """Test the connection for the SQLDatasource.
+
+        Args:
+            test_assets: If assets have been passed to the SQLDatasource, whether to test them as well.
+
+        Raises:
+            TestConnectionError
+        """
+        try:
+            engine: sqlalchemy.engine.Engine = self.get_engine()
+            engine.connect()
+        except Exception as e:
+            raise TestConnectionError(
+                "Attempt to connect to datasource failed with the following error message: "
+                f"{str(e)}"
+            )
+        if self.assets and test_assets:
+            for asset in self.assets.values():
+                asset.test_connection()
+
     def add_table_asset(
         self,
         name: str,
diff --git a/great_expectations/experimental/datasources/sqlite_datasource.py b/great_expectations/experimental/datasources/sqlite_datasource.py
index d9abd2945..c7176fbfb 100644
--- a/great_expectations/experimental/datasources/sqlite_datasource.py
+++ b/great_expectations/experimental/datasources/sqlite_datasource.py
@@ -4,7 +4,6 @@ from datetime import datetime
 from typing import TYPE_CHECKING, Dict, List, Optional, Type
 
 import pydantic
-from pydantic import constr
 from pydantic import dataclasses as pydantic_dc
 from typing_extensions import ClassVar
 
@@ -13,14 +12,6 @@ from great_expectations.experimental.datasources.interfaces import DataAsset
 if TYPE_CHECKING:
     import sqlalchemy
 
-    # The SqliteConnectionString type is defined this way because mypy can't handle
-    # constraint types. See: https://github.com/pydantic/pydantic/issues/156
-    # which suggests the solution here:
-    # https://github.com/pydantic/pydantic/issues/975#issuecomment-551147305
-    SqliteConnectionString = str
-else:
-    SqliteConnectionString = constr(regex=r"^sqlite")
-
 from typing_extensions import Literal
 
 from great_expectations.experimental.datasources.sql_datasource import (
@@ -91,11 +82,21 @@ def _get_sqlite_datetime_range(
     return DatetimeRange(min=min_max_dt[0], max=min_max_dt[1])
 
 
+class SqliteDsn(pydantic.AnyUrl):
+    allowed_schemes = {
+        "sqlite",
+        "sqlite+pysqlite",
+        "sqlite+aiosqlite",
+        "sqlite+pysqlcipher",
+    }
+    host_required = False
+
+
 class SqliteDatasource(SQLDatasource):
     """Adds a sqlite datasource to the data context.
 
     Args:
-        name: The name of this sqlite datasource
+        name: The name of this sqlite datasource.
         connection_string: The SQLAlchemy connection string used to connect to the sqlite database.
             For example: "sqlite:///path/to/file.db"
         assets: An optional dictionary whose keys are TableAsset names and whose values
@@ -109,7 +110,7 @@ class SqliteDatasource(SQLDatasource):
     # right side of the operator determines the type name
     # left side enforces the names on instance creation
     type: Literal["sqlite"] = "sqlite"  # type: ignore[assignment]
-    connection_string: SqliteConnectionString
+    connection_string: SqliteDsn
     assets: Dict[str, SqliteTableAsset] = {}  # type: ignore[assignment]
 
     def add_table_asset(


Test patch:
diff --git a/tests/experimental/datasources/test_config.py b/tests/experimental/datasources/test_config.py
index c91a4c942..6a76db469 100644
--- a/tests/experimental/datasources/test_config.py
+++ b/tests/experimental/datasources/test_config.py
@@ -3,6 +3,7 @@ import json
 import pathlib
 from pprint import pformat as pf
 from typing import Callable, List
+from unittest import mock
 
 import pydantic
 import pytest
@@ -134,7 +135,8 @@ COMBINED_ZEP_AND_OLD_STYLE_CFG_DICT = {
     ],
 )
 def test_load_config(inject_engine_lookup_double, load_method: Callable, input_):
-    loaded: GxConfig = load_method(input_)
+    with mock.patch("sqlalchemy.create_engine"):
+        loaded: GxConfig = load_method(input_)
     pp(loaded)
     assert loaded
 
@@ -240,7 +242,8 @@ def test_catch_bad_asset_configs(
     print(f"  Config\n{pf(config)}\n")
 
     with pytest.raises(pydantic.ValidationError) as exc_info:
-        GxConfig.parse_obj({"xdatasources": config})
+        with mock.patch("sqlalchemy.create_engine"):
+            GxConfig.parse_obj({"xdatasources": config})
 
     print(f"\n{exc_info.typename}:{exc_info.value}")
 
@@ -286,7 +289,8 @@ def test_general_column_splitter_errors(
 @pytest.fixture
 @functools.lru_cache(maxsize=1)
 def from_dict_gx_config() -> GxConfig:
-    gx_config = GxConfig.parse_obj(PG_COMPLEX_CONFIG_DICT)
+    with mock.patch("sqlalchemy.create_engine"):
+        gx_config = GxConfig.parse_obj(PG_COMPLEX_CONFIG_DICT)
     assert gx_config
     return gx_config
 
@@ -294,7 +298,8 @@ def from_dict_gx_config() -> GxConfig:
 @pytest.fixture
 @functools.lru_cache(maxsize=1)
 def from_json_gx_config() -> GxConfig:
-    gx_config = GxConfig.parse_raw(PG_COMPLEX_CONFIG_JSON)
+    with mock.patch("sqlalchemy.create_engine"):
+        gx_config = GxConfig.parse_raw(PG_COMPLEX_CONFIG_JSON)
     assert gx_config
     return gx_config
 
@@ -302,7 +307,8 @@ def from_json_gx_config() -> GxConfig:
 @pytest.fixture
 @functools.lru_cache(maxsize=1)
 def from_yaml_gx_config() -> GxConfig:
-    gx_config = GxConfig.parse_yaml(PG_CONFIG_YAML_STR)
+    with mock.patch("sqlalchemy.create_engine"):
+        gx_config = GxConfig.parse_yaml(PG_CONFIG_YAML_STR)
     assert gx_config
     return gx_config
 
@@ -313,7 +319,8 @@ def test_dict_config_round_trip(
     dumped: dict = from_dict_gx_config.dict()
     print(f"  Dumped Dict ->\n\n{pf(dumped)}\n")
 
-    re_loaded: GxConfig = GxConfig.parse_obj(dumped)
+    with mock.patch("sqlalchemy.create_engine"):
+        re_loaded: GxConfig = GxConfig.parse_obj(dumped)
     pp(re_loaded)
     assert re_loaded
 
@@ -326,7 +333,8 @@ def test_json_config_round_trip(
     dumped: str = from_json_gx_config.json(indent=2)
     print(f"  Dumped JSON ->\n\n{dumped}\n")
 
-    re_loaded: GxConfig = GxConfig.parse_raw(dumped)
+    with mock.patch("sqlalchemy.create_engine"):
+        re_loaded: GxConfig = GxConfig.parse_raw(dumped)
     pp(re_loaded)
     assert re_loaded
 
@@ -339,7 +347,8 @@ def test_yaml_config_round_trip(
     dumped: str = from_yaml_gx_config.yaml()
     print(f"  Dumped YAML ->\n\n{dumped}\n")
 
-    re_loaded: GxConfig = GxConfig.parse_yaml(dumped)
+    with mock.patch("sqlalchemy.create_engine"):
+        re_loaded: GxConfig = GxConfig.parse_yaml(dumped)
     pp(re_loaded)
     assert re_loaded
 
@@ -358,7 +367,8 @@ def test_yaml_file_config_round_trip(
 
     print(f"  yaml_file -> \n\n{yaml_file.read_text()}")
 
-    re_loaded: GxConfig = GxConfig.parse_yaml(yaml_file)
+    with mock.patch("sqlalchemy.create_engine"):
+        re_loaded: GxConfig = GxConfig.parse_yaml(yaml_file)
     pp(re_loaded)
     assert re_loaded
 
diff --git a/tests/experimental/datasources/test_metadatasource.py b/tests/experimental/datasources/test_metadatasource.py
index 91b7c14aa..351540151 100644
--- a/tests/experimental/datasources/test_metadatasource.py
+++ b/tests/experimental/datasources/test_metadatasource.py
@@ -153,6 +153,9 @@ class TestMisconfiguredMetaDatasource:
                 def execution_engine_type(self) -> Type[ExecutionEngine]:
                     return DummyExecutionEngine
 
+                def test_connection(self) -> None:
+                    ...
+
         # check that no types were registered
         assert len(empty_sources.type_lookup) < 1
 
@@ -162,8 +165,11 @@ class TestMisconfiguredMetaDatasource:
         class MissingExecEngineTypeDatasource(Datasource):
             type: str = "valid"
 
+            def test_connection(self) -> None:
+                ...
+
         with pytest.raises(NotImplementedError):
-            MissingExecEngineTypeDatasource(name="name")
+            MissingExecEngineTypeDatasource(name="name").get_execution_engine()
 
     def test_ds_assets_type_field_not_set(self, empty_sources: _SourceFactories):
 
@@ -183,9 +189,23 @@ class TestMisconfiguredMetaDatasource:
                 def execution_engine_type(self) -> Type[ExecutionEngine]:
                     return DummyExecutionEngine
 
+                def test_connection(self) -> None:
+                    ...
+
         # check that no types were registered
         assert len(empty_sources.type_lookup) < 1
 
+    def test_ds_test_connection_not_defined(self, empty_sources: _SourceFactories):
+        class MissingTestConnectionDatasource(Datasource):
+            type: str = "valid"
+
+            @property
+            def execution_engine_type(self) -> Type[ExecutionEngine]:
+                return DummyExecutionEngine
+
+        with pytest.raises(NotImplementedError):
+            MissingTestConnectionDatasource(name="name").test_connection()
+
 
 def test_minimal_ds_to_asset_flow(context_sources_cleanup):
     # 1. Define Datasource & Assets
@@ -204,6 +224,9 @@ def test_minimal_ds_to_asset_flow(context_sources_cleanup):
         def execution_engine_type(self) -> Type[ExecutionEngine]:
             return DummyExecutionEngine
 
+        def test_connection(self):
+            ...
+
         def add_red_asset(self, asset_name: str) -> RedAsset:
             asset = RedAsset(name=asset_name)
             self.assets[asset_name] = asset
diff --git a/tests/experimental/datasources/test_postgres_datasource.py b/tests/experimental/datasources/test_postgres_datasource.py
index b3fa4c907..1a3076833 100644
--- a/tests/experimental/datasources/test_postgres_datasource.py
+++ b/tests/experimental/datasources/test_postgres_datasource.py
@@ -1,6 +1,7 @@
 from contextlib import contextmanager
 from pprint import pprint
-from typing import Callable, ContextManager
+from typing import Callable, ContextManager, Optional, Tuple
+from unittest import mock
 
 import pytest
 from pydantic import ValidationError
@@ -44,10 +45,12 @@ def _source(
     original_override = PostgresDatasource.execution_engine_override
     try:
         PostgresDatasource.execution_engine_override = execution_eng_cls
-        yield PostgresDatasource(
-            name="my_datasource",
-            connection_string=connection_string,
-        )
+        with mock.patch("sqlalchemy.create_engine"):
+            postgres_datasource = PostgresDatasource(
+                name="my_datasource",
+                connection_string=connection_string,
+            )
+        yield postgres_datasource
     finally:
         PostgresDatasource.execution_engine_override = original_override
 
@@ -65,7 +68,7 @@ def test_construct_postgres_datasource(create_source):
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
         assert source.name == "my_datasource"
-        assert isinstance(source.execution_engine, SqlAlchemyExecutionEngine)
+        assert source.execution_engine_type is SqlAlchemyExecutionEngine
         assert source.assets == {}
 
 
@@ -95,7 +98,8 @@ def test_add_table_asset_with_splitter(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        with mock.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
         assert asset == list(source.assets.values())[0]
@@ -119,7 +123,8 @@ def test_add_table_asset_with_no_splitter(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        with mock.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]
         assert_table_asset(
@@ -153,18 +158,34 @@ def test_construct_table_asset_directly_with_no_splitter(create_source):
         assert_batch_request(asset.get_batch_request(), "my_datasource", "my_asset", {})
 
 
+def create_and_add_table_asset_without_testing_connection(
+    source: PostgresDatasource,
+    name: str,
+    table_name: str,
+    column_splitter: Optional[SqlYearMonthSplitter] = None,
+) -> Tuple[PostgresDatasource, TableAsset]:
+    table_asset = TableAsset(
+        name=name,
+        table_name=table_name,
+        column_splitter=column_splitter,
+    )
+    # TODO: asset custom init
+    table_asset._datasource = source
+    source.assets[table_asset.name] = table_asset
+    return source, table_asset
+
+
 @pytest.mark.unit
 def test_construct_table_asset_directly_with_splitter(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = TableAsset(
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source,
             name="my_asset",
             table_name="my_table",
             column_splitter=SqlYearMonthSplitter(column_name="col"),
         )
-        # TODO: asset custom init
-        asset._datasource = source
         assert_table_asset(
             asset,
             "my_asset",
@@ -194,7 +215,9 @@ def test_datasource_gets_batch_list_no_splitter(create_source):
     with create_source(
         validate_batch_spec=validate_batch_spec, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         source.get_batch_list_from_batch_request(asset.get_batch_request())
 
 
@@ -237,7 +260,9 @@ def test_datasource_gets_batch_list_splitter_with_unspecified_batch_request_opti
     with create_source(
         validate_batch_spec=collect_batch_spec, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
         empty_batch_request = asset.get_batch_request()
         assert empty_batch_request.options == {}
@@ -258,7 +283,9 @@ def test_datasource_gets_batch_list_splitter_with_batch_request_options_set_to_n
     with create_source(
         validate_batch_spec=collect_batch_spec, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
         batch_request_with_none = asset.get_batch_request(
             asset.batch_request_options_template()
@@ -282,7 +309,9 @@ def test_datasource_gets_batch_list_splitter_with_partially_specified_batch_requ
     with create_source(
         validate_batch_spec=collect_batch_spec, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
         batches = source.get_batch_list_from_batch_request(
             asset.get_batch_request({"year": 2022})
@@ -303,7 +332,7 @@ def test_datasource_gets_batch_list_splitter_with_partially_specified_batch_requ
         metadatas = [batch.metadata for batch in batches]
         for month in _DEFAULT_TEST_MONTHS:
             expected_metadata = {"month": month, "year": 2022}
-            expected_metadata in metadatas
+            assert expected_metadata in metadatas
 
 
 @pytest.mark.unit
@@ -323,7 +352,9 @@ def test_datasource_gets_batch_list_with_fully_specified_batch_request_options(
     with create_source(
         validate_batch_spec=validate_batch_spec, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
         batches = source.get_batch_list_from_batch_request(
             asset.get_batch_request({"month": 1, "year": 2022})
@@ -358,7 +389,9 @@ def test_bad_batch_request_passed_into_get_batch_list_from_batch_request(
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
 
         src, ast, op = batch_request_args
@@ -387,7 +420,9 @@ def test_get_batch_list_from_batch_request_with_good_batch_request(
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
         batch_request = BatchRequest(
             datasource_name=source.name,
@@ -414,7 +449,9 @@ def test_get_batch_list_from_batch_request_with_malformed_batch_request(
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
         src, ast, op = batch_request_args
         batch_request = BatchRequest(
@@ -430,7 +467,9 @@ def test_get_bad_batch_request(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col")
         with pytest.raises(ge_exceptions.InvalidBatchRequestError):
             asset.get_batch_request({"invalid_key": None})
@@ -497,7 +536,9 @@ def test_sort_batch_list_by_metadata(sort_info, create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col").add_sorters(sort_keys)
         batch_request = BatchRequest(
             datasource_name=source.name,
@@ -523,7 +564,9 @@ def test_sort_batch_list_by_unknown_key(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col").add_sorters(
             ["yr", "month"]
         )
@@ -565,7 +608,9 @@ def test_data_source_json_has_properties(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col").add_sorters(
             ["year", "month"]
         )
@@ -579,7 +624,9 @@ def test_data_source_str_has_properties(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col").add_sorters(
             ["year", "month"]
         )
@@ -592,7 +639,9 @@ def test_datasource_dict_has_properties(create_source):
     with create_source(
         validate_batch_spec=lambda _: None, dialect="postgresql"
     ) as source:
-        asset = source.add_table_asset(name="my_asset", table_name="my_table")
+        source, asset = create_and_add_table_asset_without_testing_connection(
+            source=source, name="my_asset", table_name="my_table"
+        )
         asset.add_year_and_month_splitter(column_name="my_col").add_sorters(
             ["year", "month"]
         )
diff --git a/tests/experimental/datasources/test_sqlite_datasource.py b/tests/experimental/datasources/test_sqlite_datasource.py
index 87126cc30..f0cd15ab3 100644
--- a/tests/experimental/datasources/test_sqlite_datasource.py
+++ b/tests/experimental/datasources/test_sqlite_datasource.py
@@ -5,6 +5,7 @@ from pydantic import ValidationError
 
 from great_expectations.data_context.util import file_relative_path
 from great_expectations.experimental.datasources import SqliteDatasource
+from great_expectations.experimental.datasources.sqlite_datasource import SqliteDsn
 
 
 @pytest.mark.unit
@@ -36,8 +37,14 @@ def test_connection_string_starts_with_sqlite():
 def test_connection_string_that_does_not_start_with_sqlite():
     name = "sqlite_datasource"
     connection_string = "stuff+sqlite:///path/to/database/file.db"
-    with pytest.raises(ValidationError):
+    with pytest.raises(ValidationError) as e:
         SqliteDatasource(
             name=name,
             connection_string=connection_string,
         )
+    assert str(e.value) == (
+        "1 validation error for SqliteDatasource\n"
+        "connection_string\n"
+        "  URL scheme not permitted (type=value_error.url.scheme; "
+        f"allowed_schemes={SqliteDsn.allowed_schemes})"
+    )