@@ -161,6 +161,21 @@ To test across clouds, on top of setting up Redis, you also need GCP / AWS / Sno
 
 Then run `make test-python-integration`. Note that for Snowflake / GCP / AWS, this will create new temporary tables / datasets.
 
+#### (Experimental) Run full integration tests against containerized services
+Test across clouds requires existing accounts on GCP / AWS / Snowflake, and may incur costs when using these services.
+
+For this approach of running tests, you'll need to have docker set up locally: [Get Docker](https://docs.docker.com/get-docker/)
+
+It's possible to run some integration tests against emulated local versions of these services, using ephemeral containers. 
+These tests create new temporary tables / datasets locally only, and they are cleaned up. when the containers are torn down.
+
+The services with containerized replacements currently implemented are:
+- Datastore
+- Redis
+
+You can run `make test-python-integration-container` to run tests against the containerized versions of dependencies.
+
+
 ## Feast Java Serving
 See [Java contributing guide](java/CONTRIBUTING.md)
 
 
@@ -67,6 +67,9 @@ test-python:
 test-python-integration:
 	FEAST_USAGE=False IS_TEST=True python -m pytest -n 8 --integration sdk/python/tests
 
+test-python-integration-container:
+	FEAST_USAGE=False IS_TEST=True FEAST_LOCAL_ONLINE_CONTAINER=True python -m pytest -n 8 --integration sdk/python/tests
+
 test-python-universal-contrib:
 	PYTHONPATH='.' FULL_REPO_CONFIGS_MODULE=sdk.python.feast.infra.offline_stores.contrib.contrib_repo_configuration FEAST_USAGE=False IS_TEST=True python -m pytest -n 8 --integration --universal sdk/python/tests
 
 
@@ -126,7 +126,7 @@
         "pytest-mock==1.10.4",
         "Sphinx!=4.0.0,<4.4.0",
         "sphinx-rtd-theme",
-        "testcontainers==3.4.2",
+        "testcontainers>=3.5",
         "adlfs==0.5.9",
         "firebase-admin==4.5.2",
         "pre-commit",
 
@@ -184,6 +184,8 @@ def cleanup():
         e.feature_store.teardown()
         if proc.is_alive():
             proc.kill()
+        if e.online_store_creator:
+            e.online_store_creator.teardown()
 
     request.addfinalizer(cleanup)
 
@@ -245,6 +247,8 @@ def go_data_sources(request, go_environment):
     def cleanup():
         # logger.info("Running cleanup in %s, Request: %s", worker_id, request.param)
         go_environment.data_source_creator.teardown()
+        if go_environment.online_store_creator:
+            go_environment.online_store_creator.teardown()
 
     request.addfinalizer(cleanup)
     return construct_universal_test_data(go_environment)
@@ -259,6 +263,8 @@ def e2e_data_sources(environment: Environment, request):
 
     def cleanup():
         environment.data_source_creator.teardown()
+        if environment.online_store_creator:
+            environment.online_store_creator.teardown()
 
     request.addfinalizer(cleanup)
 
 
@@ -1,15 +1,18 @@
 from dataclasses import dataclass
-from typing import Dict, Type, Union
+from typing import Dict, Optional, Type, Union
 
 from tests.integration.feature_repos.universal.data_source_creator import (
     DataSourceCreator,
 )
 from tests.integration.feature_repos.universal.data_sources.file import (
     FileDataSourceCreator,
 )
+from tests.integration.feature_repos.universal.online_store_creator import (
+    OnlineStoreCreator,
+)
 
 
-@dataclass(frozen=True)
+@dataclass(frozen=False)
 class IntegrationTestRepoConfig:
     """
     This class should hold all possible parameters that may need to be varied by individual tests.
@@ -19,19 +22,26 @@ class IntegrationTestRepoConfig:
     online_store: Union[str, Dict] = "sqlite"
 
     offline_store_creator: Type[DataSourceCreator] = FileDataSourceCreator
+    online_store_creator: Optional[Type[OnlineStoreCreator]] = None
 
     full_feature_names: bool = True
     infer_features: bool = False
     python_feature_server: bool = False
     go_feature_server: bool = False
 
     def __repr__(self) -> str:
-        if isinstance(self.online_store, str):
-            online_store_type = self.online_store
-        elif self.online_store["type"] == "redis":
-            online_store_type = self.online_store.get("redis_type", "redis")
+        if not self.online_store_creator:
+            if isinstance(self.online_store, str):
+                online_store_type = self.online_store
+            elif isinstance(self.online_store, dict):
+                if self.online_store["type"] == "redis":
+                    online_store_type = self.online_store.get("redis_type", "redis")
+                else:
+                    online_store_type = self.online_store["type"]
+            else:
+                online_store_type = self.online_store.__name__
         else:
-            online_store_type = self.online_store["type"]
+            online_store_type = self.online_store_creator.__name__
 
         return ":".join(
             [
 
@@ -44,6 +44,15 @@
     create_order_feature_view,
     create_pushable_feature_view,
 )
+from tests.integration.feature_repos.universal.online_store.datastore import (
+    DatastoreOnlineStoreCreator,
+)
+from tests.integration.feature_repos.universal.online_store.redis import (
+    RedisOnlineStoreCreator,
+)
+from tests.integration.feature_repos.universal.online_store_creator import (
+    OnlineStoreCreator,
+)
 
 DYNAMO_CONFIG = {"type": "dynamodb", "region": "us-west-2"}
 # Port 12345 will chosen as default for redis node configuration because Redis Cluster is started off of nodes
@@ -115,6 +124,18 @@
 else:
     FULL_REPO_CONFIGS = DEFAULT_FULL_REPO_CONFIGS
 
+if os.getenv("FEAST_LOCAL_ONLINE_CONTAINER", "False").lower() == "true":
+    replacements = {"datastore": DatastoreOnlineStoreCreator}
+    replacement_dicts = [(REDIS_CONFIG, RedisOnlineStoreCreator)]
+    for c in FULL_REPO_CONFIGS:
+        if isinstance(c.online_store, dict):
+            for _replacement in replacement_dicts:
+                if c.online_store == _replacement[0]:
+                    c.online_store_creator = _replacement[1]
+        elif c.online_store in replacements:
+            c.online_store_creator = replacements[c.online_store]
+
+
 GO_REPO_CONFIGS = [
     IntegrationTestRepoConfig(online_store=REDIS_CONFIG, go_feature_server=True,),
 ]
@@ -299,6 +320,7 @@ class Environment:
     data_source_creator: DataSourceCreator
     python_feature_server: bool
     worker_id: str
+    online_store_creator: Optional[OnlineStoreCreator] = None
 
     def __post_init__(self):
         self.end_date = datetime.utcnow().replace(microsecond=0, second=0, minute=0)
@@ -341,9 +363,16 @@ def construct_test_environment(
     project = f"{test_suite_name}_{run_id}_{run_num}"
 
     offline_creator: DataSourceCreator = test_repo_config.offline_store_creator(project)
-
     offline_store_config = offline_creator.create_offline_store_config()
-    online_store = test_repo_config.online_store
+
+    if test_repo_config.online_store_creator:
+        online_creator = test_repo_config.online_store_creator(project)
+        online_store = (
+            test_repo_config.online_store
+        ) = online_creator.create_online_store()
+    else:
+        online_creator = None
+        online_store = test_repo_config.online_store
 
     repo_dir_name = tempfile.mkdtemp()
 
@@ -392,6 +421,7 @@ def construct_test_environment(
         data_source_creator=offline_creator,
         python_feature_server=test_repo_config.python_feature_server,
         worker_id=worker_id,
+        online_store_creator=online_creator,
     )
 
     return environment
 
@@ -9,6 +9,9 @@
 
 
 class DataSourceCreator(ABC):
+    def __init__(self, project_name: str):
+        self.project_name = project_name
+
     @abstractmethod
     def create_data_source(
         self,
 
@@ -18,8 +18,8 @@ class BigQueryDataSourceCreator(DataSourceCreator):
     dataset: Optional[Dataset] = None
 
     def __init__(self, project_name: str):
+        super().__init__(project_name)
         self.client = bigquery.Client()
-        self.project_name = project_name
         self.gcp_project = self.client.project
         self.dataset_id = f"{self.gcp_project}.{project_name}"
 
 
@@ -22,7 +22,7 @@ class FileDataSourceCreator(DataSourceCreator):
     files: List[Any]
 
     def __init__(self, project_name: str):
-        self.project_name = project_name
+        super().__init__(project_name)
         self.files = []
 
     def create_data_source(
 
@@ -19,8 +19,7 @@ class RedshiftDataSourceCreator(DataSourceCreator):
     tables: List[str] = []
 
     def __init__(self, project_name: str):
-        super().__init__()
-        self.project_name = project_name
+        super().__init__(project_name)
         self.client = aws_utils.get_redshift_data_client("us-west-2")
         self.s3 = aws_utils.get_s3_resource("us-west-2")