diff --git a/src/utils/saved_prompts.py b/src/utils/saved_prompts.py index 7058b4c67..d7f12d3b8 100644 --- a/src/utils/saved_prompts.py +++ b/src/utils/saved_prompts.py @@ -1,7 +1,21 @@ -"""Validation helpers for saved prompts.""" +"""Validation helpers and data access for saved prompts.""" +from sqlalchemy.exc import IntegrityError -class SavedPromptValidationError(Exception): +import constants +from app.database import get_session +from log import get_logger +from models.database.saved_prompts import SavedPrompt +from utils.suid import get_suid + +logger = get_logger(__name__) + + +class SavedPromptError(Exception): + """Base class for saved-prompt domain errors.""" + + +class SavedPromptValidationError(SavedPromptError): """Invalid saved-prompt field values.""" @@ -9,6 +23,18 @@ class SavedPromptLimitExceededError(SavedPromptValidationError): """Per-user saved-prompt count would exceed the configured maximum.""" +class SavedPromptNotFoundError(SavedPromptError): + """No saved prompt exists for the given identifier.""" + + +class SavedPromptAccessDeniedError(SavedPromptError): + """The saved prompt exists but is not owned by the requesting user.""" + + +class SavedPromptConflictError(SavedPromptError): + """A saved prompt conflicts with an existing unique constraint.""" + + def validate_saved_prompt_quota( current_count: int, max_prompts_per_user: int, @@ -90,3 +116,122 @@ def validate_saved_prompt_content( f"Saved prompt content length {len(content)} exceeds maximum " f"{max_content_length}" ) + + +def create_saved_prompt( + user_id: str, + name: str, + content: str, + max_prompts_per_user: int, +) -> SavedPrompt: + """Create a saved prompt for a user after enforcing the per-user quota. + + Caller is responsible for validating ``name`` and ``content``. This function + counts existing prompts for ``user_id``, enforces ``max_prompts_per_user``, + inserts a new row with a generated id, and returns the persisted entity with + timestamps loaded. + + Parameters: + user_id: Owner of the saved prompt. + name: Display name as provided by the caller (not stripped here). + content: Prompt body as provided by the caller. + max_prompts_per_user: Maximum prompts the user may hold. + + Returns: + The created ``SavedPrompt`` with id and timestamps populated. + + Raises: + SavedPromptLimitExceededError: If the user is already at the limit. + SavedPromptConflictError: If insert violates a unique constraint + (practically always duplicate ``(user_id, name)``). + """ + with get_session() as session: + current_count = session.query(SavedPrompt).filter_by(user_id=user_id).count() + validate_saved_prompt_quota(current_count, max_prompts_per_user) + + saved_prompt = SavedPrompt( + id=get_suid(), + user_id=user_id, + name=name, + content=content, + ) + session.add(saved_prompt) + try: + session.commit() + except IntegrityError as exc: + logger.debug( + "Saved prompt create conflict for user_id=%s", + user_id, + ) + raise SavedPromptConflictError("Saved prompt name already exists") from exc + + # reload server default timestamps so they remain usable after the session closes + session.refresh(saved_prompt) + logger.debug( + "Created saved prompt id=%s for user_id=%s", + saved_prompt.id, + user_id, + ) + return saved_prompt + + +def list_saved_prompts_by_user(user_id: str) -> list[SavedPrompt]: + """List saved prompts for a user ordered by created_at descending. + + Results are capped at ``SAVED_PROMPTS_MAX_PER_USER_UPPER_BOUND`` so a + misconfigured or out-of-band insert path cannot materialize unbounded rows. + + Parameters: + user_id: Owner whose prompts should be returned. + + Returns: + List of ``SavedPrompt`` rows for the user. Empty list if none exist. + Tie order when ``created_at`` values are equal is database-defined. + """ + with get_session() as session: + return ( + session.query(SavedPrompt) + .filter_by(user_id=user_id) + .order_by(SavedPrompt.created_at.desc()) + .limit(constants.SAVED_PROMPTS_MAX_PER_USER_UPPER_BOUND) + .all() + ) + + +def delete_saved_prompt_by_id_and_user(prompt_id: str, user_id: str) -> None: + """Delete a saved prompt only if it belongs to the given user. + + Parameters: + prompt_id: Primary key of the saved prompt. + user_id: Authenticated user attempting the delete. + + Raises: + SavedPromptNotFoundError: If no row exists for ``prompt_id``. + SavedPromptAccessDeniedError: If the row exists but ``user_id`` does not + match the owner. + """ + with get_session() as session: + saved_prompt = session.query(SavedPrompt).filter_by(id=prompt_id).first() + if saved_prompt is None: + logger.debug( + "Saved prompt not found for delete prompt_id=%s user_id=%s", + prompt_id, + user_id, + ) + raise SavedPromptNotFoundError("Saved prompt not found") + + if saved_prompt.user_id != user_id: + logger.debug( + "Saved prompt access denied for delete prompt_id=%s user_id=%s", + prompt_id, + user_id, + ) + raise SavedPromptAccessDeniedError("Saved prompt access denied") + + session.delete(saved_prompt) + session.commit() + logger.debug( + "Deleted saved prompt id=%s for user_id=%s", + prompt_id, + user_id, + ) diff --git a/tests/unit/utils/test_saved_prompts.py b/tests/unit/utils/test_saved_prompts.py index 644cafce7..400fd03e4 100644 --- a/tests/unit/utils/test_saved_prompts.py +++ b/tests/unit/utils/test_saved_prompts.py @@ -1,14 +1,74 @@ -"""Unit tests for saved prompt validation helpers.""" +"""Unit tests for saved prompt validation helpers and data access.""" + +from collections.abc import Generator +from datetime import timedelta import pytest +from pytest_mock import MockerFixture +from sqlalchemy import create_engine +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool +from models.database.base import Base +from models.database.saved_prompts import SavedPrompt from utils.saved_prompts import ( + SavedPromptAccessDeniedError, + SavedPromptConflictError, SavedPromptLimitExceededError, + SavedPromptNotFoundError, SavedPromptValidationError, + create_saved_prompt, + delete_saved_prompt_by_id_and_user, + list_saved_prompts_by_user, validate_saved_prompt_content, validate_saved_prompt_name, validate_saved_prompt_quota, ) +from utils.suid import get_suid + + +@pytest.fixture(name="sqlite_engine") +def sqlite_engine_fixture() -> Generator[Engine, None, None]: + """Provide a function-scoped in-memory SQLite engine with tables created. + + Yields: + Engine: SQLAlchemy engine bound to an in-memory SQLite database. + """ + engine = create_engine( + "sqlite:///:memory:", + echo=False, + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(engine) + yield engine + engine.dispose() + + +@pytest.fixture(name="patch_saved_prompts_get_session") +def patch_saved_prompts_get_session_fixture( + mocker: MockerFixture, sqlite_engine: Engine +) -> None: + """Patch utils.saved_prompts.get_session to use the in-memory engine. + + Parameters: + mocker: pytest-mock fixture. + sqlite_engine: Function-scoped in-memory engine. + """ + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + + def _get_session() -> Session: + """Create a Session bound to the in-memory test engine. + + Returns: + Session: A new SQLAlchemy session for patched DAL calls. + """ + return session_factory() + + mocker.patch("utils.saved_prompts.get_session", side_effect=_get_session) class TestValidateSavedPromptQuota: @@ -48,10 +108,6 @@ def test_rejects_when_max_is_zero(self) -> None: ): validate_saved_prompt_quota(0, 0) - def test_limit_exceeded_is_validation_error_subclass(self) -> None: - """Test SavedPromptLimitExceededError subclasses SavedPromptValidationError.""" - assert issubclass(SavedPromptLimitExceededError, SavedPromptValidationError) - class TestValidateSavedPromptName: """Test cases for validate_saved_prompt_name.""" @@ -181,3 +237,172 @@ def test_content_rejected_when_max_length_is_zero(self) -> None: match="Saved prompt content length 1 exceeds maximum 0", ): validate_saved_prompt_content("a", max_content_length=0) + + +@pytest.mark.usefixtures("patch_saved_prompts_get_session") +class TestCreateSavedPrompt: + """Test cases for create_saved_prompt.""" + + def test_create_persists_and_returns_entity(self, sqlite_engine: Engine) -> None: + """Test create returns a persisted SavedPrompt with id and fields.""" + created = create_saved_prompt( + user_id="user-1", + name="My Prompt", + content="Hello", + max_prompts_per_user=50, + ) + + assert created.id + assert created.user_id == "user-1" + assert created.name == "My Prompt" + assert created.content == "Hello" + + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + with session_factory() as session: + stored = session.get(SavedPrompt, created.id) + assert stored is not None + assert stored.name == "My Prompt" + assert stored.content == "Hello" + + def test_create_return_value_has_usable_timestamps_after_session_close( + self, + ) -> None: + """Test timestamps are readable on the returned object after DAL returns.""" + created = create_saved_prompt( + user_id="user-1", + name="Timed", + content="Body", + max_prompts_per_user=50, + ) + + assert created.created_at is not None + assert created.updated_at is not None + + def test_create_at_limit_raises(self) -> None: + """Test create raises when the user already has max_prompts_per_user prompts.""" + create_saved_prompt("user-1", "one", "c1", max_prompts_per_user=1) + + with pytest.raises(SavedPromptLimitExceededError): + create_saved_prompt("user-1", "two", "c2", max_prompts_per_user=1) + + def test_create_duplicate_name_raises_conflict(self) -> None: + """Test duplicate (user_id, name) raises SavedPromptConflictError.""" + create_saved_prompt("user-1", "same", "first", max_prompts_per_user=50) + + with pytest.raises(SavedPromptConflictError) as exc_info: + create_saved_prompt("user-1", "same", "second", max_prompts_per_user=50) + + assert str(exc_info.value) == "Saved prompt name already exists" + + def test_create_allows_same_name_for_different_users(self) -> None: + """Test the same name may exist for different users.""" + first = create_saved_prompt("user-a", "shared", "a", max_prompts_per_user=50) + second = create_saved_prompt("user-b", "shared", "b", max_prompts_per_user=50) + assert first.id != second.id + + +@pytest.mark.usefixtures("patch_saved_prompts_get_session") +class TestListSavedPromptsByUser: + """Test cases for list_saved_prompts_by_user.""" + + def test_list_empty_returns_empty_list(self) -> None: + """Test listing for a user with no prompts returns [].""" + assert list_saved_prompts_by_user("nobody") == [] + + def test_list_returns_only_that_users_prompts_ordered_by_created_at_desc( + self, sqlite_engine: Engine + ) -> None: + """Test list is user-scoped and ordered by created_at descending.""" + older = create_saved_prompt("user-1", "first", "c1", max_prompts_per_user=50) + newer = create_saved_prompt("user-1", "second", "c2", max_prompts_per_user=50) + create_saved_prompt("user-2", "other", "c3", max_prompts_per_user=50) + + # Concurrent inserts can share the same created_at; force a clear order. + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + with session_factory() as session: + older_row = session.get(SavedPrompt, older.id) + newer_row = session.get(SavedPrompt, newer.id) + assert older_row is not None + assert newer_row is not None + older_row.created_at = newer_row.created_at - timedelta(seconds=1) + session.commit() + + results = list_saved_prompts_by_user("user-1") + + assert [p.id for p in results] == [newer.id, older.id] + assert all(p.user_id == "user-1" for p in results) + + def test_list_caps_at_configured_upper_bound( + self, mocker: MockerFixture, sqlite_engine: Engine + ) -> None: + """Test list applies SAVED_PROMPTS_MAX_PER_USER_UPPER_BOUND as a hard cap.""" + mocker.patch( + "utils.saved_prompts.constants.SAVED_PROMPTS_MAX_PER_USER_UPPER_BOUND", + 2, + ) + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + with session_factory() as session: + for index in range(3): + session.add( + SavedPrompt( + id=get_suid(), + user_id="user-1", + name=f"prompt-{index}", + content=f"content-{index}", + ) + ) + session.commit() + + results = list_saved_prompts_by_user("user-1") + assert len(results) == 2 + + +@pytest.mark.usefixtures("patch_saved_prompts_get_session") +class TestDeleteSavedPromptByIdAndUser: + """Test cases for delete_saved_prompt_by_id_and_user.""" + + def test_delete_owned_prompt(self, sqlite_engine: Engine) -> None: + """Test deleting an owned prompt removes the row.""" + created = create_saved_prompt( + "user-1", "to-delete", "body", max_prompts_per_user=50 + ) + + delete_saved_prompt_by_id_and_user(created.id, "user-1") + + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + with session_factory() as session: + assert session.get(SavedPrompt, created.id) is None + + def test_delete_missing_raises_not_found(self) -> None: + """Test deleting an unknown id raises SavedPromptNotFoundError.""" + with pytest.raises(SavedPromptNotFoundError) as exc_info: + delete_saved_prompt_by_id_and_user("missing-id", "user-1") + + assert str(exc_info.value) == "Saved prompt not found" + + def test_delete_other_users_prompt_raises_access_denied( + self, sqlite_engine: Engine + ) -> None: + """Test delete by non-owner raises access denied and leaves the row.""" + created = create_saved_prompt( + "owner", "private", "body", max_prompts_per_user=50 + ) + + with pytest.raises(SavedPromptAccessDeniedError) as exc_info: + delete_saved_prompt_by_id_and_user(created.id, "intruder") + + assert str(exc_info.value) == "Saved prompt access denied" + + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + with session_factory() as session: + assert session.get(SavedPrompt, created.id) is not None