Piotr Oleszczyk
ac829171d9
- Add backend/innercontext/mcp_server.py with tools covering products, inventory, routines, skin snapshots, medications, lab results, and grooming schedule - Mount MCP app at /mcp in main.py using combine_lifespans - Fix test isolation: patch app.router.lifespan_context in conftest to avoid StreamableHTTPSessionManager single-run limitation Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
import os
|
|
from contextlib import asynccontextmanager
|
|
|
|
# Must be set before importing db (which calls create_engine at module level)
|
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
from sqlmodel.pool import StaticPool
|
|
|
|
import db as db_module
|
|
from db import get_session
|
|
from main import app
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _db_only_lifespan(a):
|
|
"""Lifespan without the MCP server for test isolation.
|
|
|
|
StreamableHTTPSessionManager.run() can only be called once per instance,
|
|
which conflicts with the per-test TestClient lifecycle. We replace the
|
|
combined (db + MCP) lifespan with one that only does DB setup.
|
|
"""
|
|
db_module.create_db_and_tables()
|
|
yield
|
|
|
|
|
|
@pytest.fixture()
|
|
def session(monkeypatch):
|
|
"""Per-test fresh SQLite in-memory database with full isolation."""
|
|
engine = create_engine(
|
|
"sqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
# Patch before TestClient triggers lifespan (which calls create_db_and_tables)
|
|
monkeypatch.setattr(db_module, "engine", engine)
|
|
import innercontext.models # noqa: F401 — populate SQLModel.metadata
|
|
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as s:
|
|
yield s
|
|
# monkeypatch auto-restores db_module.engine after test
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(session, monkeypatch):
|
|
"""TestClient using the per-test session for every request."""
|
|
# Replace combined (db+MCP) lifespan with DB-only to avoid the
|
|
# StreamableHTTPSessionManager single-run limitation.
|
|
monkeypatch.setattr(app.router, "lifespan_context", _db_only_lifespan)
|
|
|
|
def _override():
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = _override
|
|
with TestClient(app) as c:
|
|
yield c
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
# ---- Shared data fixtures ----
|
|
|
|
|
|
@pytest.fixture()
|
|
def product_data():
|
|
return {
|
|
"name": "CeraVe Moisturising Cream",
|
|
"brand": "CeraVe",
|
|
"category": "moisturizer",
|
|
"recommended_time": "both",
|
|
"leave_on": True,
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def created_product(client, product_data):
|
|
r = client.post("/products", json=product_data)
|
|
assert r.status_code == 201
|
|
return r.json()
|
|
|
|
|
|
@pytest.fixture()
|
|
def medication_data():
|
|
return {"kind": "prescription", "product_name": "Epiduo"}
|
|
|
|
|
|
@pytest.fixture()
|
|
def created_medication(client, medication_data):
|
|
r = client.post("/health/medications", json=medication_data)
|
|
assert r.status_code == 201
|
|
return r.json()
|
|
|
|
|
|
@pytest.fixture()
|
|
def created_routine(client):
|
|
r = client.post(
|
|
"/routines", json={"routine_date": "2026-02-26", "part_of_day": "am"}
|
|
)
|
|
assert r.status_code == 201
|
|
return r.json()
|