diff --git a/src/models/database/__init__.py b/src/models/database/__init__.py index a01a86e5c..9a4d012fb 100644 --- a/src/models/database/__init__.py +++ b/src/models/database/__init__.py @@ -1 +1,5 @@ """Database models package.""" + +from models.database.prompts import Prompt, PromptVersion + +__all__ = ["Prompt", "PromptVersion"] diff --git a/src/models/database/prompts.py b/src/models/database/prompts.py new file mode 100644 index 000000000..cf701acda --- /dev/null +++ b/src/models/database/prompts.py @@ -0,0 +1,58 @@ +"""Persistence models for the versioned prompt catalog.""" + +from datetime import datetime + +from sqlalchemy import JSON, DateTime, ForeignKey, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column, validates + +from models.database.base import Base +from prompts.exceptions import PromptValidationError + + +class Prompt(Base): # pylint: disable=too-few-public-methods + """Prompt identity with a pointer to the default version.""" + + __tablename__ = "prompt" + + id: Mapped[str] = mapped_column(primary_key=True) + + name: Mapped[str] = mapped_column(String(255)) + + default_version: Mapped[int] = mapped_column() + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +class PromptVersion(Base): # pylint: disable=too-few-public-methods + """Immutable version row containing the template text and declared variables.""" + + __tablename__ = "prompt_version" + + # Foreign key is not turned on by default for sqlite, but a PromptVersion will always + # be created after a Prompt is created in the database operation logic. Therefore, + # a PromptVersion with a nonexistent prompt.id will never be created. + prompt_id: Mapped[str] = mapped_column(ForeignKey("prompt.id"), primary_key=True) + + version: Mapped[int] = mapped_column(primary_key=True) + + template: Mapped[str] = mapped_column(Text()) + + variables: Mapped[list[str]] = mapped_column(JSON, default=list) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + @validates("variables") + def validate_unique_variables(self, _key: str, value: list[str]) -> list[str]: + """Ensure no duplicate variable names.""" + if len(value) != len(set(value)): + raise PromptValidationError("The variables contain duplicate names") + + return value diff --git a/src/prompts/__init__.py b/src/prompts/__init__.py new file mode 100644 index 000000000..14fcd39dd --- /dev/null +++ b/src/prompts/__init__.py @@ -0,0 +1 @@ +"""Prompts module for the versioned prompt catalog.""" diff --git a/src/prompts/exceptions.py b/src/prompts/exceptions.py new file mode 100644 index 000000000..8b23ce113 --- /dev/null +++ b/src/prompts/exceptions.py @@ -0,0 +1,9 @@ +"""Domain exceptions for the prompts module.""" + + +class PromptError(Exception): + """Base class for prompt domain errors.""" + + +class PromptValidationError(PromptError): + """Invalid prompt field values.""" diff --git a/tests/unit/models/database/test_prompts.py b/tests/unit/models/database/test_prompts.py new file mode 100644 index 000000000..6e998d20a --- /dev/null +++ b/tests/unit/models/database/test_prompts.py @@ -0,0 +1,311 @@ +"""Unit tests for Prompt and PromptVersion database models.""" + +from collections.abc import Generator + +import pytest +from sqlalchemy import String, create_engine, inspect, text +from sqlalchemy.engine import Engine +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from models.database.base import Base +from models.database.prompts import Prompt, PromptVersion +from prompts.exceptions import PromptValidationError + + +@pytest.fixture(name="sqlite_engine") +def sqlite_engine_fixture() -> Generator[Engine, None, None]: + """Provide an in-memory SQLite engine with all tables created. + + Yields: + Engine: A SQLAlchemy Engine bound to an in-memory SQLite database with + tables created from Base.metadata. + + """ + 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="db_session") +def db_session_fixture(sqlite_engine: Engine) -> Generator[Session, None, None]: + """Provide a database session for tests. + + Parameters: + sqlite_engine: In-memory SQLite engine fixture. + + Yields: + Session: A SQLAlchemy Session bound to the test engine. + + """ + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + session = session_factory() + try: + yield session + finally: + session.close() + + +@pytest.fixture(name="sample_prompt") +def sample_prompt_fixture(db_session: Session) -> Prompt: + """Insert and return a sample Prompt row. + + Parameters: + db_session: Active database session. + + Returns: + Prompt: The persisted prompt instance. + + """ + prompt = Prompt(id="pmpt_001", name="greeting", default_version=1) + db_session.add(prompt) + db_session.commit() + return prompt + + +class TestPromptModelDefinition: + """Test cases for Prompt model schema definition.""" + + def test_tablename(self) -> None: + """Test Prompt uses the expected table name.""" + assert Prompt.__tablename__ == "prompt" + + def test_expected_columns_exist(self) -> None: + """Test Prompt defines all expected columns.""" + column_names = {col.name for col in Prompt.__table__.columns} + expected = {"id", "name", "default_version", "created_at", "updated_at"} + assert column_names == expected + + def test_name_column_max_length(self) -> None: + """Test name column has max length 255.""" + name_column = Prompt.__table__.c.name + assert isinstance(name_column.type, String) + assert name_column.type.length == 255 + + def test_id_is_primary_key(self) -> None: + """Test id column is the primary key.""" + assert Prompt.__table__.c.id.primary_key + + +class TestPromptVersionModelDefinition: + """Test cases for PromptVersion model schema definition.""" + + def test_tablename(self) -> None: + """Test PromptVersion uses the expected table name.""" + assert PromptVersion.__tablename__ == "prompt_version" + + def test_expected_columns_exist(self) -> None: + """Test PromptVersion defines all expected columns.""" + column_names = {col.name for col in PromptVersion.__table__.columns} + expected = {"prompt_id", "version", "template", "variables", "created_at"} + assert column_names == expected + + def test_composite_primary_key(self) -> None: + """Test primary key is (prompt_id, version).""" + pk_columns = {col.name for col in PromptVersion.__table__.primary_key.columns} + assert pk_columns == {"prompt_id", "version"} + + def test_prompt_id_has_foreign_key(self) -> None: + """Test prompt_id references prompt.id.""" + fk_targets = { + fk.target_fullname + for fk in PromptVersion.__table__.c.prompt_id.foreign_keys + } + assert "prompt.id" in fk_targets + + +class TestPromptDatabaseOperations: + """Test cases for Prompt database operations.""" + + def test_tables_created_by_metadata(self, sqlite_engine: Engine) -> None: + """Test prompt and prompt_version tables are created via Base.metadata.""" + inspector = inspect(sqlite_engine) + table_names = inspector.get_table_names() + assert "prompt" in table_names + assert "prompt_version" in table_names + + def test_insert_and_read_prompt(self, db_session: Session) -> None: + """Test inserting and reading back a Prompt row.""" + prompt = Prompt(id="pmpt_001", name="greeting", default_version=1) + db_session.add(prompt) + db_session.commit() + + result = db_session.get(Prompt, "pmpt_001") + assert result is not None + assert result.id == "pmpt_001" + assert result.name == "greeting" + assert result.default_version == 1 + assert result.created_at is not None + assert result.updated_at is not None + + def test_duplicate_prompt_id_raises_integrity_error( + self, db_session: Session + ) -> None: + """Test inserting duplicate prompt id raises IntegrityError.""" + db_session.add(Prompt(id="pmpt_001", name="first", default_version=1)) + db_session.commit() + + db_session.add(Prompt(id="pmpt_001", name="second", default_version=1)) + with pytest.raises(IntegrityError): + db_session.commit() + + +class TestPromptVersionDatabaseOperations: + """Test cases for PromptVersion database operations.""" + + def test_insert_and_read_version( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test inserting and reading back a PromptVersion row.""" + version = PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="Hello {{name}}", + variables=["name"], + ) + db_session.add(version) + db_session.commit() + + result = db_session.get(PromptVersion, (sample_prompt.id, 1)) + assert result is not None + assert result.prompt_id == sample_prompt.id + assert result.version == 1 + assert result.template == "Hello {{name}}" + assert result.variables == ["name"] + assert result.created_at is not None + + def test_multiple_versions_for_same_prompt( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test that a prompt can have multiple versions.""" + v1 = PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="Hello {{name}}", + variables=["name"], + ) + v2 = PromptVersion( + prompt_id=sample_prompt.id, + version=2, + template="Hi {{name}}, welcome to {{place}}", + variables=["name", "place"], + ) + db_session.add_all([v1, v2]) + db_session.commit() + + results = ( + db_session.query(PromptVersion) + .filter_by(prompt_id=sample_prompt.id) + .order_by(PromptVersion.version) + .all() + ) + assert len(results) == 2 + assert results[0].version == 1 + assert results[1].version == 2 + + def test_duplicate_prompt_id_and_version_raises_integrity_error( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test inserting duplicate (prompt_id, version) raises IntegrityError.""" + db_session.add( + PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="first", + variables=[], + ) + ) + db_session.commit() + + db_session.add( + PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="second", + variables=[], + ) + ) + with pytest.raises(IntegrityError): + db_session.commit() + + def test_foreign_key_enforced(self, db_session: Session) -> None: + """Test inserting a version for a non-existent prompt raises IntegrityError.""" + db_session.execute(text("PRAGMA foreign_keys = ON")) + version = PromptVersion( + prompt_id="nonexistent", + version=1, + template="Hello", + variables=[], + ) + db_session.add(version) + with pytest.raises(IntegrityError): + db_session.commit() + + def test_empty_variables_defaults_to_list( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test that variables defaults to an empty list.""" + version = PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="No placeholders here", + ) + db_session.add(version) + db_session.commit() + + result = db_session.get(PromptVersion, (sample_prompt.id, 1)) + assert result is not None + assert result.variables == [] + + +class TestPromptVersionVariableValidation: + """Test cases for PromptVersion variable uniqueness validation.""" + + def test_unique_variables_accepted(self) -> None: + """Test that a list of unique variable names is accepted.""" + version = PromptVersion( + prompt_id="pmpt_001", + version=1, + template="{{a}} {{b}}", + variables=["a", "b"], + ) + assert version.variables == ["a", "b"] + + def test_duplicate_variables_raises_validation_error(self) -> None: + """Test that duplicate variable names raise PromptValidationError.""" + with pytest.raises(PromptValidationError, match="duplicate names"): + PromptVersion( + prompt_id="pmpt_001", + version=1, + template="{{a}} {{a}}", + variables=["a", "a"], + ) + + def test_empty_variables_accepted(self) -> None: + """Test that an empty variable list is accepted.""" + version = PromptVersion( + prompt_id="pmpt_001", + version=1, + template="No variables", + variables=[], + ) + assert version.variables == [] + + def test_single_variable_accepted(self) -> None: + """Test that a single variable is accepted.""" + version = PromptVersion( + prompt_id="pmpt_001", + version=1, + template="{{name}}", + variables=["name"], + ) + assert version.variables == ["name"]