feat(#27): Gate B [07/10]: FastAPI health endpoint + app factory (H-1)

- App factory pattern (FUND-8): create_app() produces fresh app per test
- Logging config (structured, no secrets)
- Config validation: SLAB_AES_KEY required, WOO_BASE_URL HTTPS-only
- GET /api/health returns 200 {status: ok, version, woo_reachable: false}
- SQLite + alembic skeleton: full DATA-MODEL schema, initial migration
- API contract tests via httpx TestClient: 5 cases all green
  - test_returns_200, test_returns_ok_status, test_woo_reachable_false
  - test_response_shape, test_with_test_client_lifespan (FUND-9/10)
This commit is contained in:
grokbot_ty
2026-09-07 07:53:36 +00:00
committed by grokbot_ty
parent d5d8c49e04
commit e5514fe3bf
29 changed files with 915 additions and 0 deletions
+21
View File
@@ -0,0 +1,21 @@
FROM python:3.13-slim
WORKDIR /app
COPY backend/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY backend/ .
# Create data and images directories
RUN mkdir -p /data /images /app/prompts
# Bind mount for prompts (read-only)
# Volumes for data and images are handled by docker-compose
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
EXPOSE 8000
CMD ["uvicorn", "app.main:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000"]
+45
View File
@@ -0,0 +1,45 @@
# Alembic configuration for SlabUploader
[alembic]
script_location = alembic
sqlalchemy.url = sqlite:///data/slab.db
# Templates used to generate migration files
file_template = %%(year)s_%%(month)s_%%(day)s_%%(rev)s_%%(slug)s
# Timezone for file timestamps
timezone = UTC
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = INFO
formatter = generic
[formatter_generic]
format = %(asctime)s %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %Y-%m-%dT%H:%M:%S%z
View File
+65
View File
@@ -0,0 +1,65 @@
"""Alembic environment configuration.
Sets up sys.path so that app packages are importable when running
alembic from the backend directory.
Alembic uses sync connections for migrations even with an async app.
"""
import os
import sys
# Ensure the backend root is on the path so app.* is importable
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
# Import models so that Base.metadata includes all tables
from app.db.models import Base
config = context.config
fileConfig(config.config_file_name)
def run_migrations_online() -> None:
"""Run migrations in 'online' mode using sync engine."""
connectable = engine_from_config(
config.get_section(config.config_ini_section),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=Base.metadata,
render_as_batch=True,
)
with context.begin_transaction():
context.run_migrations()
connectable.dispose()
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode."""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=Base.metadata,
render_as_batch=True,
literal_binds=True,
dialect_opts={"paramstyle": "batch"},
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
@@ -0,0 +1,178 @@
"""initial schema.
Revision ID: initial
Revises:
Create Date: 2026-09-07
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "initial"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
# slabs
op.create_table(
"slabs",
sa.Column("id", sa.Text(), nullable=False, primary_key=True),
sa.Column("sku", sa.String(length=24), nullable=False),
sa.Column("status", sa.Text(), nullable=False, server_default="draft"),
sa.Column("length_in", sa.Float(), nullable=False),
sa.Column("thickness_in", sa.Float(), nullable=False),
sa.Column("sqft", sa.Float(), nullable=True),
sa.Column("bdft", sa.Float(), nullable=True),
sa.Column("width_min_in", sa.Float(), nullable=True),
sa.Column("width_max_in", sa.Float(), nullable=True),
sa.Column("width_avg_in", sa.Float(), nullable=True),
sa.Column("species_id", sa.Integer(), nullable=True),
sa.Column("wood_category_ids", sa.JSON(), nullable=True),
sa.Column("edge_type_term_id", sa.Integer(), nullable=True),
sa.Column("figure_term_ids", sa.JSON(), nullable=True),
sa.Column("grade_term_ids", sa.JSON(), nullable=True),
sa.Column("thickness_term_id", sa.Integer(), nullable=True),
sa.Column("moisture_term_id", sa.Integer(), nullable=True),
sa.Column("fig_tag_ids", sa.JSON(), nullable=True),
sa.Column("feat_tag_ids", sa.JSON(), nullable=True),
sa.Column("price_per_bdft", sa.Float(), nullable=True),
sa.Column("price", sa.Float(), nullable=False),
sa.Column("price_source", sa.Text(), nullable=True),
sa.Column("title", sa.Text(), nullable=True),
sa.Column("short_title", sa.Text(), nullable=True),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("content_source", sa.Text(), nullable=True),
sa.Column("client_rev", sa.Integer(), nullable=False, server_default="1"),
sa.Column("server_rev", sa.Integer(), nullable=False, server_default="1"),
sa.Column("created_at", sa.Text(), nullable=False, server_default=sa.text("datetime('now')")),
sa.Column("updated_at", sa.Text(), nullable=False, server_default=sa.text("datetime('now')")),
sa.Column("inference_thread_id", sa.Text(), nullable=True),
sa.CheckConstraint("length_in >= 0"),
sa.CheckConstraint("thickness_in >= 0"),
sa.CheckConstraint("price >= 0"),
sa.UniqueConstraint("sku"),
)
op.create_index("idx_slabs_status", "slabs", ["status"])
op.create_index("idx_slabs_updated_at", "slabs", ["updated_at"])
# slab_photos
op.create_table(
"slab_photos",
sa.Column("id", sa.Text(), nullable=False, primary_key=True),
sa.Column("slab_id", sa.Text(), nullable=False),
sa.Column("kind", sa.Text(), nullable=False),
sa.Column("role", sa.Text(), nullable=True),
sa.Column("seq", sa.Integer(), nullable=False, server_default="0"),
sa.Column("original_path", sa.Text(), nullable=True),
sa.Column("processed_path", sa.Text(), nullable=True),
sa.Column("created_at", sa.Text(), server_default=sa.text("datetime('now')")),
sa.ForeignKeyConstraint(["slab_id"], ["slabs.id"], ondelete="CASCADE"),
)
# pricing_rules
op.create_table(
"pricing_rules",
sa.Column("id", sa.Text(), nullable=False, primary_key=True),
sa.Column("species", sa.Text(), nullable=False),
sa.Column("price_per_bdft", sa.Float(), nullable=False),
sa.Column("created_at", sa.Text(), server_default=sa.text("datetime('now')")),
sa.Column("updated_at", sa.Text(), server_default=sa.text("datetime('now')")),
sa.UniqueConstraint("species"),
)
# woo_categories
op.create_table(
"woo_categories",
sa.Column("woo_id", sa.Integer(), nullable=False, primary_key=True),
sa.Column("name", sa.Text(), nullable=True),
sa.Column("slug", sa.Text(), nullable=True),
sa.Column("parent_id", sa.Integer(), nullable=True),
sa.Column("synced_at", sa.Text(), nullable=True),
sa.UniqueConstraint("woo_id"),
)
# woo_attributes
op.create_table(
"woo_attributes",
sa.Column("woo_id", sa.Integer(), nullable=False, primary_key=True),
sa.Column("name", sa.Text(), nullable=True),
sa.Column("slug", sa.Text(), nullable=True),
sa.Column("type", sa.Text(), server_default="select"),
sa.Column("synced_at", sa.Text(), nullable=True),
sa.UniqueConstraint("woo_id"),
)
# woo_attribute_terms
op.create_table(
"woo_attribute_terms",
sa.Column("woo_id", sa.Integer(), nullable=False, primary_key=True),
sa.Column("attribute_id", sa.Integer(), nullable=False),
sa.Column("name", sa.Text(), nullable=True),
sa.Column("slug", sa.Text(), nullable=True),
sa.Column("synced_at", sa.Text(), nullable=True),
sa.ForeignKeyConstraint(["attribute_id"], ["woo_attributes.woo_id"]),
sa.UniqueConstraint("woo_id"),
)
# woo_tags
op.create_table(
"woo_tags",
sa.Column("woo_id", sa.Integer(), nullable=False, primary_key=True),
sa.Column("name", sa.Text(), nullable=True),
sa.Column("slug", sa.Text(), nullable=True),
sa.Column("synced_at", sa.Text(), nullable=True),
sa.UniqueConstraint("woo_id"),
)
# settings
op.create_table(
"settings",
sa.Column("key", sa.Text(), nullable=False, primary_key=True),
sa.Column("value_enc", sa.LargeBinary(), nullable=True),
sa.Column("updated_at", sa.Text(), server_default=sa.text("datetime('now')")),
)
# sync_log
op.create_table(
"sync_log",
sa.Column("id", sa.Integer(), nullable=False, autoincrement=True, primary_key=True),
sa.Column("slab_id", sa.Text(), nullable=True),
sa.Column("sku", sa.Text(), nullable=True),
sa.Column("woo_product_id", sa.Integer(), nullable=True),
sa.Column("status", sa.Text(), nullable=True),
sa.Column("http_status", sa.Integer(), nullable=True),
sa.Column("error", sa.Text(), nullable=True),
sa.Column("detail", sa.Text(), nullable=True),
sa.Column("payload_hash", sa.Text(), nullable=True),
sa.Column("created_at", sa.Text(), server_default=sa.text("datetime('now')")),
)
# inference_log
op.create_table(
"inference_log",
sa.Column("id", sa.Integer(), nullable=False, autoincrement=True, primary_key=True),
sa.Column("slab_id", sa.Text(), nullable=True),
sa.Column("call", sa.Text(), nullable=True),
sa.Column("request_payload", sa.Text(), nullable=True),
sa.Column("response_payload", sa.Text(), nullable=True),
sa.Column("created_at", sa.Text(), server_default=sa.text("datetime('now')")),
)
def downgrade() -> None:
op.drop_table("inference_log")
op.drop_table("sync_log")
op.drop_table("settings")
op.drop_table("woo_tags")
op.drop_table("woo_attribute_terms")
op.drop_table("woo_attributes")
op.drop_table("woo_categories")
op.drop_table("pricing_rules")
op.drop_table("slab_photos")
op.drop_index("idx_slabs_updated_at", table_name="slabs")
op.drop_index("idx_slabs_status", table_name="slabs")
op.drop_table("slabs")
View File
View File
+23
View File
@@ -0,0 +1,23 @@
"""Health endpoint router."""
from fastapi import APIRouter
router = APIRouter()
@router.get("/api/health")
async def health_check() -> dict:
"""Deploy health and smoke path.
Returns:
Health status with woo_reachable flag. When Woo credentials
are not configured, woo_reachable is False.
Contract (OPENAPI):
GET /api/health → { "status": "ok", "version": "...", "woo_reachable": bool }
"""
return {
"status": "ok",
"version": "0.1.0",
"woo_reachable": False,
}
+46
View File
@@ -0,0 +1,46 @@
"""Application settings loaded from environment."""
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""FastAPI application settings.
WOO_BASE_URL is the deploy-time SoT for the WooCommerce store.
It must be HTTPS. FastAPI refuses startup when missing or HTTP.
SLAB_AES_KEY is the AES-GCM master key (32-byte base64). Required.
Missing key refuses startup.
All Woo credentials and inference secrets are stored AES-GCM encrypted
in the settings table at runtime, not passed as env vars.
"""
model_config = SettingsConfigDict(env_prefix="SLAB_", env_file=".env", env_file_encoding="utf-8")
aes_key: str | None = None
woo_base_url: str | None = None
app_hostname: str | None = None
# Image output constants (from TECH-SPEC)
aspect_ratio: str = "3:4"
output_px: int = 1600
output_px_min: int = 1600
fill_target: float = 0.80
def validate(self) -> None:
"""Refuse startup on invalid config."""
if self.aes_key is None:
raise ValueError("SLAB_AES_KEY environment variable is required")
if len(self.aes_key) < 32:
raise ValueError("SLAB_AES_KEY must be at least 32 bytes base64")
if self.woo_base_url is None:
raise ValueError("WOO_BASE_URL environment variable is required")
if not self.woo_base_url.startswith("https://"):
raise ValueError("WOO_BASE_URL must be HTTPS")
@property
def woo_credentials_configured(self) -> bool:
"""Whether Woo application password has been stored (checked via DB at runtime)."""
return False # Always false from config alone; checked via settings table at runtime
View File
+84
View File
@@ -0,0 +1,84 @@
"""SQLite engine and session management.
Uses SQLAlchemy async engine with aiosqlite driver.
For tests, pass `test_db_url` via the settings or use a test fixture
to point at an in-memory SQLite database.
"""
import logging
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.db.models import Base
logger = logging.getLogger(__name__)
# Default DB path for production
DEFAULT_DB_URL = "sqlite+aiosqlite:///data/slab.db"
_engine: object = None
_session_factory: object = None
_test_url: str | None = None
def set_test_url(url: str) -> None:
"""Override the database URL for testing. Must be called before create_app."""
global _test_url
_test_url = url
def get_db_url() -> str:
"""Return the database URL, preferring test URL if set."""
if _test_url:
return _test_url
return DEFAULT_DB_URL
async def init_db() -> None:
"""Initialize the async engine, session factory, and create tables.
For in-memory SQLite tests, creates tables directly.
For file-based SQLite, attempts alembic migrations first, falling
back to direct table creation.
"""
global _engine, _session_factory
db_url = get_db_url()
logger.info("Connecting to %s", db_url)
_engine = create_async_engine(
db_url,
echo=False,
connect_args={"check_same_thread": False},
)
_session_factory = async_sessionmaker(
_engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False,
)
# Create all tables (works for both in-memory and file-based)
async with _engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
logger.info("Database initialized")
async def close_db_connection() -> None:
"""Dispose of the engine (closes all connections)."""
global _engine
if _engine is not None:
await _engine.dispose()
_engine = None
async def get_session() -> AsyncSession:
"""Get a database session."""
if _session_factory is None:
raise RuntimeError("Database not initialized. Call init_db() first.")
return _session_factory()
+185
View File
@@ -0,0 +1,185 @@
"""SQLAlchemy ORM models for the SlabUploader database.
These models map to the schema defined in docs/DATA-MODEL.md.
"""
import enum
from sqlalchemy import (
Index,
JSON,
CheckConstraint,
Enum,
Float,
ForeignKey,
Integer,
LargeBinary,
String,
Text,
UniqueConstraint,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.sql import func
# SQLite doesn't have native JSON or enum types; we use TEXT/JSON via SQLAlchemy's
# dialect-specific types.
class Base(DeclarativeBase):
pass
class SlabStatus(str, enum.Enum):
DRAFT = "draft"
CALIBRATED = "calibrated"
READY = "ready"
PUBLISHING = "publishing"
PUBLISHED = "published"
FAILED = "failed"
class Slab(Base):
__tablename__ = "slabs"
id: Mapped[str] = mapped_column(Text, primary_key=True)
sku: Mapped[str] = mapped_column(String(24), unique=True, nullable=False)
status: Mapped[SlabStatus] = mapped_column(
Enum(SlabStatus, native_enum=False, values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=SlabStatus.DRAFT,
)
length_in: Mapped[float] = mapped_column(Float, nullable=False)
thickness_in: Mapped[float] = mapped_column(Float, nullable=False)
sqft: Mapped[float | None] = mapped_column(Float, nullable=True)
bdft: Mapped[float | None] = mapped_column(Float, nullable=True)
width_min_in: Mapped[float | None] = mapped_column(Float, nullable=True)
width_max_in: Mapped[float | None] = mapped_column(Float, nullable=True)
width_avg_in: Mapped[float | None] = mapped_column(Float, nullable=True)
species_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
wood_category_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
edge_type_term_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
figure_term_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
grade_term_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
thickness_term_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
moisture_term_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
fig_tag_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
feat_tag_ids: Mapped[list[int] | None] = mapped_column(JSON, nullable=True)
price_per_bdft: Mapped[float | None] = mapped_column(Float, nullable=True)
price: Mapped[float] = mapped_column(Float, nullable=False)
price_source: Mapped[str | None] = mapped_column(Text, nullable=True)
title: Mapped[str | None] = mapped_column(Text, nullable=True)
short_title: Mapped[str | None] = mapped_column(Text, nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
content_source: Mapped[str | None] = mapped_column(Text, nullable=True)
client_rev: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
server_rev: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
created_at: Mapped[str] = mapped_column(Text, nullable=False, server_default=func.now())
updated_at: Mapped[str] = mapped_column(Text, nullable=False, server_default=func.now())
inference_thread_id: Mapped[str | None] = mapped_column(Text, nullable=True)
__table_args__ = (
CheckConstraint("length_in >= 0"),
CheckConstraint("thickness_in >= 0"),
CheckConstraint("price >= 0"),
Index("idx_slabs_status", "status"),
Index("idx_slabs_updated_at", "updated_at"),
)
class SlabPhoto(Base):
__tablename__ = "slab_photos"
id: Mapped[str] = mapped_column(Text, primary_key=True)
slab_id: Mapped[str] = mapped_column(
Text, ForeignKey("slabs.id", ondelete="CASCADE"), nullable=False
)
kind: Mapped[str] = mapped_column(Text, nullable=False)
role: Mapped[str | None] = mapped_column(Text, nullable=True)
seq: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
original_path: Mapped[str | None] = mapped_column(Text, nullable=True)
processed_path: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
class PricingRule(Base):
__tablename__ = "pricing_rules"
id: Mapped[str] = mapped_column(Text, primary_key=True)
species: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
price_per_bdft: Mapped[float] = mapped_column(Float, nullable=False)
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
updated_at: Mapped[str] = mapped_column(Text, server_default=func.now())
class WooCategory(Base):
__tablename__ = "woo_categories"
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(Text)
slug: Mapped[str] = mapped_column(Text)
parent_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
synced_at: Mapped[str] = mapped_column(Text)
class WooAttribute(Base):
__tablename__ = "woo_attributes"
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(Text)
slug: Mapped[str] = mapped_column(Text)
type: Mapped[str] = mapped_column(Text, default="select")
synced_at: Mapped[str] = mapped_column(Text)
class WooAttributeTerm(Base):
__tablename__ = "woo_attribute_terms"
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
attribute_id: Mapped[int] = mapped_column(
Integer, ForeignKey("woo_attributes.woo_id"), nullable=False
)
name: Mapped[str] = mapped_column(Text)
slug: Mapped[str] = mapped_column(Text)
synced_at: Mapped[str] = mapped_column(Text)
class WooTag(Base):
__tablename__ = "woo_tags"
woo_id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(Text)
slug: Mapped[str] = mapped_column(Text)
synced_at: Mapped[str] = mapped_column(Text)
class Setting(Base):
__tablename__ = "settings"
key: Mapped[str] = mapped_column(Text, primary_key=True)
value_enc: Mapped[bytes | None] = mapped_column(LargeBinary, nullable=True)
updated_at: Mapped[str] = mapped_column(Text, server_default=func.now())
class SyncLog(Base):
__tablename__ = "sync_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
slab_id: Mapped[str | None] = mapped_column(Text, nullable=True)
sku: Mapped[str | None] = mapped_column(Text, nullable=True)
woo_product_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
status: Mapped[str | None] = mapped_column(Text, nullable=True)
http_status: Mapped[int | None] = mapped_column(Integer, nullable=True)
error: Mapped[str | None] = mapped_column(Text, nullable=True)
detail: Mapped[str | None] = mapped_column(Text, nullable=True)
payload_hash: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
class InferenceLog(Base):
__tablename__ = "inference_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
slab_id: Mapped[str | None] = mapped_column(Text, nullable=True)
call: Mapped[str | None] = mapped_column(Text, nullable=True)
request_payload: Mapped[str | None] = mapped_column(Text, nullable=True)
response_payload: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[str] = mapped_column(Text, server_default=func.now())
View File
+26
View File
@@ -0,0 +1,26 @@
"""Logging configuration."""
import logging
import sys
def configure_logging() -> None:
"""Configure root logger with structured formatting.
Format: ISO-8601 timestamp, level, logger name, message.
Secrets (AES keys, Woo passwords, inference keys) are never logged.
"""
root = logging.getLogger()
root.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.INFO)
formatter = logging.Formatter(
"%(asctime)s %(levelname)s %(name)s: %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S%z",
)
handler.setFormatter(formatter)
if not root.handlers:
root.addHandler(handler)
+49
View File
@@ -0,0 +1,49 @@
"""App factory and lifespan."""
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.api import health
from app.config import Settings
from app.db.engine import close_db_connection, init_db
from app.logging_config import configure_logging
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan: startup and shutdown hooks.
Handles database initialization, logging configuration,
and cleanup on shutdown.
"""
configure_logging()
logging.info("SlabUploader starting up")
await init_db()
yield
logging.info("SlabUploader shutting down")
await close_db_connection()
def create_app(settings: Settings | None = None) -> FastAPI:
"""Create a fresh FastAPI app instance.
Each call produces an independent app with its own database connection
and cleared dependency overrides. This is the factory used by tests.
Args:
settings: Application settings. Uses env vars if None.
Returns:
Configured FastAPI application.
"""
if settings is None:
settings = Settings()
settings.validate()
app = FastAPI(title="SlabUploader", version="0.1.0", lifespan=lifespan)
app.include_router(health.router)
return app
+9
View File
@@ -0,0 +1,9 @@
# Call 1 prompt — species detection
# Read fresh each invocation. Edit on host without restart.
Identify the slab species, figure, edge type, and grade from the provided images.
Return results constrained to the provided taxonomy only. Do not invent species,
figure terms, or grade terms that are not in the taxonomy list.
Confidence threshold: 0.7. Below threshold, return empty for that field.
+7
View File
@@ -0,0 +1,7 @@
# Call 2 prompt — content generation
# Read fresh each invocation. Edit on host without restart.
Generate a title, short title, and description for the slab using the provided
deterministic dimensions, species, and figure information.
Do not invent dimensions. Use the provided length, thickness, widths, sqft, and bdft.
View File
View File
+7
View File
@@ -0,0 +1,7 @@
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."]
[tool.ruff]
line-length = 100
target-version = "py313"
+3
View File
@@ -0,0 +1,3 @@
pytest>=8.3,<9
httpx>=0.28,<1
respx>=0.22,<1
+9
View File
@@ -0,0 +1,9 @@
fastapi>=0.115,<1
uvicorn[standard]>=0.34,<1
sqlalchemy>=2.0,<3
aiosqlite>=0.20,<1
alembic>=1.14,<2
httpx>=0.28,<1
pydantic>=2.9,<3
pydantic-settings>=2.7,<3
python-dotenv>=1.0,<2
View File
+45
View File
@@ -0,0 +1,45 @@
"""Shared test fixtures for all backend tests.
App factory pattern (FUND-8): each test gets a fresh app instance.
FUND-9: TestClient manages lifespan.
FUND-10: dependency_overrides cleared after every test.
"""
import pytest
from app.config import Settings
from app.db import engine
from app.main import create_app
@pytest.fixture(autouse=True)
def _test_db() -> None:
"""Set up a fresh in-memory SQLite database for each test."""
engine.set_test_url("sqlite+aiosqlite:///:memory:")
yield
engine._test_url = None
engine._engine = None
engine._session_factory = None
@pytest.fixture
def settings() -> Settings:
"""Create test settings with in-memory SQLite."""
return Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk", # 32 bytes base64
woo_base_url="https://test.example.com",
app_hostname="test.local",
)
@pytest.fixture
def app(settings: Settings) -> None:
"""Yield a fresh FastAPI app with test settings, then clean up.
Each test gets its own app instance (FUND-8).
TestClient manages the lifespan (FUND-9).
dependency_overrides are cleared after each test (FUND-10).
"""
app_instance = create_app(settings)
yield app_instance
app_instance.dependency_overrides.clear()
View File
+113
View File
@@ -0,0 +1,113 @@
"""H-1: Health endpoint contract test.
Tests via httpx TestClient. Each test gets a fresh app instance (FUND-8).
H-1: GET /api/health returns 200 {status: ok} with woo_reachable false
when credentials are not configured.
FUND-8: App factory creates fresh instance per test.
FUND-9: TestClient context manager manages lifespan.
FUND-10: dependency_overrides cleared after every test.
"""
from fastapi.testclient import TestClient
from app.main import create_app
class TestHealthEndpoint:
"""GET /api/health contract tests."""
def test_returns_200(self) -> None:
"""H-1: Health endpoint returns 200."""
from app.config import Settings
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
client = TestClient(app)
response = client.get("/api/health")
assert response.status_code == 200
client.close()
app.dependency_overrides.clear()
def test_returns_ok_status(self) -> None:
"""H-1: Health endpoint returns status ok."""
from app.config import Settings
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
client = TestClient(app)
response = client.get("/api/health")
data = response.json()
assert data["status"] == "ok"
client.close()
app.dependency_overrides.clear()
def test_woo_reachable_false_when_credentials_not_configured(self) -> None:
"""H-1: woo_reachable is false when Woo credentials not configured."""
from app.config import Settings
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
client = TestClient(app)
response = client.get("/api/health")
data = response.json()
assert data["woo_reachable"] is False
client.close()
app.dependency_overrides.clear()
def test_response_shape(self) -> None:
"""H-1: Response contains exactly status, version, and woo_reachable."""
from app.config import Settings
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
client = TestClient(app)
response = client.get("/api/health")
data = response.json()
assert "status" in data
assert "version" in data
assert "woo_reachable" in data
assert isinstance(data["status"], str)
assert isinstance(data["version"], str)
assert isinstance(data["woo_reachable"], bool)
client.close()
app.dependency_overrides.clear()
def test_with_test_client_lifespan(self) -> None:
"""FUND-9: TestClient with app manages lifespan correctly."""
from app.config import Settings
settings = Settings(
aes_key="dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk",
woo_base_url="https://test.example.com",
)
app = create_app(settings)
with TestClient(app) as client:
response = client.get("/api/health")
assert response.status_code == 200
assert response.json()["status"] == "ok"
app.dependency_overrides.clear()
View File