وبلاگ و سئو: پیش‌نویس/انتشار مقاله + کلمه کلیدی سئو در دیتابیس + OG/canonical/JSON-LD و متای سمت سرور + تغییر رمز و محدودسازی ورود + تست‌ها

- مقاله‌ها ستون status (draft/published) گرفتند — GET /api/site و همه انتهای عمومی فقط منتشرشده را برمی‌گردانند؛ autosave دیگر پیش‌نویس نیمه‌کاره را زنده نمی‌کند
- کلمه کلیدی هدف چک‌لیست سئو فیلد seo_keyphrase شد (قبلاً localStorage بود)
- robots.txt و sitemap.xml از بک‌اند + سرو /blog و /projects با تزریق title/description/OG/JSON-LD از دیتابیس برای خزنده‌های بدون JS (app/seo_pages.py)
- فرانت: applySeo در seo.ts (OG/توییتر/canonical/JSON-LD)، ادغام پیش‌نویس برای ادمین، برچسب پیش‌نویس در وبلاگ، حالت خالی وبلاگ، دکمه انتشار در صفحه نوشتن
- تغییر رمز ادمین POST /api/auth/change-password + دیالوگ «رمز» در نوار مدیریت
- محدودسازی ورود: ۵ تلاش ناموفق در ۱۵ دقیقه از هر IP → 429
- ساخت دکمه + برای ردیف سئوی صفحه‌های بدون ردیف (SingletonPill)
- تست‌های pytest (۱۵ مورد) + پین‌کردن requirements + حذف package.json سرگردان فرانت
This commit is contained in:
MMD
2026-09-07 04:26:18 +03:30
parent a29deb8bb1
commit b31124e1e4
31 changed files with 1202 additions and 152 deletions
+14 -3
View File
@@ -86,6 +86,13 @@ Authorization: Bearer <access_token>
> نام کاربری و رمز عبور پیش‌فرض از تنظیمات `.env` خوانده می‌شود — حتماً در محیط تولید تغییرشان دهید.
تغییر رمز (با توکن):
```
POST /api/auth/change-password
{ "current_password": "...", "new_password": "..." }
```
## فهرست انتهاپوینت‌ها
### عمومی (بدون نیاز به توکن)
@@ -97,8 +104,8 @@ Authorization: Bearer <access_token>
| GET | `/api/services` | خدمات |
| GET | `/api/projects` | همه پروژه‌ها |
| GET | `/api/projects/{slug}` | جزئیات یک پروژه |
| GET | `/api/articles` | همه مقالات |
| GET | `/api/articles/{slug}` | جزئیات یک مقاله |
| GET | `/api/articles` | مقاله‌های منتشرشده (پیش‌نویس نمی‌آید) |
| GET | `/api/articles/{slug}` | جزئیات یک مقاله منتشرشده |
| GET | `/api/testimonials` | نظرات مشتریان |
| GET | `/api/faqs` | سوالات متداول |
| GET | `/api/team` | اعضای تیم |
@@ -109,7 +116,11 @@ Authorization: Bearer <access_token>
| GET | `/api/site` | **همه محتوا در یک درخواست** |
| POST | `/api/contact` | ثبت پیام تماس |
| POST | `/api/newsletter` | عضویت در خبرنامه |
| POST | `/api/auth/login` | ورود ادمین و دریافت توکن |
| POST | `/api/auth/login` | ورود ادمین و دریافت توکن — بعد از ۵ تلاش ناموفق در ۱۵ دقیقه، 429 |
| GET | `/robots.txt` | قوانین خزنده‌ها |
| GET | `/sitemap.xml` | نقشه سایت — مقاله/پروژه‌های منتشرشده |
| GET | `/blog/{slug}` | HTML صفحه مقاله با متای سئوی تزریق‌شده |
| GET | `/projects/{slug}` | HTML صفحه پروژه با متای سئوی تزریق‌شده |
### مدیریت (نیاز به توکن Bearer)
+23 -3
View File
@@ -4,15 +4,17 @@ import uuid
from datetime import datetime, timezone
import boto3
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status
from sqlalchemy import func
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from .. import models, schemas
from ..auth import create_access_token, get_current_admin, verify_password
from ..auth import create_access_token, get_current_admin, hash_password, verify_password
from ..config import settings
from ..database import get_db
from ..rate_limit import clear as clear_rate_limit
from ..rate_limit import ensure_allowed, record_failure
from ..schemas import (
ArticleIn,
ArticleOut,
@@ -23,6 +25,7 @@ from ..schemas import (
BrandIn,
BrandOut,
BrandPatch,
ChangePasswordIn,
ContactSectionIn,
ContactSectionOut,
ContactSectionPatch,
@@ -105,16 +108,33 @@ router = APIRouter(dependencies=[Depends(get_current_admin)])
@auth_router.post("/auth/login", response_model=TokenOut)
def login(payload: LoginIn, db: Session = Depends(get_db)):
def login(payload: LoginIn, request: Request, db: Session = Depends(get_db)):
ensure_allowed(request)
user = db.query(models.AdminUser).filter(models.AdminUser.username == payload.username).first()
if user is None or not verify_password(payload.password, user.password_hash):
record_failure(request)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="نام کاربری یا رمز عبور اشتباه است",
)
clear_rate_limit(request)
return TokenOut(access_token=create_access_token(user.username))
@auth_router.post("/auth/change-password")
def change_password(
payload: ChangePasswordIn,
admin: models.AdminUser = Depends(get_current_admin),
db: Session = Depends(get_db),
):
"""تغییر رمز ادمین واردشده — رمز فعلی باید درست باشد."""
if not verify_password(payload.current_password, admin.password_hash):
raise HTTPException(status_code=400, detail="رمز عبور فعلی اشتباه است")
admin.password_hash = hash_password(payload.new_password)
db.commit()
return {"ok": True}
def register_crud(path: str, model, out_schema, create_schema, patch_schema=None) -> None:
"""ثبت مسیرهای استاندارد CRUD (لیست، ساخت، دریافت، ویرایش کامل، ویرایش جزئی،
حذف و ترتیب‌بندی) روی روتر مدیریت."""
+16 -3
View File
@@ -55,12 +55,21 @@ def get_project(slug: str, db: Session = Depends(get_db)):
@router.get("/articles", response_model=list[ArticleOut])
def get_articles(db: Session = Depends(get_db)):
return db.query(models.Article).order_by(models.Article.position, models.Article.id).all()
return (
db.query(models.Article)
.filter(models.Article.status == "published")
.order_by(models.Article.position, models.Article.id)
.all()
)
@router.get("/articles/{slug}", response_model=ArticleOut)
def get_article(slug: str, db: Session = Depends(get_db)):
article = db.query(models.Article).filter(models.Article.slug == slug).first()
article = (
db.query(models.Article)
.filter(models.Article.slug == slug, models.Article.status == "published")
.first()
)
if article is None:
raise HTTPException(status_code=404, detail="مقاله یافت نشد")
return article
@@ -116,7 +125,11 @@ def get_site(db: Session = Depends(get_db)):
"page_heroes": db.query(models.PageHeroContent).order_by(models.PageHeroContent.id).all(),
"services": db.query(models.Service).order_by(models.Service.position, models.Service.id).all(),
"projects": db.query(models.Project).order_by(models.Project.position, models.Project.id).all(),
"articles": db.query(models.Article).order_by(models.Article.position, models.Article.id).all(),
# فقط مقاله‌های منتشرشده — پیش‌نویس‌ها از انتهای مدیریتی خوانده می‌شوند
"articles": db.query(models.Article)
.filter(models.Article.status == "published")
.order_by(models.Article.position, models.Article.id)
.all(),
"testimonials": db.query(models.Testimonial).order_by(models.Testimonial.position, models.Testimonial.id).all(),
"faqs": db.query(models.Faq).order_by(models.Faq.position, models.Faq.id).all(),
"team": db.query(models.TeamMember).order_by(models.TeamMember.position, models.TeamMember.id).all(),
+6 -1
View File
@@ -23,9 +23,14 @@ class Settings(BaseSettings):
s3_access_key: str = "aramesh"
s3_secret_key: str = "aramesh-minio-secret"
s3_bucket: str = "aramesh-media"
# پیشوند عمومی مسیر تصاویر — nginx آن را به باکت پروکسی می‌کند
# پیشوند عمومی مسیر تصاویر — Traefik آن را به باکت پروکسی می‌کند
media_public_base: str = "/media"
media_max_upload_mb: int = 10
# نشانی کامل سایت (مثل https://aramesh.example) — برای sitemap و آدرس مطلق OG
site_url: str = ""
# مسیر بیلد فرانت‌اند (شامل index.html) — برای سرو HTML با متای سئو در صفحات جزئیات
frontend_dist: str = "../frontend/dist"
settings = Settings()
+3
View File
@@ -6,6 +6,7 @@ from .api import admin, forms, public
from .config import settings
from .database import Base, engine
from .migrations import run_all as run_migrations
from .seo_pages import router as seo_router
Base.metadata.create_all(bind=engine)
run_migrations()
@@ -33,6 +34,8 @@ app.include_router(public.router, prefix="/api", tags=["عمومی"])
app.include_router(forms.router, prefix="/api", tags=["فرم‌ها"])
app.include_router(admin.auth_router, prefix="/api", tags=["احراز هویت"])
app.include_router(admin.router, prefix="/api/admin", tags=["پنل مدیریت"])
# robots.txt، sitemap.xml و HTML صفحات وبلاگ/پروژه با متای سئو — در ریشه دامنه
app.include_router(seo_router, tags=["سئو"])
@app.get("/")
+7
View File
@@ -25,6 +25,13 @@ def run_ddl_migrations() -> None:
conn.execute(
text(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS position INTEGER NOT NULL DEFAULT 0")
)
# مقالات موجود منتشر شده‌اند؛ فقط مقاله‌های تازه می‌توانند پیش‌نویس باشند
conn.execute(
text("ALTER TABLE articles ADD COLUMN IF NOT EXISTS status VARCHAR(16) NOT NULL DEFAULT 'published'")
)
conn.execute(
text("ALTER TABLE articles ADD COLUMN IF NOT EXISTS seo_keyphrase VARCHAR(128) NOT NULL DEFAULT ''")
)
def paragraphs_to_doc(paragraphs: list) -> dict:
+4
View File
@@ -78,6 +78,10 @@ class Article(Base):
read: Mapped[str] = mapped_column(String(32), default="")
excerpt: Mapped[str] = mapped_column(Text, default="")
body: Mapped[list] = mapped_column(JSON, default=list)
# draft | published — پیش‌نویس فقط برای ادمین دیده می‌شود
status: Mapped[str] = mapped_column(String(16), default="published")
# کلمه کلیدی هدف چک‌لیست سئو — راهنمای نوشتن، در سایت نمایش داده نمی‌شود
seo_keyphrase: Mapped[str] = mapped_column(String(128), default="")
class Testimonial(Base):
+49
View File
@@ -0,0 +1,49 @@
"""محدودسازی تلاش ورود — جلوگیری از حمله brute-force روی /api/auth/login.
پیش‌فرض: حداکثر ۵ ورود ناموفق در هر ۱۵ دقیقه برای هر IP؛ ورود موفق شمارنده را پاک می‌کند.
شمارنده در حافظه همین پروسه است — با ری‌استارت بک‌اند پاک می‌شود و همین کافی است.
"""
import time
from fastapi import HTTPException, Request, status
MAX_FAILED_ATTEMPTS = 5
WINDOW_SECONDS = 15 * 60
_failed: dict[str, list[float]] = {}
def client_ip(request: Request) -> str:
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
return forwarded.split(",")[0].strip()
return request.client.host if request.client else "unknown"
def _prune(key: str, now: float) -> list[float]:
recent = [t for t in _failed.get(key, []) if now - t < WINDOW_SECONDS]
_failed[key] = recent
return recent
def ensure_allowed(request: Request) -> None:
"""اگر این IP سهمیه تلاش ناموفقش پر شده باشد، درخواست با 429 رد می‌شود."""
key = client_ip(request)
recent = _prune(key, time.monotonic())
if len(recent) >= MAX_FAILED_ATTEMPTS:
retry_after = max(1, int(WINDOW_SECONDS - (time.monotonic() - recent[0])))
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="تلاش‌های ناموفق زیاد است — بعداً دوباره امتحان کنید",
headers={"Retry-After": str(retry_after)},
)
def record_failure(request: Request) -> None:
_failed.setdefault(client_ip(request), []).append(time.monotonic())
def clear(request: Request) -> None:
"""با ورود موفق، تاریخچه تلاش‌های ناموفق IP پاک می‌شود."""
_failed.pop(client_ip(request), None)
+11
View File
@@ -59,6 +59,8 @@ class ArticleOut(ORMModel):
read: str
excerpt: str
body: dict
status: str
seo_keyphrase: str
class TestimonialOut(ORMModel):
@@ -292,6 +294,11 @@ class TokenOut(BaseModel):
token_type: str = "bearer"
class ChangePasswordIn(BaseModel):
current_password: str
new_password: str = Field(min_length=8, max_length=128)
class BrandIn(BaseModel):
name: str = ""
logo: str = ""
@@ -346,6 +353,8 @@ class ArticleIn(BaseModel):
read: str = ""
excerpt: str = ""
body: dict = Field(default_factory=lambda: dict(EMPTY_DOC))
status: Literal["draft", "published"] = "draft"
seo_keyphrase: str = Field(default="", max_length=128)
_validate_body = field_validator("body")(validate_doc_body)
@@ -548,6 +557,8 @@ class ArticlePatch(BaseModel):
read: Optional[str] = None
excerpt: Optional[str] = None
body: Optional[dict] = None
status: Optional[Literal["draft", "published"]] = None
seo_keyphrase: Optional[str] = None
_validate_body = field_validator("body")(validate_doc_body)
TestimonialPatch = _make_patch("TestimonialPatch", TestimonialIn)
FaqPatch = _make_patch("FaqPatch", FaqIn)
+239
View File
@@ -0,0 +1,239 @@
"""سئوی سمت سرور — robots.txt، sitemap.xml و سرو HTML صفحات وبلاگ/پروژه با متای تزریق‌شده.
فرانت‌اند یک SPA است؛ خزنده‌های اجتماعی (تلگرام، واتس‌اپ و...) جاوااسکریپت اجرا نمی‌کنند.
این ماژول همان index.html بیلد فرانت‌اند را می‌خواند و پیش از ارسال، تگ‌های
title / description / OG / canonical / JSON-LD را از دیتابیس داخل آن تزریق می‌کند —
بازدیدکننده‌های عادی همان SPA همیشگی را می‌بینند و خزنده‌ها متای درست را می‌خوانند.
"""
import html
import json
import re
from pathlib import Path
from fastapi import APIRouter, Depends, Response
from fastapi.responses import HTMLResponse
from sqlalchemy.orm import Session
from . import models
from .config import settings
from .database import get_db
router = APIRouter()
STATIC_PATHS = ["/", "/about", "/services", "/projects", "/investment", "/blog", "/faq", "/contact"]
def _absolute(path: str) -> str:
if not path:
return ""
if path.startswith(("http://", "https://")):
return path
site = settings.site_url.rstrip("/")
return f"{site}{path}" if site else path
def _read_index() -> str | None:
try:
return (Path(settings.frontend_dist) / "index.html").read_text(encoding="utf-8")
except OSError:
return None
def _doc_text(doc: object, limit: int = 300) -> str:
"""متن ساده سند TipTap — برای noscript و پیش‌نمایش."""
words: list[str] = []
def walk(node: object) -> None:
if not isinstance(node, dict):
return
if node.get("type") == "text" and isinstance(node.get("text"), str):
words.append(node["text"])
for child in node.get("content") or []:
walk(child)
if isinstance(doc, dict):
for block in doc.get("content") or []:
walk(block)
text = " ".join(words).strip()
return text[:limit] + ("…" if len(text) > limit else "")
def _shell(
*,
title: str,
description: str,
path: str,
og_image: str = "",
og_type: str = "website",
jsonld: dict | None = None,
noscript: str = "",
) -> HTMLResponse:
"""index.html فرانت‌اند با متای تزریق‌شده؛ اگر بیلد در دسترس نبود، صفحه حداقلی."""
title = title.strip() or "گروه ساختمانی آرامش"
description = description.strip()
canonical = _absolute(path)
image = _absolute(og_image)
esc = html.escape
# تگ title جای تگ موجود index.html را می‌گیرد و در بلوک تکرار نمی‌شود
tags = [
f'<meta name="description" content="{esc(description)}" />',
f'<meta property="og:title" content="{esc(title)}" />',
f'<meta property="og:description" content="{esc(description)}" />',
f'<meta property="og:type" content="{esc(og_type)}" />',
f'<meta property="og:url" content="{esc(canonical)}" />',
f'<meta name="twitter:card" content="summary_large_image" />',
]
if image:
tags.append(f'<meta property="og:image" content="{esc(image)}" />')
if jsonld:
payload = json.dumps(jsonld, ensure_ascii=False).replace("</", "<\\/")
tags.append(f'<script type="application/ld+json">{payload}</script>')
block = "\n ".join(tags)
raw = _read_index()
if raw is None:
# بیلد فرانت‌اند سوار نشده — صفحه حداقلی با خودِ متا تا خزنده‌ها دست خالی نمانند
body = f"<h1>{esc(title)}</h1><p>{esc(description)}</p>"
return HTMLResponse(
f'<!doctype html><html lang="fa" dir="rtl"><head><meta charset="utf-8">'
f"<title>{esc(title)}</title>{block}</head><body>{body}</body></html>"
)
raw = re.sub(r"<title>.*?</title>", f"<title>{esc(title)}</title>", raw, count=1, flags=re.S)
raw = raw.replace("</head>", f" {block}\n </head>", 1)
if noscript:
raw = raw.replace("</body>", f"<noscript>{noscript}</noscript>\n </body>", 1)
return HTMLResponse(raw)
def _brand_name(db: Session) -> str:
brand = db.query(models.Brand).first()
return brand.name if brand else "گروه ساختمانی آرامش"
def _page_meta(db: Session, page_key: str) -> models.SeoMeta | None:
return db.query(models.SeoMeta).filter(models.SeoMeta.page_key == page_key).first()
# ---------- robots.txt و sitemap.xml ----------
@router.get("/robots.txt", response_class=Response)
def robots(db: Session = Depends(get_db)):
lines = ["User-agent: *", "Disallow: /admin", "Disallow: /api", ""]
if settings.site_url:
lines.append(f"Sitemap: {_absolute('/sitemap.xml')}")
return Response(content="\n".join(lines), media_type="text/plain; charset=utf-8")
@router.get("/sitemap.xml", response_class=Response)
def sitemap(db: Session = Depends(get_db)):
urls = list(STATIC_PATHS)
urls += [
f"/blog/{a.slug}"
for a in db.query(models.Article)
.filter(models.Article.status == "published")
.order_by(models.Article.position, models.Article.id)
.all()
]
urls += [
f"/projects/{p.slug}"
for p in db.query(models.Project).order_by(models.Project.position, models.Project.id).all()
]
items = "\n".join(
f" <url><loc>{html.escape(_absolute(u))}</loc></url>" for u in urls
)
xml = (
'<?xml version="1.0" encoding="UTF-8"?>\n'
'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">\n'
f"{items}\n</urlset>\n"
)
return Response(content=xml, media_type="application/xml; charset=utf-8")
# ---------- صفحات وبلاگ ----------
@router.get("/blog", response_class=HTMLResponse)
def blog_list(db: Session = Depends(get_db)):
meta = _page_meta(db, "blog")
return _shell(
title=meta.title if meta and meta.title else "وبلاگ | " + _brand_name(db),
description=meta.description if meta else "",
path="/blog",
og_image=meta.image if meta else "",
)
@router.get("/blog/{slug}", response_class=HTMLResponse)
def blog_detail(slug: str, db: Session = Depends(get_db)):
article = (
db.query(models.Article)
.filter(models.Article.slug == slug, models.Article.status == "published")
.first()
)
if article is None:
# مقاله ناموجود یا پیش‌نویس — پوسته ساده بدون متا؛ SPA پیام «یافت نشد» را نشان می‌دهد
return _shell(title=_brand_name(db), description="", path=f"/blog/{slug}")
brand = _brand_name(db)
first_text = _doc_text(article.body) if not isinstance(article.body, list) else " ".join(article.body[:2])
jsonld = {
"@context": "https://schema.org",
"@type": "Article",
"headline": article.title,
"description": article.excerpt,
"image": _absolute(article.image),
"inLanguage": "fa",
"author": {"@type": "Organization", "name": brand},
"publisher": {"@type": "Organization", "name": brand},
"mainEntityOfPage": _absolute(f"/blog/{article.slug}"),
}
noscript = f"<h1>{html.escape(article.title)}</h1><p>{html.escape(first_text)}</p>"
return _shell(
title=f"{article.title} | {brand}",
description=article.excerpt,
path=f"/blog/{article.slug}",
og_image=article.image,
og_type="article",
jsonld=jsonld,
noscript=noscript,
)
# ---------- صفحات پروژه ----------
@router.get("/projects", response_class=HTMLResponse)
def projects_list(db: Session = Depends(get_db)):
meta = _page_meta(db, "projects")
return _shell(
title=meta.title if meta and meta.title else "پروژه‌ها | " + _brand_name(db),
description=meta.description if meta else "",
path="/projects",
)
@router.get("/projects/{slug}", response_class=HTMLResponse)
def project_detail(slug: str, db: Session = Depends(get_db)):
project = db.query(models.Project).filter(models.Project.slug == slug).first()
if project is None:
return _shell(title=_brand_name(db), description="", path=f"/projects/{slug}")
brand = _brand_name(db)
return _shell(
title=f"{project.title} | {brand}",
description=project.summary,
path=f"/projects/{project.slug}",
og_image=project.image,
jsonld={
"@context": "https://schema.org",
"@type": "WebPage",
"name": project.title,
"description": project.summary,
"image": _absolute(project.image),
"inLanguage": "fa",
},
)
+4
View File
@@ -0,0 +1,4 @@
# وابستگی‌های توسعه و تست — pip install -r requirements-dev.txt
-r requirements.txt
pytest==8.3.4
httpx==0.27.2
+13 -11
View File
@@ -1,11 +1,13 @@
fastapi>=0.115.8
uvicorn[standard]>=0.34.0
sqlalchemy>=2.0.36
psycopg[binary]>=3.2.3
pydantic>=2.10.4
pydantic-settings>=2.6.1
PyJWT>=2.10.1
bcrypt>=4.2.1
email-validator>=2.2.0
boto3>=1.35.0
python-multipart>=0.0.18
# نسخه‌ها دقیق پین شده‌اند تا ارتقای خودکار یک وابستگی، استقرار را نشکند.
# ارتقا: نسخه را دستی بالا ببرید و روی محیط آماده‌سازی تست کنید.
fastapi==0.115.8
uvicorn[standard]==0.34.0
sqlalchemy==2.0.36
psycopg[binary]==3.2.3
pydantic==2.10.4
pydantic-settings==2.6.1
PyJWT==2.10.1
bcrypt==4.2.1
email-validator==2.2.0
boto3==1.35.0
python-multipart==0.0.18
+93
View File
@@ -0,0 +1,93 @@
"""پیکربندی تست‌ها — دیتابیس SQLite فایل‌محور و داده‌های حداقلی.
متغیرهای محیطی پیش از ایمپورت app تنظیم می‌شوند تا settings همان‌ها را بخواند.
"""
import os
import pathlib
os.environ["DATABASE_URL"] = "sqlite:///./test.db?check_same_thread=false"
os.environ["SECRET_KEY"] = "test-secret-key"
os.environ["ADMIN_USERNAME"] = "admin"
os.environ["ADMIN_PASSWORD"] = "test-admin-pass-1"
os.environ["SITE_URL"] = "https://test.example"
import pytest
from fastapi.testclient import TestClient
from app import models
from app.auth import hash_password
from app.config import settings
from app.database import Base, SessionLocal, engine
from app.main import app
TEST_DB = pathlib.Path("test.db")
DOC = {"type": "doc", "content": [{"type": "paragraph"}]}
@pytest.fixture(scope="session", autouse=True)
def seed_db(tmp_path_factory):
# اتصالات ساخت جدول در ایمپورت ماژول باید بسته شوند تا فایل قدیمی حذف شود
engine.dispose()
if TEST_DB.exists():
TEST_DB.unlink()
Base.metadata.create_all(bind=engine)
# dist ساختگی فرانت‌اند برای تست تزریق متا در صفحات جزئیات
dist = tmp_path_factory.mktemp("dist")
(dist / "index.html").write_text(
"<!doctype html><html><head><title>Shell</title></head>"
"<body><div id=\"root\"></div></body></html>",
encoding="utf-8",
)
settings.frontend_dist = str(dist)
settings.site_url = "https://test.example"
db = SessionLocal()
try:
db.add(models.AdminUser(username="admin", password_hash=hash_password("test-admin-pass-1")))
db.add(models.Brand(name="گروه ساختمانی آرامش"))
db.add(
models.Article(
slug="published-post", title="مقاله منتشرشده", status="published",
excerpt="چکیده مقاله", body=DOC, position=1,
)
)
db.add(
models.Article(
slug="draft-post", title="پیش‌نویس", status="draft",
excerpt="چکیده پیش‌نویس", body=DOC, position=2,
)
)
db.commit()
finally:
db.close()
yield
db.close()
engine.dispose()
if TEST_DB.exists():
TEST_DB.unlink()
@pytest.fixture(autouse=True)
def reset_rate_limiter():
from app import rate_limit
rate_limit._failed.clear()
yield
rate_limit._failed.clear()
@pytest.fixture()
def client():
return TestClient(app)
@pytest.fixture()
def auth_client(client):
res = client.post("/api/auth/login", json={"username": "admin", "password": "test-admin-pass-1"})
assert res.status_code == 200, res.text
token = res.json()["access_token"]
client.headers.update({"Authorization": f"Bearer {token}"})
yield client
client.headers.pop("Authorization", None)
+182
View File
@@ -0,0 +1,182 @@
"""تست‌های API — مسیرهای عمومی، احراز هویت، CRUD مدیریت، پیش‌نویس مقاله و سئو."""
import re
DOC = {"type": "doc", "content": [{"type": "paragraph"}]}
# ---------- عمومی ----------
def test_health(client):
res = client.get("/api/health")
assert res.status_code == 200
assert res.json()["status"] == "ok"
def test_site_hides_draft_articles(client):
res = client.get("/api/site")
assert res.status_code == 200
slugs = [a["slug"] for a in res.json()["articles"]]
assert "published-post" in slugs
assert "draft-post" not in slugs
def test_public_articles_hide_drafts(client):
res = client.get("/api/articles")
assert all(a["status"] == "published" for a in res.json())
assert client.get("/api/articles/draft-post").status_code == 404
def test_admin_endpoints_reject_anonymous(client):
assert client.get("/api/admin/articles").status_code == 401
# ---------- احراز هویت ----------
def test_login_success_and_failure(client):
assert (
client.post("/api/auth/login", json={"username": "admin", "password": "wrong"}).status_code
== 401
)
res = client.post("/api/auth/login", json={"username": "admin", "password": "test-admin-pass-1"})
assert res.status_code == 200
assert res.json()["access_token"]
def test_change_password(auth_client, client):
res = auth_client.post(
"/api/auth/change-password",
json={"current_password": "wrong", "new_password": "brand-new-pass-9"},
)
assert res.status_code == 400
res = auth_client.post(
"/api/auth/change-password",
json={"current_password": "test-admin-pass-1", "new_password": "brand-new-pass-9"},
)
assert res.status_code == 200
# رمز قدیمی رد و رمز جدید قبول می‌شود
assert (
client.post("/api/auth/login", json={"username": "admin", "password": "test-admin-pass-1"}).status_code
== 401
)
assert (
client.post("/api/auth/login", json={"username": "admin", "password": "brand-new-pass-9"}).status_code
== 200
)
# بازگرداندن رمز برای تست‌های بعدی
res = client.post("/api/auth/login", json={"username": "admin", "password": "brand-new-pass-9"})
auth = {"Authorization": f"Bearer {res.json()['access_token']}"}
res = client.post(
"/api/auth/change-password",
json={"current_password": "brand-new-pass-9", "new_password": "test-admin-pass-1"},
headers=auth,
)
assert res.status_code == 200
def test_login_rate_limited(client):
for _ in range(5):
res = client.post("/api/auth/login", json={"username": "admin", "password": "wrong"})
assert res.status_code == 401
res = client.post("/api/auth/login", json={"username": "admin", "password": "wrong"})
assert res.status_code == 429
assert "Retry-After" in res.headers
# ---------- CRUD مدیریت ----------
def test_article_crud_and_draft_flow(auth_client, client):
created = auth_client.post(
"/api/admin/articles",
json={"slug": "flow-post", "title": "مقاله جریان", "body": DOC},
)
assert created.status_code == 201, created.text
article = created.json()
# مقاله تازه پیش‌نویس است و در سایت عمومی دیده نمی‌شود
assert article["status"] == "draft"
assert "flow-post" not in [a["slug"] for a in client.get("/api/articles").json()]
res = auth_client.patch(f"/api/admin/articles/{article['id']}", json={"seo_keyphrase": "ساختمان"})
assert res.json()["seo_keyphrase"] == "ساختمان"
res = auth_client.patch(f"/api/admin/articles/{article['id']}", json={"status": "published"})
assert res.status_code == 200
assert "flow-post" in [a["slug"] for a in client.get("/api/articles").json()]
assert auth_client.delete(f"/api/admin/articles/{article['id']}").status_code == 204
assert "flow-post" not in [a["slug"] for a in client.get("/api/articles").json()]
def test_service_roundtrip_and_reorder(auth_client):
ids = []
for name in ("خدمت الف", "خدمت ب"):
res = auth_client.post("/api/admin/services", json={"title": name})
assert res.status_code == 201
ids.append(res.json()["id"])
res = auth_client.post("/api/admin/services/reorder", json={"ids": ids[::-1]})
assert res.status_code == 200
ordered = [s["id"] for s in res.json() if s["id"] in ids]
assert ordered == ids[::-1]
res = auth_client.patch(f"/api/admin/services/{ids[0]}", json={"desc": "توضیح تازه"})
assert res.json()["desc"] == "توضیح تازه"
for i in ids:
assert auth_client.delete(f"/api/admin/services/{i}").status_code == 204
def test_seo_meta_create_and_patch(auth_client):
res = auth_client.post(
"/api/admin/seo-meta", json={"page_key": "test-page", "title": "عنوان", "description": "توضیح"}
)
assert res.status_code == 201
meta_id = res.json()["id"]
res = auth_client.patch(f"/api/admin/seo-meta/{meta_id}", json={"title": "عنوان تازه"})
assert res.json()["title"] == "عنوان تازه"
assert auth_client.delete(f"/api/admin/seo-meta/{meta_id}").status_code == 204
# ---------- سئو ----------
def test_robots_txt(client):
res = client.get("/robots.txt")
assert res.status_code == 200
assert "Disallow: /admin" in res.text
assert "Sitemap: https://test.example/sitemap.xml" in res.text
def test_sitemap_includes_published_only(client):
res = client.get("/sitemap.xml")
assert res.status_code == 200
assert "/blog/published-post" in res.text
assert "/blog/draft-post" not in res.text
assert res.text.startswith("<?xml")
def test_head_injection_for_article(client):
res = client.get("/blog/published-post")
assert res.status_code == 200
assert "مقاله منتشرشده" in res.text
assert 'property="og:title"' in res.text
assert "application/ld+json" in res.text
assert "https://test.example/blog/published-post" in res.text
def test_head_not_injected_for_draft(client):
res = client.get("/blog/draft-post")
assert res.status_code == 200
assert "پیش‌نویس" not in res.text # عنوان پیش‌نویس افشا نمی‌شود
assert "application/ld+json" not in res.text
def test_head_replaces_title_once(client):
res = client.get("/blog/published-post")
assert len(re.findall(r"<title>", res.text)) == 1