Fix: Ignore expired environments in the selector

sqlmesh/core/selector.py

@@ -68,6 +68,9 @@ def select_models(
             A dictionary of models.
         """
         target_env = self._state_reader.get_environment(Environment.sanitize_name(target_env_name))
+        if target_env and target_env.expired:
+            target_env = None
+
         if not target_env and fallback_env_name:
             target_env = self._state_reader.get_environment(
                 Environment.sanitize_name(fallback_env_name)

tests/core/test_selector.py

@@ -14,6 +14,7 @@
 from sqlmesh.core.selector import Selector
 from sqlmesh.core.snapshot import SnapshotChangeCategory
 from sqlmesh.utils import UniqueKeyDict
+from sqlmesh.utils.date import now_timestamp
 
 
 @pytest.mark.parametrize(
@@ -185,6 +186,81 @@ def test_select_models(mocker: MockerFixture, make_snapshot, default_catalog: t.
     )
 
 
+def test_select_models_expired_environment(mocker: MockerFixture, make_snapshot):
+    modified_model_v1 = SqlModel(
+        name="db.modified_model",
+        query=d.parse_one("SELECT a + 1 FROM db.added_model"),
+    )
+    modified_model_v2 = SqlModel(
+        name="db.modified_model",
+        query=d.parse_one("SELECT a + 2 FROM db.added_model"),
+    )
+    removed_model = SqlModel(
+        name="db.removed_model",
+        query=d.parse_one("SELECT a FROM db.added_model"),
+    )
+    standalone_audit = StandaloneAudit(
+        name="test_audit", query=d.parse_one("SELECT * FROM added_model WHERE a IS NULL")
+    )
+
+    modified_model_v1_snapshot = make_snapshot(modified_model_v1)
+    modified_model_v1_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    removed_model_snapshot = make_snapshot(removed_model)
+    removed_model_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    standalone_audit_snapshot = make_snapshot(standalone_audit)
+    standalone_audit_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    prod_env = Environment(
+        name="prod",
+        snapshots=[modified_model_v1_snapshot.table_info],
+        start_at="2023-01-01",
+        end_at="2023-02-01",
+        plan_id="test_plan_id",
+    )
+
+    env_name = "test_env"
+    dev_env = Environment(
+        name=env_name,
+        snapshots=[modified_model_v1_snapshot.table_info, removed_model_snapshot.table_info],
+        start_at="2023-01-01",
+        end_at="2023-02-01",
+        plan_id="test_plan_id",
+    )
+
+    state_reader_mock = mocker.Mock()
+    state_reader_mock.get_environment.side_effect = (
+        lambda name: prod_env if name == "prod" else dev_env
+    )
+
+    all_snapshots = {
+        modified_model_v1_snapshot.snapshot_id: modified_model_v1_snapshot,
+        removed_model_snapshot.snapshot_id: removed_model_snapshot,
+    }
+    state_reader_mock.get_snapshots.side_effect = lambda snapshots: {
+        s.snapshot_id: all_snapshots[s.snapshot_id] for s in snapshots
+    }
+
+    local_models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
+    local_models[modified_model_v2.fqn] = modified_model_v2
+    selector = Selector(state_reader_mock, local_models)
+
+    _assert_models_equal(
+        selector.select_models(["*.modified_model"], env_name, fallback_env_name="prod"),
+        {
+            removed_model.fqn: removed_model,
+            modified_model_v2.fqn: modified_model_v2,
+        },
+    )
+
+    dev_env.expiration_ts = now_timestamp() - 1
+    _assert_models_equal(
+        selector.select_models(["*.modified_model"], env_name, fallback_env_name="prod"),
+        {
+            modified_model_v2.fqn: modified_model_v2,
+        },
+    )
+
+
 def test_select_change_schema(mocker: MockerFixture, make_snapshot):
     parent = SqlModel(
         name="db.parent",