perf: bulk-authorize push devices (Closes #725)
This commit is contained in:
1 parent
c5e6e17fc1
commit
02a472f841
7 files changed
+191
-33
No files matched your search
@@ -296,3 +296,45 @@ async def test_deadline_reminders_bound_fanout_and_isolate_failed_devices(tmp_pa
|
||||
|
||||
assert delivered == 3
|
||||
assert peak == 2
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_deadline_reminders_bulk_authorize_due_devices_and_prune_inactive(tmp_path):
|
||||
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
||||
for session_id in ("active-device", "revoked-device"):
|
||||
store.upsert(session_id, {
|
||||
"endpoint": f"https://push.example/{session_id}",
|
||||
"keys": {"p256dh": "public-key", "auth": "auth-secret"},
|
||||
})
|
||||
store.set_deadline_preferences(
|
||||
session_id, enabled=True, timezone="UTC", reminder_hour=9
|
||||
)
|
||||
authorization_calls = []
|
||||
sent = []
|
||||
|
||||
async def assigned():
|
||||
return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}
|
||||
|
||||
async def session_statuses(management_ids):
|
||||
authorization_calls.append(list(management_ids))
|
||||
return {
|
||||
management_id: "active" if management_id == "active-device" else "revoked"
|
||||
for management_id in management_ids
|
||||
}
|
||||
|
||||
async def send(subscription, _payload):
|
||||
sent.append(subscription["endpoint"])
|
||||
|
||||
delivered = await dispatch_deadline_reminders(
|
||||
store,
|
||||
PushConfiguration("public", "private", "mailto:ops@example.com"),
|
||||
assigned,
|
||||
send,
|
||||
now=datetime(2026, 8, 13, 10, 0, tzinfo=timezone.utc),
|
||||
session_statuses=session_statuses,
|
||||
)
|
||||
|
||||
assert delivered == 1
|
||||
assert authorization_calls == [["active-device", "revoked-device"]]
|
||||
assert sent == ["https://push.example/active-device"]
|
||||
assert store.is_subscribed("revoked-device") is False
|
||||
@@ -204,6 +204,24 @@ async def test_managed_session_active_applies_configured_idle_deadline(monkeypat
|
||||
assert calls == [("active-device", 321), ("idle-device", 321)]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_managed_session_statuses_apply_configured_idle_deadline_in_one_call(monkeypatch):
|
||||
calls = []
|
||||
|
||||
class Store:
|
||||
def managed_statuses(self, management_ids, *, idle_timeout_seconds):
|
||||
calls.append((list(management_ids), idle_timeout_seconds))
|
||||
return {management_id: "active" for management_id in set(management_ids)}
|
||||
|
||||
monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "321")
|
||||
monkeypatch.setattr(dashboard_auth, "_session_store", lambda: Store())
|
||||
|
||||
assert await dashboard_auth.managed_session_statuses(
|
||||
["device-b", "device-a", "device-b"]
|
||||
) == {"device-a": "active", "device-b": "active"}
|
||||
assert calls == [(["device-b", "device-a", "device-b"], 321)]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch):
|
||||
captured = {}
|
||||
@@ -225,7 +243,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
|
||||
await main._push_poll_loop()
|
||||
|
||||
assert captured["unread"] is main.gitea_proxy.unread_notification_snapshot
|
||||
assert captured["session_active"] is dashboard_auth.managed_session_active
|
||||
assert captured["session_statuses"] is dashboard_auth.managed_session_statuses
|
||||
assert captured["max_concurrency"] == 3
|
||||
assert captured["max_individual_notifications"] == 4
|
||||
|
||||
@@ -881,19 +899,26 @@ async def test_dispatch_removes_inactive_sessions_without_blocking_active_device
|
||||
async def unread():
|
||||
return {"items": [{"id": 21}]}
|
||||
|
||||
async def session_active(management_id):
|
||||
return management_id == "active-device"
|
||||
authorization_calls = []
|
||||
|
||||
async def session_statuses(management_ids):
|
||||
authorization_calls.append(list(management_ids))
|
||||
return {
|
||||
management_id: "active" if management_id == "active-device" else "revoked"
|
||||
for management_id in management_ids
|
||||
}
|
||||
|
||||
async def send(subscription, _payload):
|
||||
sent.append(subscription["endpoint"])
|
||||
|
||||
config = PushConfiguration("public", "private", "mailto:ops@example.com")
|
||||
assert await dispatch_unread_updates(
|
||||
store, config, unread, send, session_active=session_active
|
||||
store, config, unread, send, session_statuses=session_statuses
|
||||
) == 1
|
||||
assert store.is_subscribed("expired-device") is False
|
||||
assert store.is_subscribed("active-device") is True
|
||||
assert sent == ["https://push.example/active-device"]
|
||||
assert authorization_calls == [["active-device", "expired-device"]]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@@ -908,7 +933,7 @@ async def test_dispatch_fails_closed_and_retains_subscriptions_when_session_regi
|
||||
async def unread():
|
||||
return {"items": [{"id": 22}]}
|
||||
|
||||
async def registry_unavailable(_management_id):
|
||||
async def registry_unavailable(_management_ids):
|
||||
raise RuntimeError("session registry unavailable")
|
||||
|
||||
async def send(subscription, _payload):
|
||||
@@ -916,7 +941,7 @@ async def test_dispatch_fails_closed_and_retains_subscriptions_when_session_regi
|
||||
|
||||
config = PushConfiguration("public", "private", "mailto:ops@example.com")
|
||||
assert await dispatch_unread_updates(
|
||||
store, config, unread, send, session_active=registry_unavailable
|
||||
store, config, unread, send, session_statuses=registry_unavailable
|
||||
) == 0
|
||||
assert store.is_subscribed("active-device") is True
|
||||
assert sent == []
|
||||
|
||||
@@ -102,6 +102,45 @@ def test_managed_session_status_enforces_absolute_and_idle_expiry(tmp_path):
|
||||
assert store.managed_status("missing-device", idle_timeout_seconds=900) == "revoked"
|
||||
|
||||
|
||||
def test_managed_session_statuses_classify_a_batch_with_one_read(tmp_path, monkeypatch):
|
||||
now = [1_000.0]
|
||||
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
||||
store.activate("active-session", 3_000, management_id="active-device")
|
||||
store.activate("idle-session", 3_000, management_id="idle-device")
|
||||
store.activate("expired-session", 1_500, management_id="expired-device")
|
||||
now[0] = 2_000.0
|
||||
store.touch("active-session", 3_000, idle_timeout_seconds=1_500)
|
||||
statements = []
|
||||
connect = sqlite3.connect
|
||||
|
||||
def traced_connect(*args, **kwargs):
|
||||
connection = connect(*args, **kwargs)
|
||||
connection.set_trace_callback(statements.append)
|
||||
return connection
|
||||
|
||||
monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect)
|
||||
|
||||
assert store.managed_statuses(
|
||||
["active-device", "idle-device", "expired-device", "missing-device", "active-device"],
|
||||
idle_timeout_seconds=900,
|
||||
) == {
|
||||
"active-device": "active",
|
||||
"idle-device": "idle",
|
||||
"expired-device": "revoked",
|
||||
"missing-device": "revoked",
|
||||
}
|
||||
assert [statement for statement in statements if statement.lstrip().upper().startswith("SELECT")] == [
|
||||
next(statement for statement in statements if "FROM active_sessions" in statement)
|
||||
]
|
||||
|
||||
|
||||
def test_managed_session_statuses_skips_database_for_an_empty_batch(tmp_path, monkeypatch):
|
||||
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
||||
monkeypatch.setattr(store, "_connect", lambda *args, **kwargs: pytest.fail("database opened"))
|
||||
|
||||
assert store.managed_statuses([], idle_timeout_seconds=900) == {}
|
||||
|
||||
|
||||
def test_touch_extends_only_the_matching_live_session(tmp_path):
|
||||
now = [1_000.0]
|
||||
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
||||
|
||||
Reference in new issue
Block a user