API Testing

uv add --dev pytest pytest-asyncio httpx

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
testpaths = ["tests"]

docker exec -it notes-db psql -U app -c "CREATE DATABASE notes_test;"

# tests/conftest.py
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.main import create_app
from app.core.db import get_db
from app.models import Base
from app.core.security import hash_password, create_token
from app.models.user import User

TEST_DB_URL = "postgresql+asyncpg://app:secret@localhost:5432/notes_test"


@pytest.fixture(scope="session")
async def engine():
    eng = create_async_engine(TEST_DB_URL)
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)   # test এর শুরুতে নতুন table
    yield eng
    await eng.dispose()


@pytest.fixture
async def db(engine):
    # প্রতিটা test একটা transaction এর ভেতরে, শেষে rollback → database পরিষ্কার
    async with engine.connect() as conn:
        trans = await conn.begin()
        session = AsyncSession(bind=conn, join_transaction_mode="create_savepoint", expire_on_commit=False)
        yield session
        await session.close()
        await trans.rollback()


@pytest.fixture
async def client(db):
    app = create_app()
    app.dependency_overrides[get_db] = lambda: db       # app কে test session দাও
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


async def _make_user(db, email: str) -> User:
    user = User(email=email, name=email.split("@")[0], password_hash=hash_password("secret123"))
    db.add(user)
    await db.flush()
    return user


@pytest.fixture
async def alice(db) -> User:
    return await _make_user(db, "alice@test.com")


@pytest.fixture
async def bob(db) -> User:
    return await _make_user(db, "bob@test.com")


def auth(user: User) -> dict:
    return {"Authorization": f"Bearer {create_token(user.id, 'access', 15)}"}

# tests/test_notes_api.py
from tests.conftest import auth


async def test_create_note(client, alice):
    r = await client.post("/v1/notes", json={"title": "বাজারের list"}, headers=auth(alice))
    assert r.status_code == 201
    body = r.json()
    assert body["title"] == "বাজারের list"
    assert body["is_pinned"] is False
    assert "owner_id" not in body              # ভেতরের field বাইরে যায় না


async def test_create_note_requires_login(client):
    r = await client.post("/v1/notes", json={"title": "x"})
    assert r.status_code == 401


async def test_create_note_validation(client, alice):
    r = await client.post("/v1/notes", json={"title": ""}, headers=auth(alice))
    assert r.status_code == 422
    assert "title" in r.json()["error"]["fields"]


async def test_list_shows_only_own_notes(client, alice, bob):
    await client.post("/v1/notes", json={"title": "Alice এর"}, headers=auth(alice))
    await client.post("/v1/notes", json={"title": "Bob এর"}, headers=auth(bob))

    r = await client.get("/v1/notes", headers=auth(alice))
    titles = [n["title"] for n in r.json()["items"]]
    assert titles == ["Alice এর"]

import pytest


@pytest.fixture
async def alice_note(client, alice) -> int:
    r = await client.post("/v1/notes", json={"title": "গোপন"}, headers=auth(alice))
    return r.json()["id"]


@pytest.mark.parametrize(
    ("method", "path_tpl", "body"),
    [
        ("GET", "/v1/notes/{id}", None),
        ("PATCH", "/v1/notes/{id}", {"title": "হ্যাক"}),
        ("DELETE", "/v1/notes/{id}", None),
    ],
)
async def test_other_user_cannot_access(client, bob, alice_note, method, path_tpl, body):
    r = await client.request(method, path_tpl.format(id=alice_note), json=body, headers=auth(bob))
    assert r.status_code in (403, 404)


async def test_expired_token_rejected(client, alice):
    from app.core.security import create_token
    expired = create_token(alice.id, "access", minutes=-1)
    r = await client.get("/v1/notes", headers={"Authorization": f"Bearer {expired}"})
    assert r.status_code == 401
    assert r.json()["error"]["code"] == "TOKEN_EXPIRED"


async def test_refresh_token_cannot_be_used_as_access(client, alice):
    from app.core.security import create_token
    refresh = create_token(alice.id, "refresh", minutes=60)
    r = await client.get("/v1/notes", headers={"Authorization": f"Bearer {refresh}"})
    assert r.status_code == 401

class FakeStorage:
    def __init__(self):
        self.files: dict[str, bytes] = {}

    async def put(self, key: str, data: bytes, content_type: str) -> None:
        self.files[key] = data


@pytest.fixture
def fake_storage(client):
    fake = FakeStorage()
    client._transport.app.dependency_overrides[get_storage] = lambda: fake
    return fake


async def test_upload_image(client, alice, alice_note, fake_storage):
    r = await client.post(
        f"/v1/notes/{alice_note}/image",
        files={"file": ("a.png", PNG_BYTES, "image/png")},
        headers=auth(alice),
    )
    assert r.status_code == 200
    assert len(fake_storage.files) == 1

services:
  postgres:
    image: postgres:17
    env:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: notes_test
    ports: ["5432:5432"]