learningcircuit--local-deep-research/tests/chat/test_chat_session_lifecycle.py

# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client
"""
Tests for chat session lifecycle management.

Tests session creation, archiving, deletion, and status transitions.
"""

import json
import pytest


class TestSessionCreation:
    """Tests for session creation scenarios."""

    def test_create_session_with_empty_initial_query(
        self, authenticated_client
    ):
        """Test creating a session with empty initial query."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": ""},
            content_type="application/json",
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["success"] is True
        assert "session_id" in data

    def test_create_session_with_whitespace_only_query(
        self, authenticated_client
    ):
        """Test creating a session with whitespace-only query."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "   \t\n  "},
            content_type="application/json",
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["success"] is True

    def test_create_session_with_empty_json_object_succeeds(
        self, authenticated_client
    ):
        """An empty JSON object {} creates a session with default values
        (no initial_query / title). This is what the frontend sends for a
        blank 'New Chat', so it must succeed."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            json={},
            content_type="application/json",
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["success"] is True
        assert "session_id" in data

    def test_create_session_with_no_body_is_rejected(
        self, authenticated_client
    ):
        """A request with no JSON body is rejected with 400. All chat write
        endpoints are guarded by @require_json_body, which requires a JSON
        object. The frontend always
        sends a body, so this only rejects malformed/empty requests."""
        response = authenticated_client.post(
            "/api/chat/sessions",
            content_type="application/json",
        )
        assert response.status_code == 400
        data = json.loads(response.data)
        assert data["success"] is False

    def test_create_multiple_sessions(self, authenticated_client):
        """Test creating multiple sessions for the same user."""
        session_ids = []
        for i in range(5):
            response = authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Query {i}"},
                content_type="application/json",
            )
            assert response.status_code == 200
            data = json.loads(response.data)
            session_ids.append(data["session_id"])

        # All session IDs should be unique
        assert len(session_ids) == len(set(session_ids))

        # All sessions should be listable
        list_response = authenticated_client.get("/api/chat/sessions")
        data = json.loads(list_response.data)
        listed_ids = [s["id"] for s in data["sessions"]]
        for sid in session_ids:
            assert sid in listed_ids


class TestSessionArchiving:
    """Tests for session archiving functionality."""

    def test_archive_session(self, authenticated_client):
        """Test archiving a session changes its status."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Archive it
        archive_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "archived"},
            content_type="application/json",
        )
        assert archive_resp.status_code == 200

        # Verify status changed
        get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        data = json.loads(get_resp.data)
        assert data["session"]["status"] == "archived"

    def test_archived_session_not_in_active_list(self, authenticated_client):
        """Test that archived sessions don't appear in active list."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Archive it
        authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "archived"},
            content_type="application/json",
        )

        # Should not be in active list
        list_resp = authenticated_client.get("/api/chat/sessions?status=active")
        data = json.loads(list_resp.data)
        active_ids = [s["id"] for s in data["sessions"]]
        assert session_id not in active_ids

    def test_archived_session_in_archived_list(self, authenticated_client):
        """Test that archived sessions appear in archived list."""
        # Create and archive session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "archived"},
            content_type="application/json",
        )

        # Should be in archived list
        list_resp = authenticated_client.get(
            "/api/chat/sessions?status=archived"
        )
        data = json.loads(list_resp.data)
        archived_ids = [s["id"] for s in data["sessions"]]
        assert session_id in archived_ids

    def test_reactivate_archived_session(self, authenticated_client):
        """Test that an archived session can be reactivated."""
        # Create and archive session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "archived"},
            content_type="application/json",
        )

        # Reactivate the session
        reactivate_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "active"},
            content_type="application/json",
        )
        assert reactivate_resp.status_code == 200

        # Verify status changed back to active
        get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        data = json.loads(get_resp.data)
        assert data["session"]["status"] == "active"


class TestSessionDeletion:
    """Tests for session deletion functionality."""

    def test_delete_via_status_rejected(self, authenticated_client):
        """Test that PATCH with status=deleted is rejected (use DELETE endpoint)."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Attempt delete via status — should be rejected
        delete_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "deleted"},
            content_type="application/json",
        )
        assert delete_resp.status_code == 400

    def test_delete_session_via_delete_endpoint(self, authenticated_client):
        """Test soft-deleting a session via DELETE endpoint."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Delete via endpoint
        delete_resp = authenticated_client.delete(
            f"/api/chat/sessions/{session_id}"
        )
        assert delete_resp.status_code == 200
        data = json.loads(delete_resp.data)
        assert data["success"] is True

    def test_deleted_session_not_in_active_list(self, authenticated_client):
        """Test that deleted sessions don't appear in active list."""
        # Create and delete session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        authenticated_client.delete(f"/api/chat/sessions/{session_id}")

        # Should not be in active list
        list_resp = authenticated_client.get("/api/chat/sessions?status=active")
        data = json.loads(list_resp.data)
        active_ids = [s["id"] for s in data["sessions"]]
        assert session_id not in active_ids

    def test_deleted_session_not_in_all_list(self, authenticated_client):
        """Test that hard-deleted sessions do not appear in 'all' list."""
        # Create and delete session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        authenticated_client.delete(f"/api/chat/sessions/{session_id}")

        # Should NOT be in 'all' list (hard-deleted)
        list_resp = authenticated_client.get("/api/chat/sessions?status=all")
        data = json.loads(list_resp.data)
        all_ids = [s["id"] for s in data["sessions"]]
        assert session_id not in all_ids

    def test_delete_nonexistent_session(self, authenticated_client):
        """Test deleting a session that doesn't exist returns 404."""
        response = authenticated_client.delete(
            "/api/chat/sessions/nonexistent-session-id-12345"
        )
        assert response.status_code == 404
        data = json.loads(response.data)
        assert data["success"] is False


class TestSessionTitleUpdates:
    """Tests for session title update functionality."""

    def test_update_title(self, authenticated_client):
        """Test updating session title."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Original query"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Update title
        new_title = "Updated Research Title"
        update_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": new_title},
            content_type="application/json",
        )
        assert update_resp.status_code == 200

        # Verify title changed
        get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        data = json.loads(get_resp.data)
        assert data["session"]["title"] == new_title

    def test_update_title_empty_string(self, authenticated_client):
        """Test updating session title to empty string."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Update to empty title - should be rejected
        update_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": ""},
            content_type="application/json",
        )
        assert update_resp.status_code == 400

    def test_update_title_too_long(self, authenticated_client):
        """Test updating session title with too long value."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Try to update with 501+ character title
        long_title = "A" * 501
        update_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": long_title},
            content_type="application/json",
        )
        assert update_resp.status_code == 400
        data = json.loads(update_resp.data)
        assert "too long" in data["error"].lower()

    def test_update_title_and_status_together(self, authenticated_client):
        """Test updating both title and status in one request."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Update both
        update_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": "New Title", "status": "archived"},
            content_type="application/json",
        )
        assert update_resp.status_code == 200

        # Verify both changed
        get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}")
        data = json.loads(get_resp.data)
        assert data["session"]["title"] == "New Title"
        assert data["session"]["status"] == "archived"


class TestSessionStatusValidation:
    """Tests for session status validation."""

    def test_invalid_status_rejected(self, authenticated_client):
        """Test that invalid status values are rejected."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Try invalid status
        update_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": "invalid_status"},
            content_type="application/json",
        )
        assert update_resp.status_code == 400
        data = json.loads(update_resp.data)
        assert "invalid" in data["error"].lower()

    @pytest.mark.parametrize("status", ["active", "archived"])
    def test_valid_status_values(self, status, authenticated_client):
        """Test that valid status values are accepted."""
        # Create session
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Update to valid status
        update_resp = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"status": status},
            content_type="application/json",
        )
        assert update_resp.status_code == 200


class TestListSessionsPagination:
    """Tests for session listing pagination."""

    def test_list_with_limit(self, authenticated_client):
        """Test listing sessions with limit."""
        # Create 5 sessions
        for i in range(5):
            authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Query {i}"},
                content_type="application/json",
            )

        # List with limit
        response = authenticated_client.get("/api/chat/sessions?limit=3")
        data = json.loads(response.data)
        assert len(data["sessions"]) <= 3

    def test_list_with_offset(self, authenticated_client):
        """Test listing sessions with offset."""
        # Create 5 sessions
        created_ids = []
        for i in range(5):
            resp = authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Query {i}"},
                content_type="application/json",
            )
            created_ids.append(json.loads(resp.data)["session_id"])

        # Get all sessions
        all_resp = authenticated_client.get("/api/chat/sessions?limit=100")
        all_sessions = json.loads(all_resp.data)["sessions"]

        # Get with offset
        offset_resp = authenticated_client.get("/api/chat/sessions?offset=2")
        offset_sessions = json.loads(offset_resp.data)["sessions"]

        # Offset should skip first 2
        assert len(offset_sessions) == len(all_sessions) - 2

    def test_list_with_limit_and_offset(self, authenticated_client):
        """Test listing sessions with both limit and offset."""
        # Create 10 sessions
        for i in range(10):
            authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Query {i}"},
                content_type="application/json",
            )

        # Get page 2 (offset 3, limit 3)
        response = authenticated_client.get(
            "/api/chat/sessions?limit=3&offset=3"
        )
        data = json.loads(response.data)
        assert len(data["sessions"]) <= 3