diff --git a/tests/test_datastore.py b/tests/test_datastore.py
index 26a41bb7c1..b45c7a82b4 100644
--- a/tests/test_datastore.py
+++ b/tests/test_datastore.py
@@ -1877,6 +1877,47 @@ def testCacheExpiryDatasetsFromDisabled(self) -> None:
             self.assertExpiration(cache_manager, 5, threshold + 1)
             self.assertIn(f"{mode}={threshold}", str(cache_manager))
 
+    def testExpirationModeOverride(self) -> None:
+        threshold = 2  # Keep 2 datasets.
+        mode = "datasets"
+        config_str = self._expiration_config(mode, threshold)
+
+        mode = "size"
+        threshold = 55
+        with unittest.mock.patch.dict(
+            os.environ,
+            {"DAF_BUTLER_CACHE_EXPIRATION_MODE": f"{mode}={threshold}"},
+        ):
+            cache_manager = self._make_cache_manager(config_str)
+            self.assertExpiration(cache_manager, 10, 6)
+            self.assertIn(f"{mode}={threshold}", str(cache_manager))
+
+        # Check we get a warning with unrecognized form.
+        with unittest.mock.patch.dict(
+            os.environ,
+            {"DAF_BUTLER_CACHE_EXPIRATION_MODE": "something"},
+        ):
+            with self.assertLogs(level="WARNING") as cm:
+                self._make_cache_manager(config_str)
+            self.assertIn("Unrecognized form (something)", cm.output[0])
+
+        with unittest.mock.patch.dict(
+            os.environ,
+            {"DAF_BUTLER_CACHE_EXPIRATION_MODE": "something=5"},
+        ):
+            with self.assertRaises(ValueError) as cm:
+                self._make_cache_manager(config_str)
+            self.assertIn("Unrecognized value", str(cm.exception))
+
+    def testMissingThreshold(self) -> None:
+        threshold = ""
+        mode = "datasets"
+        config_str = self._expiration_config(mode, threshold)
+
+        with self.assertRaises(ValueError) as cm:
+            self._make_cache_manager(config_str)
+        self.assertIn("Cache expiration threshold", str(cm.exception))
+
     def testCacheExpiryDatasetsComposite(self) -> None:
         threshold = 2  # Keep 2 datasets.
         mode = "datasets"
@@ -1908,13 +1949,26 @@ def testCacheExpirySize(self) -> None:
         self.assertIn(f"{mode}={threshold}", str(cache_manager))
 
     def testDisabledCache(self) -> None:
+        # Configure an active cache but disable via environment.
+        threshold = 2
+        mode = "datasets"
+        config_str = self._expiration_config(mode, threshold)
+
+        with unittest.mock.patch.dict(
+            os.environ,
+            {"DAF_BUTLER_CACHE_EXPIRATION_MODE": "disabled"},
+        ):
+            env_cache_manager = self._make_cache_manager(config_str)
+
+        # Configure to be disabled
         threshold = 0
         mode = "disabled"
         config_str = self._expiration_config(mode, threshold)
-        cache_manager = self._make_cache_manager(config_str)
+        cfg_cache_manager = self._make_cache_manager(config_str)
 
         for cache_manager in (
-            self._make_cache_manager(config_str),
+            cfg_cache_manager,
+            env_cache_manager,
             DatastoreCacheManager.create_disabled(universe=DimensionUniverse()),
         ):
             for uri, ref in zip(self.files, self.refs, strict=True):
@@ -1923,6 +1977,7 @@ def testDisabledCache(self) -> None:
                 self.assertFalse(cache_manager.known_to_cache(ref))
                 with cache_manager.find_in_cache(ref, ".txt") as found:
                     self.assertIsNone(found, msg=f"{cache_manager}")
+                self.assertIn("disabled", str(cache_manager))
 
     def assertExpiration(
         self, cache_manager: DatastoreCacheManager, n_datasets: int, n_retained: int