73 lines
2.5 KiB
Python
73 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,
|
|
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)
|