diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 77467576e4..2a4cb74634 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -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)
 
diff --git a/Makefile b/Makefile
index 9358001f3c..8d56f57516 100644
--- a/Makefile
+++ b/Makefile
@@ -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
 
diff --git a/sdk/python/setup.py b/sdk/python/setup.py
index 1853f144d9..ee8625e39d 100644
--- a/sdk/python/setup.py
+++ b/sdk/python/setup.py
@@ -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",
diff --git a/sdk/python/tests/conftest.py b/sdk/python/tests/conftest.py
index 1254604a0b..15e8ffc201 100644
--- a/sdk/python/tests/conftest.py
+++ b/sdk/python/tests/conftest.py
@@ -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)
 
diff --git a/sdk/python/tests/integration/feature_repos/integration_test_repo_config.py b/sdk/python/tests/integration/feature_repos/integration_test_repo_config.py
index 25650eced9..99e8512007 100644
--- a/sdk/python/tests/integration/feature_repos/integration_test_repo_config.py
+++ b/sdk/python/tests/integration/feature_repos/integration_test_repo_config.py
@@ -1,5 +1,5 @@
 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,
@@ -7,9 +7,12 @@
 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,6 +22,7 @@ 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
@@ -26,12 +30,18 @@ class IntegrationTestRepoConfig:
     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(
             [
diff --git a/sdk/python/tests/integration/feature_repos/repo_configuration.py b/sdk/python/tests/integration/feature_repos/repo_configuration.py
index 18f1ece8eb..cf53e76c58 100644
--- a/sdk/python/tests/integration/feature_repos/repo_configuration.py
+++ b/sdk/python/tests/integration/feature_repos/repo_configuration.py
@@ -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
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_source_creator.py b/sdk/python/tests/integration/feature_repos/universal/data_source_creator.py
index 2a13cff3be..7c58c80e49 100644
--- a/sdk/python/tests/integration/feature_repos/universal/data_source_creator.py
+++ b/sdk/python/tests/integration/feature_repos/universal/data_source_creator.py
@@ -9,6 +9,9 @@
 
 
 class DataSourceCreator(ABC):
+    def __init__(self, project_name: str):
+        self.project_name = project_name
+
     @abstractmethod
     def create_data_source(
         self,
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/bigquery.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/bigquery.py
index cb7113bf66..186ce8457d 100644
--- a/sdk/python/tests/integration/feature_repos/universal/data_sources/bigquery.py
+++ b/sdk/python/tests/integration/feature_repos/universal/data_sources/bigquery.py
@@ -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}"
 
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/file.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/file.py
index 4ae067728c..20974cf469 100644
--- a/sdk/python/tests/integration/feature_repos/universal/data_sources/file.py
+++ b/sdk/python/tests/integration/feature_repos/universal/data_sources/file.py
@@ -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(
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/redshift.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/redshift.py
index db007d83ad..795fcfbbbb 100644
--- a/sdk/python/tests/integration/feature_repos/universal/data_sources/redshift.py
+++ b/sdk/python/tests/integration/feature_repos/universal/data_sources/redshift.py
@@ -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")
 
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/snowflake.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/snowflake.py
index 5be3b7383e..c4852ccbc3 100644
--- a/sdk/python/tests/integration/feature_repos/universal/data_sources/snowflake.py
+++ b/sdk/python/tests/integration/feature_repos/universal/data_sources/snowflake.py
@@ -20,8 +20,7 @@ class SnowflakeDataSourceCreator(DataSourceCreator):
     tables: List[str] = []
 
     def __init__(self, project_name: str):
-        super().__init__()
-        self.project_name = project_name
+        super().__init__(project_name)
         self.offline_store_config = SnowflakeOfflineStoreConfig(
             type="snowflake.offline",
             account=os.environ["SNOWFLAKE_CI_DEPLOYMENT"],
diff --git a/sdk/python/tests/integration/feature_repos/universal/data_sources/spark_data_source_creator.py b/sdk/python/tests/integration/feature_repos/universal/data_sources/spark_data_source_creator.py
index 2724db50a6..49a4f539d6 100644
--- a/sdk/python/tests/integration/feature_repos/universal/data_sources/spark_data_source_creator.py
+++ b/sdk/python/tests/integration/feature_repos/universal/data_sources/spark_data_source_creator.py
@@ -24,6 +24,7 @@ class SparkDataSourceCreator(DataSourceCreator):
     spark_session = None
 
     def __init__(self, project_name: str):
+        super().__init__(project_name)
         self.spark_conf = {
             "master": "local[*]",
             "spark.ui.enabled": "false",
@@ -31,7 +32,6 @@ def __init__(self, project_name: str):
             "spark.sql.parser.quotedRegexColumnNames": "true",
             "spark.sql.session.timeZone": "UTC",
         }
-        self.project_name = project_name
         if not self.spark_offline_store_config:
             self.create_offline_store_config()
         if not self.spark_session:
diff --git a/sdk/python/tests/integration/feature_repos/universal/online_store/__init__.py b/sdk/python/tests/integration/feature_repos/universal/online_store/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/sdk/python/tests/integration/feature_repos/universal/online_store/datastore.py b/sdk/python/tests/integration/feature_repos/universal/online_store/datastore.py
new file mode 100644
index 0000000000..52851e80d8
--- /dev/null
+++ b/sdk/python/tests/integration/feature_repos/universal/online_store/datastore.py
@@ -0,0 +1,38 @@
+import os
+from typing import Dict
+
+from google.cloud import datastore
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.waiting_utils import wait_for_logs
+
+from tests.integration.feature_repos.universal.online_store_creator import (
+    OnlineStoreCreator,
+)
+
+
+class DatastoreOnlineStoreCreator(OnlineStoreCreator):
+    def __init__(self, project_name: str):
+        super().__init__(project_name)
+        self.container = (
+            DockerContainer(
+                "gcr.io/google.com/cloudsdktool/cloud-sdk:380.0.0-emulators"
+            )
+            .with_command(
+                "gcloud beta emulators datastore start --project test-project --host-port 0.0.0.0:8081"
+            )
+            .with_exposed_ports("8081")
+        )
+
+    def create_online_store(self) -> Dict[str, str]:
+        self.container.start()
+        log_string_to_wait_for = r"\[datastore\] Dev App Server is now running"
+        wait_for_logs(
+            container=self.container, predicate=log_string_to_wait_for, timeout=5
+        )
+        exposed_port = self.container.get_exposed_port("8081")
+        os.environ[datastore.client.DATASTORE_EMULATOR_HOST] = f"0.0.0.0:{exposed_port}"
+        return {"type": "datastore", "project_id": "test-project"}
+
+    def teardown(self):
+        del os.environ[datastore.client.DATASTORE_EMULATOR_HOST]
+        self.container.stop()
diff --git a/sdk/python/tests/integration/feature_repos/universal/online_store/redis.py b/sdk/python/tests/integration/feature_repos/universal/online_store/redis.py
new file mode 100644
index 0000000000..073760f514
--- /dev/null
+++ b/sdk/python/tests/integration/feature_repos/universal/online_store/redis.py
@@ -0,0 +1,26 @@
+from typing import Dict
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.waiting_utils import wait_for_logs
+
+from tests.integration.feature_repos.universal.online_store_creator import (
+    OnlineStoreCreator,
+)
+
+
+class RedisOnlineStoreCreator(OnlineStoreCreator):
+    def __init__(self, project_name: str):
+        super().__init__(project_name)
+        self.container = DockerContainer("redis").with_exposed_ports("6379")
+
+    def create_online_store(self) -> Dict[str, str]:
+        self.container.start()
+        log_string_to_wait_for = "Ready to accept connections"
+        wait_for_logs(
+            container=self.container, predicate=log_string_to_wait_for, timeout=5
+        )
+        exposed_port = self.container.get_exposed_port("6379")
+        return {"type": "redis", "connection_string": f"localhost:{exposed_port},db=0"}
+
+    def teardown(self):
+        self.container.stop()
diff --git a/sdk/python/tests/integration/feature_repos/universal/online_store_creator.py b/sdk/python/tests/integration/feature_repos/universal/online_store_creator.py
new file mode 100644
index 0000000000..0fa0dbed3e
--- /dev/null
+++ b/sdk/python/tests/integration/feature_repos/universal/online_store_creator.py
@@ -0,0 +1,14 @@
+from abc import ABC
+
+from feast.repo_config import FeastConfigBaseModel
+
+
+class OnlineStoreCreator(ABC):
+    def __init__(self, project_name: str):
+        self.project_name = project_name
+
+    def create_online_store(self) -> FeastConfigBaseModel:
+        ...
+
+    def teardown(self):
+        ...