From 74a8c22478e4da40116d26a141d0a09eb9463fd3 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sat, 21 Mar 2026 14:58:36 +0000
Subject: [PATCH] test: improve test coverage for app/api/sessions.py

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
Agent-Logs-Url: https://github.com/christianlouis/DocuElevate/sessions/4fb13ab5-c829-4e56-9999-bfca5bc53f4a
---
 tests/test_api_sessions.py | 439 +++++++++++++++++++++++++++++++++++++
 1 file changed, 439 insertions(+)
 create mode 100644 tests/test_api_sessions.py

diff --git a/tests/test_api_sessions.py b/tests/test_api_sessions.py
new file mode 100644
index 00000000..3e334c5e
--- /dev/null
+++ b/tests/test_api_sessions.py
@@ -0,0 +1,439 @@
+"""Tests for the session management API endpoints (app/api/sessions.py).
+
+Covers:
+* _get_owner_id dependency helper (authenticated and unauthenticated paths)
+* GET /api/sessions/  – list sessions
+* DELETE /api/sessions/{id}  – revoke a single session
+* POST /api/sessions/revoke-all  – log off everywhere
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import secrets
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import itsdangerous
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.database import Base, get_db
+from app.models import UserSession
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+_OWNER = "sessionuser@example.com"
+_OTHER_OWNER = "other@example.com"
+_SESSION_SECRET = "test_secret_key_for_testing_must_be_at_least_32_characters_long"
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_session_cookie(session_data: dict) -> str:
+    """Encode *session_data* as a signed Starlette session cookie value."""
+    signer = itsdangerous.TimestampSigner(_SESSION_SECRET)
+    data = base64.b64encode(json.dumps(session_data).encode("utf-8"))
+    return signer.sign(data).decode("utf-8")
+
+
+def _make_user_session(
+    db,
+    user_id: str = _OWNER,
+    session_token: str | None = None,
+    expires_delta: timedelta = timedelta(days=30),
+) -> UserSession:
+    """Create and persist a UserSession in *db*."""
+    now = datetime.now(timezone.utc)
+    token = session_token or secrets.token_urlsafe(32)
+    session = UserSession(
+        session_token=token,
+        user_id=user_id,
+        ip_address="127.0.0.1",
+        user_agent="TestBrowser/1.0",
+        device_info="TestBrowser on Linux",
+        created_at=now,
+        last_active_at=now,
+        expires_at=now + expires_delta,
+    )
+    db.add(session)
+    db.commit()
+    db.refresh(session)
+    return session
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def sess_engine():
+    """In-memory SQLite engine scoped to one test."""
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine)
+    yield engine
+    Base.metadata.drop_all(bind=engine)
+
+
+@pytest.fixture()
+def sess_db(sess_engine):
+    """Database session scoped to one test."""
+    Session = sessionmaker(bind=sess_engine)
+    session = Session()
+    yield session
+    session.close()
+
+
+def _make_client(sess_engine, owner_id: str = _OWNER) -> TestClient:
+    """Return a TestClient with *owner_id* injected as the authenticated user."""
+    from app.api.sessions import _get_owner_id
+    from app.main import app
+
+    Session = sessionmaker(bind=sess_engine)
+
+    def _override_get_db():
+        session = Session()
+        try:
+            yield session
+        finally:
+            session.close()
+
+    def _override_owner():
+        return owner_id
+
+    app.dependency_overrides[get_db] = _override_get_db
+    app.dependency_overrides[_get_owner_id] = _override_owner
+
+    return TestClient(app, base_url="http://localhost", raise_server_exceptions=False)
+
+
+def _make_unauthenticated_client(sess_engine) -> TestClient:
+    """Return a TestClient with only the DB overridden (no auth injection)."""
+    from app.main import app
+
+    Session = sessionmaker(bind=sess_engine)
+
+    def _override_get_db():
+        session = Session()
+        try:
+            yield session
+        finally:
+            session.close()
+
+    app.dependency_overrides[get_db] = _override_get_db
+
+    return TestClient(app, base_url="http://localhost", raise_server_exceptions=False)
+
+
+def _cleanup():
+    """Remove all dependency overrides from the app."""
+    from app.main import app
+
+    app.dependency_overrides.clear()
+
+
+# ---------------------------------------------------------------------------
+# Tests – _get_owner_id helper
+# ---------------------------------------------------------------------------
+
+
+class TestGetOwnerId:
+    """Tests for the _get_owner_id dependency helper in app/api/sessions.py."""
+
+    @pytest.mark.unit
+    def test_unauthenticated_raises_401(self):
+        """_get_owner_id should raise HTTP 401 when the user is not authenticated."""
+        from unittest.mock import MagicMock
+
+        from fastapi import HTTPException
+
+        from app.api.sessions import _get_owner_id
+
+        mock_request = MagicMock()
+        with patch("app.api.sessions.get_current_owner_id", return_value=None):
+            with pytest.raises(HTTPException) as exc_info:
+                _get_owner_id(mock_request)
+            assert exc_info.value.status_code == 401
+            assert exc_info.value.detail == "Not authenticated"
+
+    @pytest.mark.unit
+    def test_authenticated_returns_owner_id(self):
+        """_get_owner_id should return the owner_id when the user is authenticated."""
+        from unittest.mock import MagicMock
+
+        from app.api.sessions import _get_owner_id
+
+        mock_request = MagicMock()
+        with patch("app.api.sessions.get_current_owner_id", return_value=_OWNER):
+            result = _get_owner_id(mock_request)
+            assert result == _OWNER
+
+
+# ---------------------------------------------------------------------------
+# Tests – GET /api/sessions/
+# ---------------------------------------------------------------------------
+
+
+class TestListSessions:
+    """Tests for GET /api/sessions/."""
+
+    @pytest.mark.unit
+    def test_list_sessions_empty(self, sess_engine):
+        """Returns an empty session list when no sessions exist."""
+        client = _make_client(sess_engine)
+        try:
+            resp = client.get("/api/sessions/")
+            assert resp.status_code == 200
+            data = resp.json()
+            assert data["sessions"] == []
+            assert "session_lifetime_days" in data
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_list_sessions_returns_active_sessions(self, sess_engine, sess_db):
+        """Returns session details for all active sessions belonging to the user."""
+        _make_user_session(sess_db, user_id=_OWNER)
+        _make_user_session(sess_db, user_id=_OWNER)
+        # A session owned by a different user must not appear.
+        _make_user_session(sess_db, user_id=_OTHER_OWNER)
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.get("/api/sessions/")
+            assert resp.status_code == 200
+            sessions = resp.json()["sessions"]
+            assert len(sessions) == 2
+            for s in sessions:
+                assert "id" in s
+                assert "device_info" in s
+                assert "ip_address" in s
+                assert "created_at" in s
+                assert "last_active_at" in s
+                assert "expires_at" in s
+                assert "is_current" in s
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_list_sessions_marks_current_session(self, sess_engine, sess_db):
+        """The session whose token matches request.session['_session_token'] is
+        marked ``is_current=True``; all others are ``False``."""
+        current_token = secrets.token_urlsafe(32)
+        current_session = _make_user_session(sess_db, user_id=_OWNER, session_token=current_token)
+        other_session = _make_user_session(sess_db, user_id=_OWNER)
+
+        cookie = _make_session_cookie({"_session_token": current_token})
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.get("/api/sessions/", cookies={"session": cookie})
+            assert resp.status_code == 200
+            sessions = resp.json()["sessions"]
+            session_map = {s["id"]: s for s in sessions}
+            assert session_map[current_session.id]["is_current"] is True
+            assert session_map[other_session.id]["is_current"] is False
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_list_sessions_no_current_token(self, sess_engine, sess_db):
+        """When no _session_token is present, all sessions have is_current=False."""
+        _make_user_session(sess_db, user_id=_OWNER)
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.get("/api/sessions/")
+            assert resp.status_code == 200
+            for s in resp.json()["sessions"]:
+                assert s["is_current"] is False
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_list_sessions_returns_lifetime_days(self, sess_engine):
+        """Response always includes session_lifetime_days."""
+        client = _make_client(sess_engine)
+        try:
+            resp = client.get("/api/sessions/")
+            assert resp.status_code == 200
+            assert isinstance(resp.json()["session_lifetime_days"], int)
+            assert resp.json()["session_lifetime_days"] >= 1
+        finally:
+            _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# Tests – DELETE /api/sessions/{session_id}
+# ---------------------------------------------------------------------------
+
+
+class TestRevokeSingleSession:
+    """Tests for DELETE /api/sessions/{session_id}."""
+
+    @pytest.mark.unit
+    def test_revoke_session_success(self, sess_engine, sess_db):
+        """Revoking an owned session returns 204 No Content."""
+        session = _make_user_session(sess_db, user_id=_OWNER)
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.delete(f"/api/sessions/{session.id}")
+            assert resp.status_code == 204
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_session_not_found(self, sess_engine):
+        """Revoking a non-existent session returns 404."""
+        client = _make_client(sess_engine)
+        try:
+            resp = client.delete("/api/sessions/999999")
+            assert resp.status_code == 404
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_session_belonging_to_other_user_returns_404(self, sess_engine, sess_db):
+        """A user cannot revoke another user's session (returns 404)."""
+        other_session = _make_user_session(sess_db, user_id=_OTHER_OWNER)
+
+        client = _make_client(sess_engine, owner_id=_OWNER)
+        try:
+            resp = client.delete(f"/api/sessions/{other_session.id}")
+            assert resp.status_code == 404
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_session_audit_failure_does_not_break_response(self, sess_engine, sess_db):
+        """Even if the audit service raises an exception, the response is still 204."""
+        session = _make_user_session(sess_db, user_id=_OWNER)
+
+        client = _make_client(sess_engine)
+        try:
+            with patch("app.utils.audit_service.record_event", side_effect=Exception("audit down")):
+                resp = client.delete(f"/api/sessions/{session.id}")
+            assert resp.status_code == 204
+        finally:
+            _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# Tests – POST /api/sessions/revoke-all
+# ---------------------------------------------------------------------------
+
+
+class TestRevokeAllSessions:
+    """Tests for POST /api/sessions/revoke-all."""
+
+    @pytest.mark.unit
+    def test_revoke_all_no_sessions(self, sess_engine):
+        """Returns revoked_count=0 when there are no sessions to revoke."""
+        client = _make_client(sess_engine)
+        try:
+            resp = client.post("/api/sessions/revoke-all")
+            assert resp.status_code == 200
+            data = resp.json()
+            assert data["revoked_count"] == 0
+            assert "message" in data
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_all_revokes_all_sessions(self, sess_engine, sess_db):
+        """All active sessions for the user are revoked."""
+        _make_user_session(sess_db, user_id=_OWNER)
+        _make_user_session(sess_db, user_id=_OWNER)
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.post("/api/sessions/revoke-all")
+            assert resp.status_code == 200
+            data = resp.json()
+            assert data["revoked_count"] == 2
+            assert "2" in data["message"]
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_all_preserves_current_session(self, sess_engine, sess_db):
+        """The session matching the current _session_token is NOT revoked."""
+        current_token = secrets.token_urlsafe(32)
+        current_session = _make_user_session(sess_db, user_id=_OWNER, session_token=current_token)
+        _make_user_session(sess_db, user_id=_OWNER)
+        _make_user_session(sess_db, user_id=_OWNER)
+
+        cookie = _make_session_cookie({"_session_token": current_token})
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.post("/api/sessions/revoke-all", cookies={"session": cookie})
+            assert resp.status_code == 200
+            # Only the two non-current sessions should be revoked.
+            assert resp.json()["revoked_count"] == 2
+
+            # The current session must still be active in the DB.
+            sess_db.refresh(current_session)
+            assert current_session.is_revoked is False
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_all_current_session_token_not_in_db(self, sess_engine, sess_db):
+        """When the _session_token in the cookie doesn't match any DB row,
+        all sessions are revoked (no session is preserved)."""
+        _make_user_session(sess_db, user_id=_OWNER)
+
+        # Cookie references a token that does not exist in the DB.
+        cookie = _make_session_cookie({"_session_token": "ghost_token_xyz"})
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.post("/api/sessions/revoke-all", cookies={"session": cookie})
+            assert resp.status_code == 200
+            assert resp.json()["revoked_count"] == 1
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_all_audit_failure_does_not_break_response(self, sess_engine, sess_db):
+        """Even if the audit service raises, revoke-all still returns 200."""
+        _make_user_session(sess_db, user_id=_OWNER)
+
+        client = _make_client(sess_engine)
+        try:
+            with patch("app.utils.audit_service.record_event", side_effect=Exception("audit down")):
+                resp = client.post("/api/sessions/revoke-all")
+            assert resp.status_code == 200
+            assert resp.json()["revoked_count"] == 1
+        finally:
+            _cleanup()
+
+    @pytest.mark.unit
+    def test_revoke_all_message_format(self, sess_engine, sess_db):
+        """Response message includes the count and mentions API tokens."""
+        _make_user_session(sess_db, user_id=_OWNER)
+
+        client = _make_client(sess_engine)
+        try:
+            resp = client.post("/api/sessions/revoke-all")
+            assert resp.status_code == 200
+            msg = resp.json()["message"]
+            assert "1" in msg
+            assert "API" in msg or "token" in msg.lower()
+        finally:
+            _cleanup()