Files
aramesh-backend/backend/tests/test_api.py
T

672 lines
24 KiB
Python

"""تست‌های API — مسیرهای عمومی، احراز هویت، CRUD مدیریت، پیش‌نویس مقاله و سئو."""
import re
import socket
DOC = {"type": "doc", "content": [{"type": "paragraph"}]}
# ---------- عمومی ----------
def test_live(client):
"""liveness — پروسه زنده است، بدون وابستگی."""
res = client.get("/api/live")
assert res.status_code == 200
assert res.json()["status"] == "ok"
def test_health_deep(client):
"""بررسی عمق — DB (SQLite در تست) سالم است؛ MinIO در محیط تست در دسترس نیست
پس /api/health باید 503 و وضعیت degraded با جزئیات هر وابستگی بدهد."""
res = client.get("/api/health")
assert res.status_code == 503
body = res.json()
assert body["status"] == "degraded"
assert body["checks"]["database"] == "ok"
assert body["checks"]["storage"] == "error"
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
# ---------- دوزبانه ----------
def test_bilingual_project_and_en_seo_shell(auth_client, client):
created = auth_client.post(
"/api/admin/projects",
json={
"slug": "bilingual-tower",
"title": "برج دوزبانه",
"title_en": "Bilingual Tower",
"summary": "خلاصه فارسی",
"summary_en": "English summary",
"body": [],
"body_en": ["English paragraph"],
},
)
assert created.status_code == 201, created.text
pid = created.json()["id"]
# GET /api/site هر دو زبان را برمی‌گرداند
project = next(p for p in client.get("/api/site").json()["projects"] if p["slug"] == "bilingual-tower")
assert project["title"] == "برج دوزبانه"
assert project["title_en"] == "Bilingual Tower"
assert project["body_en"] == ["English paragraph"]
# پوسته سئوی انگلیسی — جهت و متای انگلیسی و hreflang متقابل
res = client.get("/en/projects/bilingual-tower")
assert res.status_code == 200
assert 'lang="en" dir="ltr"' in res.text
assert "Bilingual Tower" in res.text
assert "English summary" in res.text
assert 'hreflang="fa"' in res.text
assert "https://test.example/projects/bilingual-tower" in res.text
assert "https://test.example/en/projects/bilingual-tower" in res.text
# sitemap دو زبانه با alternates
sm = client.get("/sitemap.xml").text
assert "xmlns:xhtml" in sm
assert 'hreflang="en"' in sm
assert "/en/projects/bilingual-tower" in sm
# پوسته فارسی هم alternates دارد
assert 'hreflang="en"' in client.get("/blog/published-post").text
# صفحه استاتیک انگلیسی از مسیر catch-all سرو می‌شود
res = client.get("/en/about")
assert res.status_code == 200
assert 'lang="en"' in res.text
assert auth_client.delete(f"/api/admin/projects/{pid}").status_code == 204
assert "bilingual-tower" not in client.get("/sitemap.xml").text
def test_bilingual_article_en_shell(auth_client, client):
articles = auth_client.get("/api/admin/articles").json()
article = next(a for a in articles if a["slug"] == "published-post")
res = auth_client.patch(
f"/api/admin/articles/{article['id']}",
json={
"title_en": "Published EN",
"excerpt_en": "English excerpt",
"body_en": DOC,
},
)
assert res.status_code == 200, res.text
res = client.get("/en/blog/published-post")
assert res.status_code == 200
assert "Published EN" in res.text
assert "English excerpt" in res.text
assert '"inLanguage": "en"' in res.text
def test_en_backfill_fills_only_empty_fields(auth_client, client):
"""backfill ترجمه‌های خالی را پر می‌کند و مقدار ثبت‌شده را بازنویسی نمی‌کند."""
from app.migrations import backfill_en_content
backfill_en_content()
brand = client.get("/api/site").json()["brand"]
assert brand["name_en"] == "Aramesh Construction Group"
# مقدار دستیِ ادمین محترم شمرده می‌شود
res = auth_client.patch(f"/api/admin/brand/{brand['id']}", json={"name_en": "Custom EN Name"})
assert res.status_code == 200
backfill_en_content()
assert client.get("/api/site").json()["brand"]["name_en"] == "Custom EN Name"
# ---------- محافظت فرم‌ها و پاک‌سازی ----------
def test_contact_honeypot_saves_nothing(client, auth_client):
"""دامنهٔ پنهان پرشده یعنی ربات — پاسخ موفقِ قلابی، بدون ذخیره."""
res = client.post(
"/api/contact",
json={"name": "ربات تستی", "phone": "09120000000", "website": "spam"},
)
assert res.status_code == 201
assert res.json()["id"] == 0 # رکورد قلابی — در دیتابیس نیست
assert auth_client.get("/api/admin/messages").json() == []
def test_contact_rate_limited(client, auth_client):
"""ارسال ششم در پنجرهٔ ۱۵ دقیقه‌ای رد می‌شود (۵ مجاز برای هر IP)."""
for _ in range(5):
res = client.post(
"/api/contact", json={"name": "کاربر تماس", "phone": "09120000000"}
)
assert res.status_code == 201, res.text
res = client.post("/api/contact", json={"name": "کاربر تماس", "phone": "09120000000"})
assert res.status_code == 429
assert "Retry-After" in res.headers
def test_newsletter_honeypot_saves_nothing(client, auth_client):
res = client.post(
"/api/newsletter", json={"email": "bot@example.com", "website": "spam"}
)
assert res.status_code == 201
assert res.json()["id"] == 0
assert auth_client.get("/api/admin/subscribers").json() == []
def test_message_and_subscriber_csv_export(auth_client):
auth_client.post("/api/contact", json={"name": "کاربر تماس", "phone": "09120000000"})
res = auth_client.get("/api/admin/messages/export.csv")
assert res.status_code == 200
assert res.headers["content-type"].startswith("text/csv")
assert "کاربر تماس" in res.text
auth_client.post("/api/newsletter", json={"email": "sale@example.com"})
res = auth_client.get("/api/admin/subscribers/export.csv")
assert res.status_code == 200
assert "sale@example.com" in res.text
def test_article_body_url_sanitization(auth_client):
"""اسکیم‌های خطرناک در href/src سند TipTap در زمان ذخیره حذف می‌شوند."""
res = auth_client.post(
"/api/admin/articles",
json={
"slug": "sanitized-post",
"title": "مقاله پاک‌سازی‌شده",
"status": "published",
"body": {
"type": "doc",
"content": [
{
"type": "paragraph",
"content": [
{
"type": "text",
"marks": [
{"type": "link", "attrs": {"href": "javascript:alert(1)"}},
],
"text": "لینک خطرناک",
}
],
}
],
},
},
)
assert res.status_code == 201, res.text
body = res.json()["body"]
marks = body["content"][0]["content"][0].get("marks", [])
assert all("href" not in m.get("attrs", {}) for m in marks)
# ---------- نام‌گذاری تصاویر، درون‌ریزی و پرسش‌وپاسخ مقاله ----------
class _FakeS3:
def __init__(self, *args, **kwargs):
self.objects = {}
def put_object(self, **kwargs):
self.objects[kwargs["Key"]] = kwargs["Body"]
def _mock_storage(monkeypatch):
"""MinIO در تست در دسترس نیست — کلاینت S3 و DNS عمومی جعل می‌شوند."""
fake = _FakeS3()
monkeypatch.setattr("app.api.admin.boto3.client", lambda *a, **k: fake)
monkeypatch.setattr(
"app.api.admin.socket.getaddrinfo",
lambda host, port, *a, **k: [(socket.AF_INET, None, None, "", ("93.184.216.34", 0))],
)
return fake
def _png_bytes() -> bytes:
from PIL import Image
import io
buf = io.BytesIO()
Image.new("RGB", (8, 8), (200, 120, 60)).save(buf, "PNG")
return buf.getvalue()
def test_upload_with_readable_name(auth_client, monkeypatch):
_mock_storage(monkeypatch)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("raw name.png", _png_bytes(), "image/png")},
data={"name": "برج الهیه tower!"},
)
assert res.status_code == 200, res.text
key = res.json()["key"]
# نام خوانا (فارسی/لاتین، بدون نویسه ناامن) + پسوند یکتا + نرمال‌شدن به WebP
assert "برج-الهیه-tower" in key
assert key.endswith(".webp")
assert res.json()["content_type"] == "image/webp"
assert "!" not in key
def _animated_gif_bytes() -> bytes:
from PIL import Image
import io
buf = io.BytesIO()
first = Image.new("RGB", (12, 8), (200, 120, 60))
second = Image.new("RGB", (12, 8), (60, 120, 200))
first.save(buf, "GIF", save_all=True, append_images=[second], duration=100, loop=0)
return buf.getvalue()
def test_upload_jpeg_converted_to_webp(auth_client, monkeypatch):
_mock_storage(monkeypatch)
from PIL import Image
import io
buf = io.BytesIO()
Image.new("RGB", (3000, 1500), (10, 10, 10)).save(buf, "JPEG", quality=90)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("tower.jpg", buf.getvalue(), "image/jpeg")},
)
assert res.status_code == 200, res.text
body = res.json()
assert body["content_type"] == "image/webp"
assert body["key"].endswith(".webp")
# سقف ضلع ۲۰۰۰ پیکسل روی خروجی اعمال شده
assert body["width"] == 2000 and body["height"] == 1000
def test_upload_animated_gif_becomes_animated_webp(auth_client, monkeypatch):
fake = _mock_storage(monkeypatch)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("anim.gif", _animated_gif_bytes(), "image/gif")},
)
assert res.status_code == 200, res.text
assert res.json()["content_type"] == "image/webp"
from PIL import Image
import io
stored = next(iter(fake.objects.values()))
out = Image.open(io.BytesIO(stored))
assert out.format == "WEBP"
assert getattr(out, "n_frames", 1) == 2 # حرکت از بین نرفته
def test_upload_undecodable_file_rejected(auth_client, monkeypatch):
_mock_storage(monkeypatch)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("notes.txt", b"this is not a media file at all", "text/plain")},
)
assert res.status_code == 415
def test_upload_video_becomes_animated_webp(auth_client, monkeypatch):
# ویدیو تنها با FFmpeg قابل پذیرش است — بدون آن ۴۱۵، با آن WebP متحرک
import shutil as _shutil
import subprocess as _subprocess
import tempfile as _tempfile
import os as _os
_mock_storage(monkeypatch)
if _shutil.which("ffmpeg") is None:
res = auth_client.post(
"/api/admin/upload",
files={"file": ("clip.mp4", b"\x00\x00\x00\x18ftypmp42", "video/mp4")},
)
assert res.status_code == 415
return
with _tempfile.TemporaryDirectory() as tmp:
clip = _os.path.join(tmp, "clip.mp4")
_subprocess.run(
["ffmpeg", "-y", "-v", "error", "-f", "lavfi",
"-i", "testsrc=duration=0.5:size=128x96:rate=10", clip],
check=True, capture_output=True, timeout=30,
)
with open(clip, "rb") as fh:
clip_bytes = fh.read()
res = auth_client.post(
"/api/admin/upload",
files={"file": ("clip.mp4", clip_bytes, "video/mp4")},
)
assert res.status_code == 200, res.text
assert res.json()["content_type"] == "image/webp"
def test_upload_without_name_uses_filename(auth_client, monkeypatch):
_mock_storage(monkeypatch)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("Elahieh Tower.jpg", _png_bytes(), "image/jpeg")},
data={"name": ""},
)
assert res.status_code == 200
assert "Elahieh-Tower" in res.json()["key"]
def test_import_image_rejects_internal_host(auth_client):
# SSRF — میزبان حلقه‌محلی/خصوصی رد می‌شود پیش از هر درخواست شبکه
res = auth_client.post("/api/admin/import-image", json={"url": "http://127.0.0.1:9000/x.png"})
assert res.status_code == 400
res = auth_client.post("/api/admin/import-image", json={"url": "http://192.168.1.5/x.png"})
assert res.status_code == 400
res = auth_client.post("/api/admin/import-image", json={"url": "file:///etc/passwd"})
assert res.status_code == 400
def test_import_image_rejects_bad_content_type(auth_client, monkeypatch):
_mock_storage(monkeypatch)
import io as _io
class FakeResponse:
def __init__(self):
self.headers = {"Content-Type": "text/html"}
def read(self, n):
return b"<html></html>"
def __enter__(self):
return self
def __exit__(self, *args):
return False
def fake_urlopen(request, timeout):
return FakeResponse()
monkeypatch.setattr("app.api.admin.urllib.request.urlopen", fake_urlopen)
res = auth_client.post("/api/admin/import-image", json={"url": "https://example.com/page"})
assert res.status_code == 415
def test_import_image_happy_path(auth_client, monkeypatch):
_mock_storage(monkeypatch)
class FakeResponse:
def __init__(self):
self.headers = {"Content-Type": "image/png"}
def read(self, n):
return _png_bytes()
def __enter__(self):
return self
def __exit__(self, *args):
return False
monkeypatch.setattr("app.api.admin.urllib.request.urlopen", lambda request, timeout: FakeResponse())
res = auth_client.post("/api/admin/import-image", json={"url": "https://wiki.example/Elahieh pic.png?token=1"})
assert res.status_code == 200, res.text
assert res.json()["url"].startswith("/media/")
assert res.json()["key"].endswith(".webp")
assert "Elahieh-pic" in res.json()["key"]
def test_article_persists_image_meta_and_faqs(auth_client, client):
created = auth_client.post(
"/api/admin/articles",
json={"slug": "meta-article", "title": "متادیتا", "body": DOC},
)
assert created.status_code == 201, created.text
article_id = created.json()["id"]
faqs = [{"question": "هزینه چقدر است؟", "answer": "بستگی به پروژه دارد."}]
res = auth_client.patch(
f"/api/admin/articles/{article_id}",
json={
"image_alt": "نمای برج الهیه",
"image_title": "برج آرامش",
"faqs": faqs,
"faqs_en": [{"question": "What is the cost?", "answer": "Depends on the project."}],
},
)
assert res.status_code == 200, res.text
assert res.json()["image_alt"] == "نمای برج الهیه"
assert res.json()["image_title"] == "برج آرامش"
assert res.json()["faqs"] == faqs
# انتشار و سپس خروجی عمومی — همین مقادیر برمی‌گردد
assert auth_client.patch(f"/api/admin/articles/{article_id}", json={"status": "published"}).status_code == 200
public = client.get("/api/articles/meta-article").json()
assert public["image_alt"] == "نمای برج الهیه"
assert public["faqs"] == faqs
assert auth_client.delete(f"/api/admin/articles/{article_id}").status_code == 204
def test_blog_shell_includes_faqpage_jsonld(auth_client, client):
created = auth_client.post(
"/api/admin/articles",
json={
"slug": "faq-article",
"title": "مقاله پرسش‌وپاسخ",
"body": DOC,
"status": "published",
"faqs": [{"question": "زمان ساخت؟", "answer": "۳۰ ماه."}],
},
)
assert created.status_code == 201, created.text
res = client.get("/blog/faq-article")
assert res.status_code == 200
assert "FAQPage" in res.text
assert "زمان ساخت؟" in res.text
# نسخه انگلیسی بدون ترجمه → همان فارسی؛ با ترجمه → انگلیسی
en = client.get("/en/blog/faq-article").text
assert "FAQPage" in en
article_id = created.json()["id"]
auth_client.patch(
f"/api/admin/articles/{article_id}",
json={"faqs_en": [{"question": "How long?", "answer": "Thirty months."}]},
)
en = client.get("/en/blog/faq-article").text
assert "How long?" in en
assert "زمان ساخت؟" not in en
auth_client.delete(f"/api/admin/articles/{article_id}")
def test_blog_shell_without_faqs_has_no_faqpage(client):
res = client.get("/blog/published-post")
assert res.status_code == 200
assert "FAQPage" not in res.text
# ---------- پیش‌نمایش خصوصی (برای انتشار عمومی خاموش می‌شود) ----------
def test_preview_mode_blocks_robots_and_sitemap(client, monkeypatch):
from app.config import settings
monkeypatch.setattr(settings, "preview_mode", True)
try:
robots = client.get("/robots.txt")
assert robots.status_code == 200
assert "Disallow: /" in robots.text
assert "Sitemap:" not in robots.text
sitemap = client.get("/sitemap.xml")
assert sitemap.status_code == 200
assert "published-post" not in sitemap.text
# لایه دوم: هدر noindex روی همه پاسخ‌ها
assert "noindex" in client.get("/api/health").headers.get("X-Robots-Tag", "")
finally:
monkeypatch.setattr(settings, "preview_mode", False)
def test_public_mode_allows_indexing(client):
robots = client.get("/robots.txt").text
assert "Disallow: /admin" in robots
assert "Sitemap:" in robots
assert "noindex" not in client.get("/api/health").headers.get("X-Robots-Tag", "")