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

الهدف

جعل TaskFlow جاهزًا للإنتاج من خلال كتابة مجموعة اختبارات شاملة وإضافة طبقات تقوية الإنتاج. بنهاية هذا المعمل، سيكون لديك على الأقل 15 اختبارًا ناجحًا و5 وسائط/ميزات إنتاجية.

هذا المعمل من جزأين: الاختبارات (60 نقطة) وتقوية الإنتاج (40 نقطة).

---

الجزء الأول: الاختبارات (60 نقطة)

1.1 — إعداد الاختبارات (tests/conftest.py)

أنشئ البنية التحتية للاختبارات مع هذه التركيبات:

• قاعدة بيانات الاختبار: أنشئ محرك SQLAlchemy غير متزامن يشير إلى قاعدة بيانات taskflow_test
• إعداد قاعدة البيانات: تركيبة بنطاق الجلسة تنشئ جميع الجداول قبل الاختبارات وتحذفها بعدها
• جلسة async: تركيبة بنطاق الدالة توفر جلسة قاعدة بيانات مع تراجع بعد كل اختبار
• عميل الاختبار: httpx.AsyncClient باستخدام ASGITransport مع تطبيق FastAPI وتجاوز تبعية قاعدة البيانات
• مساعد المصادقة: تركيبة أو دالة مساعدة تسجل مستخدمًا وترجع قاموس ترويسة المصادقة {"Authorization": "Bearer <token>"}

# 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

TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/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)

# نفّذ التركيبات: setup_database, db_session, client, auth_headers

أنشئ أيضًا قسم pytest.ini أو pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

1.2 — اختبارات المصادقة (tests/test_auth.py) — 4+ حالات اختبار

اكتب اختبارات لتدفق المصادقة:

1. التسجيل: POST /api/v1/auth/register ببيانات صالحة يرجع 201 وكائن المستخدم (بدون كلمة المرور في الاستجابة)
2. تسجيل الدخول: POST /api/v1/auth/login ببيانات اعتماد صالحة يرجع 200 و access_token
3. الوصول لمسار محمي: GET /api/v1/projects بـ token صالح يرجع 200
4. رفض token غير صالح: GET /api/v1/projects مع Authorization: Bearer invalidtoken يرجع 401

1.3 — اختبارات CRUD (tests/test_projects.py، tests/test_tasks.py) — 6+ حالات اختبار

اكتب اختبارات لعمليات CRUD للمشاريع والمهام:

1. إنشاء مشروع: POST /api/v1/projects يرجع 201 مع بيانات المشروع
2. سرد المشاريع: GET /api/v1/projects يرجع 200 مع قائمة مشاريع المستخدم المصادق
3. الحصول على مشروع بالمعرف: GET /api/v1/projects/{id} يرجع 200 مع المشروع الصحيح
4. تحديث مشروع: PATCH /api/v1/projects/{id} يرجع 200 مع الحقول المحدثة
5. حذف مشروع: DELETE /api/v1/projects/{id} يرجع 204
6. إنشاء مهمة في مشروع: POST /api/v1/projects/{id}/tasks يرجع 201 مع بيانات المهمة

1.4 — اختبارات الصلاحيات (tests/test_permissions.py) — 3+ حالات اختبار

اكتب اختبارات للتحقق من حدود التخويل:

1. غير العضو لا يمكنه الوصول للمشروع: سجّل مستخدمًا ثانيًا، حاول GET مشروع يملكه المستخدم الأول — توقع 403 أو 404
2. غير المالك لا يمكنه حذف المشروع: المستخدم الثاني يحاول DELETE مشروع المستخدم الأول — توقع 403
3. غير العضو لا يمكنه إنشاء مهمة: المستخدم الثاني يحاول POST مهمة في مشروع المستخدم الأول — توقع 403 أو 404

1.5 — اختبارات التصفح (tests/test_pagination.py) — 2+ حالات اختبار

اكتب اختبارات لمعلمات التصفح بالصفحات:

1. التصفح الافتراضي: أنشئ 5 مشاريع، GET /api/v1/projects يرجع الكل مع عدد total
2. صفحة/حجم مخصص: GET /api/v1/projects?page=1&size=2 يرجع عنصرين بالضبط و total صحيح

ملخص متطلبات الاختبارات

• 15 حالة اختبار على الأقل عبر جميع ملفات الاختبار
• جميع الاختبارات تستخدم @pytest.mark.anyio لـ async
• جميع الاختبارات تستخدم تركيبة client (بدون إعداد تطبيق يدوي)
• كل اختبار مستقل (لا يعتمد اختبار على آثار جانبية لاختبار آخر)

---

الجزء الثاني: تقوية الإنتاج (40 نقطة)

2.1 — تخزين Redis المؤقت (app/cache.py) — نمط Cache-Aside

نفّذ تخزين Redis المؤقت لنقاط النهاية كثيرة القراءة:

• أنشئ عميل Redis غير متزامن باستخدام redis.asyncio.Redis
• نفّذ دوال مساعدة get_cached(key) و set_cached(key, value, ttl)
• نفّذ invalidate_cache(pattern) لحذف المفاتيح بنمط glob
• أضف التخزين المؤقت لـ GET /api/v1/projects بمدة صلاحية 60 ثانية
• أضف التخزين المؤقت لـ GET /api/v1/projects/{id}/tasks بمدة صلاحية 60 ثانية
• أبطل التخزين المؤقت ذي الصلة عند أي عملية POST أو PATCH أو DELETE

# app/cache.py — الواجهة المطلوبة
from redis.asyncio import Redis

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

async def get_cached(key: str) -> dict | None: ...
async def set_cached(key: str, value: dict, ttl: int = 60): ...
async def invalidate_cache(pattern: str): ...

2.2 — وسيط معالج الأخطاء العام (app/middleware/error_handler.py)

أنشئ وسيطًا يلتقط جميع الاستثناءات غير المعالجة ويرجع صيغة JSON متسقة:

{
  "error": "internal_server_error",
  "message": "An unexpected error occurred",
  "detail": null
}

• التقط جميع أنواع Exception في dispatch الوسيط
• سجّل تتبع المكدس الكامل باستخدام logging في Python
• في وضع التصحيح، ضمّن تفاصيل الخطأ في حقل detail؛ في الإنتاج اجعله null
• أرجع رمز حالة 500 للاستثناءات غير المعالجة

2.3 — وسيط تسجيل الطلبات (app/middleware/logging.py)

أنشئ وسيطًا يسجل كل طلب HTTP:

• سجّل: طريقة HTTP، مسار URL، رمز حالة الاستجابة، والمدة بالمللي ثانية
• استخدم وحدة logging في Python مع مسجل مسمى (مثلاً taskflow.access)
• الصيغة: GET /api/v1/projects status=200 duration=12.3ms
• قس المدة باستخدام time.perf_counter()

2.4 — تحديد المعدل على نقاط نهاية المصادقة

أضف تحديد المعدل لمنع محاولات تسجيل الدخول بالقوة الغاشمة:

• حدد POST /api/v1/auth/login بـ 5 طلبات في الدقيقة لكل IP
• استخدم slowapi مع get_remote_address كدالة المفتاح
• أرجع رمز حالة 429 مع رسالة خطأ واضحة عند تجاوز المعدل
• سجّل معالج خطأ محدد المعدل في main.py

2.5 — وسيط CORS

كوّن CORS للإنتاج:

• اسمح بالمصادر من متغير بيئة (ALLOWED_ORIGINS، مفصولة بفواصل)
• الطرق المسموحة: GET, POST, PATCH, DELETE, OPTIONS
• الترويسات المسموحة: Content-Type, Authorization
• السماح ببيانات الاعتماد: true

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS.split(","),
    allow_credentials=True,
    allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization"],
)

2.6 — نقطة نهاية فحص الصحة (GET /health)

أنشئ نقطة نهاية تفحص حالة جميع التبعيات:

• تحقق من اتصال قاعدة البيانات بتنفيذ SELECT 1
• تحقق من اتصال Redis باستدعاء redis_client.ping()
• أرجع 200 مع {"api": "healthy", "database": "healthy", "redis": "healthy"} عند نجاح جميع الفحوصات
• أرجع 503 مع حالة unhealthy لأي فحص فاشل

---

ما يجب تقديمه

يجب أن يحتوي تقديمك على 10 أقسام ملفات في المحرر أدناه. يبدأ كل قسم بعنوان # FILE N:.

---

تلميحات

• استخدم httpx.ASGITransport(app=app) لإنشاء عميل الاختبار غير المتزامن — هذا هو النهج الموصى به من FastAPI
• تذكر أن تتجاوز get_db في تركيبة عميل الاختبار حتى تستخدم الاختبارات قاعدة بيانات الاختبار
• لاختبارات الصلاحيات، أنشئ مستخدمين منفصلين بـ tokens مصادقة منفصلة
• scan_iter في Redis مع match=pattern هي الطريقة الآمنة لـ async لإيجاد المفاتيح للإبطال
• استخدم time.perf_counter() (وليس time.time()) لقياس المدة بدقة
• فحص الصحة يجب أن يرجع 503 (Service Unavailable) إذا كانت أي تبعية معطلة، وليس 500