feat: add sliding session timeout with idle and absolute caps
This commit is contained in:
@@ -0,0 +1,213 @@
|
||||
"""Session timeout integration tests.
|
||||
|
||||
Validates the sliding-window session timeout policy: idle timeout,
|
||||
absolute timeout, and the interaction between activity and expiration.
|
||||
Every test exercises the *public API* — the only internal manipulation
|
||||
is back-dating timestamps in the database, since we cannot wait minutes
|
||||
inside a test run.
|
||||
"""
|
||||
import pytest
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from tests.integration.client.base import ApiError
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def assert_success(response: dict, message_contains: str = "") -> dict:
|
||||
"""Assert that an api_response-wrapped payload succeeded."""
|
||||
data = response.get("data", {})
|
||||
assert response.get("success") is not False, (
|
||||
f"Expected success but got error: {response.get('message')}"
|
||||
)
|
||||
if message_contains:
|
||||
assert message_contains.lower() in response.get("message", "").lower(), (
|
||||
f"Expected message to contain '{message_contains}' but got: {response.get('message')}"
|
||||
)
|
||||
return data
|
||||
|
||||
|
||||
def _get_session_row(integration_app, token: str):
|
||||
"""Look up the Session model row for a given bearer token."""
|
||||
from gatehouse_app.models.user.session import Session
|
||||
with integration_app.app_context():
|
||||
return Session.query.filter_by(token=token).first()
|
||||
|
||||
|
||||
def _touch_session(integration_app, session_id: str, **updates):
|
||||
"""Directly update columns on a Session row.
|
||||
|
||||
Only use this to simulate the passage of time — never to assert
|
||||
internal state.
|
||||
"""
|
||||
from gatehouse_app.models.user.session import Session
|
||||
with integration_app.app_context():
|
||||
sess = Session.query.get(session_id)
|
||||
for attr, value in updates.items():
|
||||
setattr(sess, attr, value)
|
||||
from gatehouse_app import db
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def logged_in_session(integration_client, create_test_user, integration_app):
|
||||
"""Register a user, log in via the API, and return the session metadata.
|
||||
|
||||
Returns dict with ``user``, ``token``, ``session_id``, ``session_row``.
|
||||
The ``session_row`` is a detached SQLAlchemy instance — re-query if
|
||||
you need fresh DB state.
|
||||
"""
|
||||
user = create_test_user(password="TestPass123!")
|
||||
integration_client.auth.login(
|
||||
email=user["email"], password="TestPass123!",
|
||||
)
|
||||
token = integration_client._token
|
||||
|
||||
session_row = _get_session_row(integration_app, token)
|
||||
assert session_row is not None, "Session row should exist after login"
|
||||
|
||||
return {
|
||||
"user": user,
|
||||
"token": token,
|
||||
"session_id": session_row.id,
|
||||
"session_row": session_row,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSessionTimeouts:
|
||||
"""Sliding-window timeout behavior exercised through the public API."""
|
||||
|
||||
def test_session_valid_before_timeout(
|
||||
self, integration_client, create_test_user,
|
||||
):
|
||||
"""SESS-01 — Fresh session is accepted.
|
||||
|
||||
A session that was just created should pass all auth checks.
|
||||
This is the baseline: if this fails, every other timeout test
|
||||
is meaningless.
|
||||
"""
|
||||
user = create_test_user(password="MyPass123!")
|
||||
integration_client.auth.login(email=user["email"], password="MyPass123!")
|
||||
|
||||
result = integration_client.auth.me()
|
||||
data = assert_success(result)
|
||||
assert data["user"]["email"] == user["email"]
|
||||
|
||||
def test_idle_timeout_rejects_token(
|
||||
self, integration_client, logged_in_session, integration_app,
|
||||
):
|
||||
"""SESS-02 — Session rejected after idle period elapses.
|
||||
|
||||
Push ``last_activity_at`` far enough into the past that the
|
||||
idle window has closed. The API must return 401.
|
||||
"""
|
||||
_touch_session(
|
||||
integration_app,
|
||||
logged_in_session["session_id"],
|
||||
last_activity_at=datetime.now(timezone.utc) - timedelta(hours=1),
|
||||
)
|
||||
|
||||
with pytest.raises(ApiError) as exc_info:
|
||||
integration_client.auth.me()
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
def test_absolute_timeout_rejects_even_active_user(
|
||||
self, integration_client, logged_in_session, integration_app,
|
||||
):
|
||||
"""SESS-03 — Absolute cap overrides recent activity.
|
||||
|
||||
Push ``created_at`` into the past so the absolute window has
|
||||
elapsed, but keep ``last_activity_at`` fresh. The session
|
||||
must still be rejected — activity cannot extend past the
|
||||
absolute limit.
|
||||
"""
|
||||
_touch_session(
|
||||
integration_app,
|
||||
logged_in_session["session_id"],
|
||||
created_at=datetime.now(timezone.utc) - timedelta(days=1),
|
||||
last_activity_at=datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
with pytest.raises(ApiError) as exc_info:
|
||||
integration_client.auth.me()
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
def test_api_request_keeps_session_alive(
|
||||
self, integration_client, logged_in_session, integration_app,
|
||||
):
|
||||
"""SESS-04 — Hitting an API endpoint extends the session.
|
||||
|
||||
Back-date ``last_activity_at`` to *just* inside the idle
|
||||
window. A subsequent API call should succeed and the session
|
||||
should remain usable — the sliding window should have reset.
|
||||
"""
|
||||
from gatehouse_app.models.user.session import Session
|
||||
from gatehouse_app import db
|
||||
|
||||
# Back-date to 10 seconds ago — still inside the idle window.
|
||||
_touch_session(
|
||||
integration_app,
|
||||
logged_in_session["session_id"],
|
||||
last_activity_at=datetime.now(timezone.utc) - timedelta(seconds=10),
|
||||
)
|
||||
|
||||
# This request should succeed AND extend the session.
|
||||
result = integration_client.auth.me()
|
||||
assert_success(result)
|
||||
|
||||
# After the request, last_activity_at should be much closer to now.
|
||||
with integration_app.app_context():
|
||||
refreshed = Session.query.get(logged_in_session["session_id"])
|
||||
now = datetime.now(timezone.utc)
|
||||
# Allow for clock skew / commit latency — should be within 30s.
|
||||
diff = abs((now - refreshed.last_activity_at.replace(tzinfo=timezone.utc)).total_seconds())
|
||||
assert diff < 30, (
|
||||
f"last_activity_at should be near-now after API call, "
|
||||
f"but delta is {diff:.1f}s"
|
||||
)
|
||||
|
||||
def test_revoked_session_rejected(
|
||||
self, integration_client, logged_in_session,
|
||||
):
|
||||
"""SESS-05 — Revoked session is rejected regardless of timing.
|
||||
|
||||
Revoke via the API, then verify the token is dead. This
|
||||
mirrors AUTH-12 but is included here so the timeout test
|
||||
suite is self-contained.
|
||||
"""
|
||||
integration_client.auth.revoke_session(logged_in_session["session_id"])
|
||||
|
||||
with pytest.raises(ApiError) as exc_info:
|
||||
integration_client.auth.me()
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
def test_refresh_endpoint_extends_session(
|
||||
self, integration_client, logged_in_session, integration_app,
|
||||
):
|
||||
"""SESS-06 — POST /auth/sessions/refresh extends the session.
|
||||
|
||||
The refresh endpoint exists so the frontend can proactively
|
||||
keep a session alive during idle UI periods. Verify it
|
||||
succeeds and returns a new ``expires_at``.
|
||||
"""
|
||||
result = integration_client.auth.refresh_session()
|
||||
data = assert_success(result, "session refreshed")
|
||||
|
||||
assert "expires_at" in data, "Response should include new expires_at"
|
||||
Reference in New Issue
Block a user