Giorgio Gilestro
a113a7f3ce
Five fixes uncovered by actually running the suite in docker-compose.test.yml:
1. (real prod bug) PATCH /api/settings/digest mutated principal.user which
require_token had loaded in a now-closed session — the commit on the
handler's session persisted nothing. Re-fetch the user via the active
session before writing.
2. Portable PK type. SQLite only auto-fills `INTEGER PRIMARY KEY`; plain
BIGINT requires explicit values. Define a `_PK` alias of
`BigInteger().with_variant(Integer(), "sqlite")` and use it for all 10
autoincrement primary keys in app/models.py. No prod-schema change
(MariaDB still gets BIGINT).
3. job_lifecycle's MariaDB GET_LOCK / RELEASE_LOCK is now gated behind
`dialect.name == "mysql"`, so the test SQLite engine doesn't trip on
the missing function. Single-process test runs can't race themselves.
4. tests/test_news_window.py seeded Headline rows without `fingerprint`,
which is NOT NULL — added an `fp-{title}` value per row.
5. tests/test_email_digest_job.py now also patches `llm_configured` to
True so the job doesn't short-circuit on the missing API key.
6. (test container hygiene) Drop `COPY tests ./tests` from the test stage
in the Dockerfile — .dockerignore excludes `tests/` (correct: prod
image must not bake tests), and docker-compose.test.yml bind-mounts
./tests at run time anyway.
Suite now: 198 passed, 5 skipped, 1 pre-existing failure
(test_default_groups_present — Phase G dropped the "pie" group from
config/default.toml but the assertion wasn't updated; unrelated to this
branch).
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""Free vs paid window clamp on /api/news.
|
|
|
|
Integration-style: spins up a real router over an in-memory aiosqlite DB.
|
|
Skips on hosts that lack aiosqlite + httpx — same pattern as
|
|
test_portfolio_sync_api.py."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
|
|
def _build_app(tmp_path):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app import db as db_mod
|
|
from app.auth import sign_session
|
|
from app.db import Base
|
|
from app.models import Headline, User
|
|
from app.routers import api as api_router
|
|
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/news.db")
|
|
factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
db_mod._engine = engine
|
|
db_mod._session_factory = factory
|
|
|
|
now = datetime.now(timezone.utc)
|
|
|
|
async def _seed():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
async with factory() as s:
|
|
s.add(User(id=1, email="free@x", tier="free"))
|
|
s.add(User(id=2, email="paid@x", tier="paid"))
|
|
for hours_old, title in ((1, "fresh"), (12, "mid"), (20, "old")):
|
|
s.add(Headline(
|
|
source="test", title=title, url=f"https://e/{title}",
|
|
category="general",
|
|
published_at=now - timedelta(hours=hours_old),
|
|
fetched_at=now,
|
|
fingerprint=f"fp-{title}",
|
|
tags=[],
|
|
))
|
|
await s.commit()
|
|
|
|
asyncio.run(_seed())
|
|
|
|
app = FastAPI()
|
|
app.include_router(api_router.router, prefix="/api")
|
|
client = TestClient(app)
|
|
return client, sign_session(1), sign_session(2)
|
|
|
|
|
|
def test_free_user_clamped_to_6h(tmp_path):
|
|
client, free_sess, _ = _build_app(tmp_path)
|
|
r = client.get("/api/news?since_hours=24",
|
|
cookies={"cassandra_session": free_sess})
|
|
assert r.status_code == 200, r.text
|
|
titles = [h["title"] for h in r.json()]
|
|
assert "fresh" in titles
|
|
assert "mid" not in titles # 12h ago, beyond 6h
|
|
assert "old" not in titles
|
|
|
|
|
|
def test_paid_user_full_24h(tmp_path):
|
|
client, _, paid_sess = _build_app(tmp_path)
|
|
r = client.get("/api/news?since_hours=24",
|
|
cookies={"cassandra_session": paid_sess})
|
|
assert r.status_code == 200, r.text
|
|
titles = [h["title"] for h in r.json()]
|
|
assert {"fresh", "mid", "old"} <= set(titles)
|