tested the api using pytest, fixtures, mocking external services

Author: ZeeshanML

pyproject.toml

@@ -18,3 +18,9 @@ dependencies = [
     "pyjwt>=2.11.0",
     "sqlalchemy>=2.0.46",
 ]
+
+[dependency-groups]
+dev = [
+    "moto[s3]>=5.1.22",
+    "pytest>=9.0.3",
+]

tests/__init__.py

@@ -0,0 +1,146 @@
+import os
+from collections.abc import AsyncGenerator
+
+
+os.environ["DATABASE_URL"] = (
+    "postgresql+psycopg://bloguser:blogpass@localhost/test_blog"
+)
+os.environ["S3_BUCKET_NAME"] = "test-bucket"
+os.environ["SECRET_KEY"] = "test-secret-key-for-testing-only"
+
+os.environ["S3_ACCESS_KEY_ID"] = "testing"
+os.environ["S3_SECRET_ACCESS_KEY"] = "testing"
+os.environ["S3_REGION"] = "eu-north-1"
+
+os.environ["AWS_ACCESS_KEY_ID"] = "testing"
+os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
+os.environ["AWS_DEFAULT_REGION"] = "eu-north-1"
+
+import boto3
+import pytest
+from httpx import ASGITransport, AsyncClient
+from moto import mock_aws
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from database import Base, get_db
+from main import app
+
+pytest_plugins = ["anyio"]
+
+@pytest.fixture(scope="session")
+def anyio_backend():
+    return "asyncio"
+
+@pytest.fixture(scope="session")
+def test_engine():
+    engine = create_async_engine(
+        os.environ["DATABASE_URL"],
+        poolclass=NullPool,
+    )
+    return engine
+
+
+@pytest.fixture(scope="session")
+async def setup_database(test_engine):
+    async with test_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    yield
+
+    async with test_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+    await test_engine.dispose()
+
+@pytest.fixture
+async def db_session(
+    test_engine,
+    setup_database,
+) -> AsyncGenerator[AsyncSession]:
+    conn = await test_engine.connect()
+    trans = await conn.begin()
+
+    test_async_session = async_sessionmaker(
+        bind=conn,
+        class_=AsyncSession,
+        expire_on_commit=False,
+        join_transaction_mode="create_savepoint",
+    )
+
+    async with test_async_session() as session:
+        try:
+            yield session
+        finally:
+            await session.close()
+            await trans.rollback()
+            await conn.close()
+
+
+@pytest.fixture
+def mocked_aws():
+    with mock_aws():
+        s3 = boto3.client("s3", region_name="eu-north-1")
+        s3.create_bucket(
+            Bucket=os.environ["S3_BUCKET_NAME"],
+            CreateBucketConfiguration={"LocationConstraint": "eu-north-1"},
+        )
+        yield s3
+
+
+@pytest.fixture
+async def client(
+    db_session: AsyncSession,
+    mocked_aws,
+) -> AsyncGenerator[AsyncClient]:
+
+    async def override_get_db():
+        yield db_session
+
+    app.dependency_overrides[get_db] = override_get_db
+
+    async with AsyncClient(
+        transport=ASGITransport(app=app),
+        base_url="http://test",
+    ) as ac:
+        yield ac
+
+    app.dependency_overrides.clear()
+
+
+async def create_test_user(
+    client: AsyncClient,
+    username: str = "testuser",
+    email: str = "test@example.com",
+    password: str = "testpassword123",
+) -> dict:
+    response = await client.post(
+        "/api/users",
+        json={
+            "username": username,
+            "email": email,
+            "password": password,
+        },
+    )
+    assert response.status_code == 201, f"Failed to create user: {response.text}"
+    return response.json()
+
+
+async def login_user(
+    client: AsyncClient,
+    email: str = "test@example.com",
+    password: str = "testpassword123",
+) -> str:
+    response = await client.post(
+        "/api/users/token",
+        data={
+            "username": email,
+            "password": password,
+        },
+    )
+    assert response.status_code == 200, f"Failed to login: {response.text}"
+    return response.json()["access_token"]
+
+
+def auth_header(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}

tests/conftest.py

@@ -0,0 +1,147 @@
+import pytest
+from httpx import AsyncClient
+
+from tests.conftest import auth_header, create_test_user, login_user
+
+@pytest.mark.anyio
+async def test_get_posts_empty(client: AsyncClient):
+    response = await client.get("/api/posts")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["posts"] == []
+    assert data["total"] == 0
+    assert data["has_more"] is False
+
+
+@pytest.mark.anyio
+async def test_get_post_not_found(client: AsyncClient):
+    response = await client.get("/api/posts/999")
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Post not found"
+
+
+@pytest.mark.anyio
+async def test_create_post_success(client: AsyncClient):
+    user = await create_test_user(client)
+    token = await login_user(client)
+    headers = auth_header(token)
+
+    response = await client.post(
+        "/api/posts",
+        json={"title": "My First Post", "content": "This is the content"},
+        headers=headers,
+    )
+
+    assert response.status_code == 201
+    data = response.json()
+    assert data["title"] == "My First Post"
+    assert data["content"] == "This is the content"
+    assert data["user_id"] == user["id"]
+    assert "id" in data
+    assert "date_posted" in data
+    assert data["author"]["username"] == "testuser"
+
+
+@pytest.mark.anyio
+async def test_create_post_unauthorized(client: AsyncClient):
+    response = await client.post(
+        "/api/posts",
+        json={"title": "Test Post", "content": "Test content"},
+    )
+
+    assert response.status_code == 401
+    assert response.json()["detail"] == "Not authenticated"
+
+
+@pytest.mark.anyio
+async def test_update_post_success(client: AsyncClient):
+    await create_test_user(client)
+    token = await login_user(client)
+    headers = auth_header(token)
+
+    response = await client.post(
+        "/api/posts",
+        json={"title": "Original Title", "content": "Original content"},
+        headers=headers,
+    )
+    post_id = response.json()["id"]
+
+    response = await client.patch(
+        f"/api/posts/{post_id}",
+        json={"title": "Updated Title"},
+        headers=headers,
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["title"] == "Updated Title"
+    assert data["content"] == "Original content"
+
+
+@pytest.mark.anyio
+async def test_update_post_wrong_user(client: AsyncClient):
+    await create_test_user(client, username="user1", email="user1@example.com")
+    token1 = await login_user(client, email="user1@example.com")
+
+    response = await client.post(
+        "/api/posts",
+        json={"title": "User 1's Post", "content": "Only user 1 can edit this"},
+        headers=auth_header(token1),
+    )
+    post_id = response.json()["id"]
+
+    await create_test_user(client, username="user2", email="user2@example.com")
+    token2 = await login_user(client, email="user2@example.com")
+
+    response = await client.patch(
+        f"/api/posts/{post_id}",
+        json={"title": "Hacked Title"},
+        headers=auth_header(token2),
+    )
+
+    assert response.status_code == 403
+    assert response.json()["detail"] == "Not authorized to update this post"
+
+
+@pytest.mark.anyio
+async def test_get_posts_with_pagination(client: AsyncClient):
+    await create_test_user(client)
+    token = await login_user(client)
+    headers = auth_header(token)
+
+    for i in range(5):
+        response = await client.post(
+            "/api/posts",
+            json={"title": f"Post {i}", "content": f"Content for post {i}"},
+            headers=headers,
+        )
+        assert response.status_code == 201
+
+    response = await client.get("/api/posts")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["total"] == 5
+    assert len(data["posts"]) == 5
+    assert data["has_more"] is False
+
+    response = await client.get("/api/posts?limit=2")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["total"] == 5
+    assert len(data["posts"]) == 2
+    assert data["has_more"] is True
+
+    response = await client.get("/api/posts?skip=2&limit=2")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["total"] == 5
+    assert len(data["posts"]) == 2
+    assert data["skip"] == 2
+    assert data["limit"] == 2
+
+
+
+
+