الاختبارات وتقوية الإنتاج

واجهة TaskFlow البرمجية لديها المصادقة ونماذج قاعدة البيانات ونقاط نهاية CRUD. قبل شحنها، تحتاج شيئين: اختبارات تثبت أنها تعمل وتقوية إنتاجية تبقيها تعمل.

هرم الاختبارات لواجهات البرمجة

الطبقة	ما تختبره	السرعة	الأدوات
الوحدة	الدوال الفردية، المدققات، الأدوات المساعدة	الأسرع	pytest
التكامل	نقطة النهاية + قاعدة البيانات + المصادقة معًا	متوسط	httpx + TestClient
من البداية للنهاية	دورة الطلب الكاملة، التدفقات متعددة الخطوات	الأبطأ	httpx.AsyncClient

لواجهة REST API مثل TaskFlow، اختبارات التكامل تعطيك أكبر قيمة. فهي تمارس المسارات واستعلامات قاعدة البيانات والمصادقة في ضربة واحدة.

التركيبات واختبارات async

يستخدم pytest التركيبات (fixtures) لإعداد وتفكيك تبعيات الاختبار. لـ FastAPI، تحتاج ثلاث تركيبات أساسية: قاعدة بيانات اختبار، وجلسة اختبار، وعميل اختبار. والاختبارات غير المتزامنة هنا تعمل على إضافة pytest التابعة لـ AnyIO، وهي ما يوفّر الواسم @pytest.mark.anyio الذي ستراه على كل اختبار — وتأتي ضمن حزمة anyio الموجودة أصلاً في شجرة التبعيات لأن Starlette وhttpx كليهما مبنيان عليها.

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

from app.main import app
from app.database import Base, get_db
from app.config import settings

# استخدام قاعدة بيانات اختبار منفصلة
TEST_DATABASE_URL = settings.database_url.replace(
    "/taskflow", "/taskflow_test"
)

engine_test = create_async_engine(TEST_DATABASE_URL, echo=False)
async_session_test = async_sessionmaker(engine_test, class_=AsyncSession, expire_on_commit=False)


@pytest.fixture(scope="session")
def anyio_backend():
    """تشغيل المجموعة على asyncio، وجعل تركيبة المحرك بنطاق الجلسة.

    بدون هذا، ترفع التركيبة التالية خطأ ScopeMismatch قبل تشغيل أي اختبار.
    """
    return "asyncio"


@pytest.fixture(scope="session", autouse=True)
async def setup_database(anyio_backend):
    """إنشاء جميع الجداول قبل الاختبارات وحذفها بعدها."""
    async with engine_test.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine_test.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture
async def db_session():
    """توفير جلسة قاعدة بيانات معاملاتية تتراجع بعد كل اختبار."""
    async with async_session_test() as session:
        yield session
        await session.rollback()


@pytest.fixture
async def client(db_session):
    """عميل HTTP للاختبار مع تجاوز تبعية قاعدة البيانات."""
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

تركيبة anyio_backend ليست اختيارية

احذف تلك الأسطر الخمسة ولن ترسب المجموعة في اختبار — بل ستفشل في الانطلاق أصلاً، وفي كل اختبار دفعة واحدة:

ScopeMismatch: You tried to access the module scoped fixture anyio_backend
with a session scoped request object.

والسبب قاعدة نطاق يسهل الوقوع فيها ويصعب تخمينها. كل تركيبة غير متزامنة تحتاج أن تعرف على أي محرك async ستعمل، فتحقن إضافة AnyIO فيها التركيبة anyio_backend. وهذه التركيبة نطاقها الافتراضي هو الوحدة (module)، وpytest يمنع تركيبة أطول عمراً من الاعتماد على أخرى أقصر عمراً. وفي اللحظة التي تجعل فيها setup_database بنطاق scope="session" — وهو ما تريده، لأن إنشاء الجداول وحذفها مرة واحدة لكل مجموعة أسرع بكثير من مرة لكل وحدة — تكون قد أنشأت هذا الاعتماد الممنوع بالضبط.

وتجاوز anyio_backend بنطاق scope="session" يوسّعها لتطابقها، فينحل كل شيء. والقاعدة العامة تستحق الحفظ: التركيبة غير المتزامنة لا يمكن أن تعمّر أطول من anyio_backend، فإن وسّعت التركيبة فوسّع المحرك معها.

TestClient مقابل httpx.AsyncClient

بما أن TaskFlow يستخدم SQLAlchemy غير المتزامن، نستخدم httpx.AsyncClient مع ASGITransport. والعامل الحاسم ليس التفضيل: جلسة قاعدة البيانات غير المتزامنة ملك لحلقة الأحداث التي أنشأتها، وTestClient يشغّل التطبيق على حلقة أخرى.

كتابة اختبارات API فعالة

نظّم اختباراتك حول الإجراءات والصلاحيات:

# tests/test_auth.py
import pytest

@pytest.mark.anyio
async def test_register_user(client):
    response = await client.post("/api/v1/auth/register", json={
        "email": "test@example.com",
        "password": "SecurePass123!",
        "full_name": "Test User"
    })
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"
    assert "password" not in response.json()


@pytest.mark.anyio
async def test_login_returns_token(client):
    # التسجيل أولاً ثم تسجيل الدخول
    await client.post("/api/v1/auth/register", json={
        "email": "login@example.com",
        "password": "SecurePass123!",
        "full_name": "Login User"
    })
    response = await client.post("/api/v1/auth/login", json={
        "email": "login@example.com",
        "password": "SecurePass123!"
    })
    assert response.status_code == 200
    assert "access_token" in response.json()


@pytest.mark.anyio
async def test_protected_route_without_token(client):
    response = await client.get("/api/v1/projects")
    assert response.status_code == 401

تخزين Redis المؤقت: نمط Cache-Aside

نمط cache-aside (التحميل الكسول) مثالي لنقاط نهاية GET: تحقق من Redis أولاً، ثم ارجع لقاعدة البيانات، ثم خزّن النتيجة.

خاصيتان في هذا الشكل تفسران لماذا هو الخيار الافتراضي. لا يُخزَّن إلا ما طُلب فعلاً، فالمخزن البارد يكلّفك طلباً بطيئاً واحداً لا مهمة إحماء. وإذا تعطّل Redis سلك كل طلب مسار الإخفاق — فتبطؤ الواجهة بدل أن تفشل، بشرط أن تدع استدعاء Redis يفشل بهدوء بدل تمرير الاستثناء.

والثمن هو قِدَم البيانات، ومدة الصلاحية هي حيث تضبط تسامحك معه. ستون ثانية تعني أن المستخدم قد يرى قائمة مشاريع متأخرة دقيقة كاملة. هذا مقبول للوحة معلومات وخاطئ لرصيد مالي.

# app/cache.py
import json
from redis.asyncio import Redis

redis_client = Redis(host="localhost", port=6379, db=0, decode_responses=True)

CACHE_TTL = 60  # ثواني


async def get_cached(key: str) -> dict | None:
    """إرجاع البيانات المخزنة مؤقتًا أو None."""
    data = await redis_client.get(key)
    return json.loads(data) if data else None


async def set_cached(key: str, value: dict, ttl: int = CACHE_TTL):
    """تخزين البيانات مع مدة صلاحية."""
    await redis_client.set(key, json.dumps(value), ex=ttl)


async def invalidate_cache(pattern: str):
    """حذف جميع المفاتيح المطابقة لنمط."""
    keys = []
    async for key in redis_client.scan_iter(match=pattern):
        keys.append(key)
    if keys:
        await redis_client.delete(*keys)

الاستخدام في مسار:

@router.get("/projects")
async def list_projects(current_user: User = Depends(get_current_user)):
    cache_key = f"projects:user:{current_user.id}"
    cached = await get_cached(cache_key)
    if cached:
        return cached

    projects = await project_service.list_for_user(current_user.id)
    result = [ProjectSchema.model_validate(p).model_dump() for p in projects]
    await set_cached(cache_key, result, ttl=60)
    return result

عند تغيير البيانات، أبطل التخزين المؤقت:

@router.post("/projects", status_code=201)
async def create_project(data: ProjectCreate, current_user: User = Depends(get_current_user)):
    project = await project_service.create(data, current_user.id)
    await invalidate_cache(f"projects:user:{current_user.id}*")
    return project

وسيط معالجة الأخطاء

معالج استثناءات عام يضمن أن كل خطأ يرجع JSON متسق بدلاً من تتبعات المكدس الخام:

# app/middleware/error_handler.py
from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import logging

logger = logging.getLogger(__name__)


class ErrorHandlerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            return await call_next(request)
        except Exception as exc:
            logger.exception(f"Unhandled error: {exc}")
            return JSONResponse(
                status_code=500,
                content={
                    "error": "internal_server_error",
                    "message": "An unexpected error occurred",
                    "detail": str(exc) if settings.DEBUG else None,
                },
            )

سجّله في main.py:

from app.middleware.error_handler import ErrorHandlerMiddleware

app.add_middleware(ErrorHandlerMiddleware)

تحديد المعدل

احمِ نقاط نهاية المصادقة من هجمات القوة الغاشمة باستخدام slowapi:

# app/middleware/rate_limit.py
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

# في مسار المصادقة:
@router.post("/login")
@limiter.limit("5/minute")
async def login(request: Request, credentials: LoginSchema):
    ...

وسيط تسجيل الطلبات

سجّل كل طلب مع الطريقة والمسار ورمز الحالة والمدة:

# app/middleware/logging.py
import time, logging
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger("taskflow.access")


class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration_ms = (time.perf_counter() - start) * 1000
        logger.info(
            f"{request.method} {request.url.path} "
            f"status={response.status_code} duration={duration_ms:.1f}ms"
        )
        return response

نقطة نهاية فحص الصحة

واجهة API إنتاجية تحتاج فحص صحة يتحقق من جميع التبعيات:

@router.get("/health")
async def health_check(db: AsyncSession = Depends(get_db)):
    checks = {"api": "healthy"}
    # فحص قاعدة البيانات
    try:
        await db.execute(text("SELECT 1"))
        checks["database"] = "healthy"
    except Exception:
        checks["database"] = "unhealthy"
    # فحص Redis
    try:
        await redis_client.ping()
        checks["redis"] = "healthy"
    except Exception:
        checks["redis"] = "unhealthy"

    status = 200 if all(v == "healthy" for v in checks.values()) else 503
    return JSONResponse(content=checks, status_code=status)

في المعمل، ستنفذ كل هذا: مجموعة اختبارات كاملة، تخزين Redis المؤقت، معالجة الأخطاء، تحديد المعدل، وفحص صحة لـ TaskFlow. :::