Security: deep /api/health (DB+MinIO probes, 503 on failure) + /api/live; rate-limited public forms with honeypots; real client IP behind Traefik (--proxy-headers); TipTap URL sanitization on write (nh3); non-root containers + slim .dockerignore; prod MinIO console bound to 127.0.0.1; HSTS/nosniff on TLS routers; JWT 8h; no default credentials in code. Reliability: Alembic migrations (create_all baseline); graceful shutdown (exec uvicorn); hardened backups (checksums, sanity check, freshness gauge, optional rclone off-site) + restore.sh + hourly db-backup sidecar (ops profile); stack.sh ops tooling (deploy/logs/health/backup/restore/monitoring); .gitattributes LF for shell scripts. Orchestration: json-file log rotation, pinned images, per-service memory limits, edge/internal networks, Traefik ping/accesslog/metrics + flood limiter, conditional minio-setup, pgvector db image, one-shot db-init (umami + aramesh_chatbot databases). Chatbot: unified from site-chatbot branch (RAG + pgvector, SSE, ChatWidget). Monitoring: Prometheus/Alertmanager/Loki/Promtail/Grafana + cAdvisor, node/postgres/blackbox exporters, provisioned dashboards and 10 alert rules; alerts delivered to Bale (tapi.bale.ai — Telegram-compatible). Analytics: Umami on shared Postgres, build-time tracker injection, custom events (contact/newsletter/phone/chat/lang) via frontend/src/analytics.ts. Hygiene: CSV export for contact messages and subscribers; upload image optimization (Pillow); CI script + GitHub-compatible workflow; audit.sh; gen-env.sh for strong credentials.
364 lines
14 KiB
Python
364 lines
14 KiB
Python
"""تستهای API — مسیرهای عمومی، احراز هویت، CRUD مدیریت، پیشنویس مقاله و سئو."""
|
|
|
|
import re
|
|
|
|
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)
|