Pin the model end_date boundary in tests

CHANGELOG.md

@@ -14,6 +14,7 @@ All notable changes to Lumen will be documented in this file.
 - Tests now verify that API-key hashes are keyed on `ENCRYPTION_KEY` (#164).
 - `LUMEN_WSGI_WORKERS` now defaults to `auto`, sized at 2× DB pool capacity (10–256); Helm `wsgiWorkers` defaults to `"auto"` (#238).
 - OAuth key requests that replace an existing key now refresh its secret in place, keeping its name and usage stats, instead of revoking it. (#182)
+- Tests now pin the model `end_date` boundary: a model is blocked and hidden from the exact end instant. (#311)
 
 ### Added
 

tests/unit/test_llm_access.py

@@ -1,10 +1,14 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 from types import SimpleNamespace
 
 import pytest
+from sqlalchemy import select
 
+from lumen.extensions import db
+from lumen.models.model_config import ModelConfig
 from lumen.services.llm import (
     _consent_satisfied,
+    _resolve_single_access,
     ack_notice_text,
     bulk_model_access_info,
     consent_state,
@@ -422,6 +426,32 @@ def _set_end_date(app, model_id, end_date):
     db.session.commit()
 
 
+#: Fixed naive-UTC instant the end_date boundary tests freeze the clock at.
+NOW = datetime(2030, 1, 1, 12, 0, 0)
+
+
+@pytest.fixture
+def frozen_llm_now(monkeypatch):
+    monkeypatch.setattr("lumen.services.llm.utcnow", lambda: NOW)
+
+
+@pytest.mark.parametrize("end_date, expected", [
+    (NOW, "blocked"),
+    (NOW + timedelta(microseconds=1), "allowed"),
+])
+def test_resolve_single_access_end_date_boundary(frozen_llm_now, end_date, expected):
+    """The model is blocked from the exact end instant, not one tick later."""
+    assert _resolve_single_access(False, None, model_end_date=end_date) == expected
+
+
+def test_model_blocked_at_exact_end_date(app, ids, frozen_llm_now):
+    entity_id, model_id = ids
+    with app.app_context():
+        _set_end_date(app, model_id, NOW)
+        assert get_model_access_status(entity_id, model_id) == "blocked"
+        assert get_model_access(entity_id, model_id) is False
+
+
 def test_future_end_date_still_allowed(app, ids):
     entity_id, model_id = ids
     with app.app_context():
@@ -452,6 +482,22 @@ def test_active_sql_filter_excludes_expired(app, ids):
         assert model_id in active_ids
 
 
+def test_active_end_date_boundary(app, ids, monkeypatch):
+    """ModelConfig.active is False from the exact end instant, in SQL and Python."""
+    monkeypatch.setattr("lumen.models.model_config.utcnow", lambda: NOW)
+    entity_id, model_id = ids
+    with app.app_context():
+        _set_end_date(app, model_id, NOW)
+        active_ids = db.session.execute(select(ModelConfig.id).where(ModelConfig.active)).scalars().all()
+        assert model_id not in active_ids
+        assert db.session.get(ModelConfig, model_id).active is False
+        # 1 s rather than 1 µs: not every database keeps sub-second precision.
+        _set_end_date(app, model_id, NOW + timedelta(seconds=1))
+        active_ids = db.session.execute(select(ModelConfig.id).where(ModelConfig.active)).scalars().all()
+        assert model_id in active_ids
+        assert db.session.get(ModelConfig, model_id).active is True
+
+
 def test_bulk_unknown_model_id_raises(app, ids):
     """An id that resolves to no model is a caller bug and must not fail open."""
     import pytest

tests/unit/test_model_cache.py

@@ -5,7 +5,7 @@
 import time
 from collections import Counter
 from contextlib import contextmanager
-from datetime import timedelta
+from datetime import datetime, timedelta
 from unittest.mock import patch
 
 import pytest
@@ -145,6 +145,18 @@ def test_end_date_passing_between_refreshes_hides_the_model(app):
             assert model_cache.resolve("ending") is None
 
 
+def test_end_date_boundary_hides_model_at_exact_instant(app):
+    """A snapshot model is hidden from the exact end instant, not one tick later."""
+    now = datetime(2030, 1, 1, 12, 0, 0)
+    with app.app_context():
+        _add_model("ending", end_date=now)
+        model_cache.refresh(app)
+        with patch("lumen.services.model_cache.utcnow", return_value=now):
+            assert model_cache.resolve("ending") is None
+        with patch("lumen.services.model_cache.utcnow", return_value=now - timedelta(microseconds=1)):
+            assert model_cache.resolve("ending") is not None
+
+
 # ---------------------------------------------------------------------------
 # Changes become visible after refresh
 # ---------------------------------------------------------------------------