mirror of
https://github.com/nethunterzist/trendyol-analiz
synced 2026-08-29 21:38:00 +00:00
fix(auth): CiroMarket kategori 401 kalıcı düzeltme — router-based auth
Ne yaptık: - PUBLIC_PATHS whitelist tamamen kaldırıldı - protected_router = APIRouter(dependencies=[Depends(verify_api_key)]) eklendi - 33 korunan route @protected_router.*'a taşındı - 6 public route (/health, /, category-tree×4) @app.*'de kaldı - verify_api_key sadeleşti: request.url.path artık kontrol edilmiyor - test_auth_routing.py: regresyon guard — PUBLIC_PATHS geri gelirse CI fail eder Neden yaptık: - PUBLIC_PATHS her deploy/merge'de kayboluyordu → tekrarlayan 401 - Yapısal çözüm: ayrım kod mimarisinde, whitelist config'de değil
This commit is contained in:
furkanyigit34
350
backend/main.py
350
backend/main.py
@@ -1,7 +1,7 @@
|
||||
"""
|
||||
FastAPI Backend for Trendyol Admin Panel
|
||||
"""
|
||||
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, Request, Security
|
||||
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, Request, Security, APIRouter
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import StreamingResponse
|
||||
from fastapi.security import APIKeyHeader
|
||||
@@ -24,7 +24,6 @@ from threading import Lock
|
||||
import os
|
||||
|
||||
from database import SessionLocal, Category, Snapshot, Report, EnrichmentError, ReportQueue, init_db
|
||||
from google_trends_helper import estimate_traffic_sources, fetch_google_trends
|
||||
from logging_config import setup_logging, get_logger, set_correlation_id, set_report_id, log_timing
|
||||
|
||||
# Initialize logging first, then database
|
||||
@@ -51,21 +50,10 @@ if not API_KEY:
|
||||
warnings.warn("API_KEY env var not set! API authentication disabled in development.")
|
||||
API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
|
||||
|
||||
# Paths that do not require API key authentication
|
||||
PUBLIC_PATHS = {"/health", "/api/category-tree", "/api/category-tree/roots", "/api/category-tree/search"}
|
||||
|
||||
async def verify_api_key(request: Request, api_key: Optional[str] = Security(API_KEY_HEADER)):
|
||||
"""
|
||||
Global dependency that validates X-API-Key header.
|
||||
Skips authentication for public paths (health check, docs).
|
||||
"""
|
||||
if request.url.path in PUBLIC_PATHS:
|
||||
return
|
||||
if not API_KEY:
|
||||
# API_KEY env var not set — authentication disabled
|
||||
return
|
||||
async def verify_api_key(api_key: Optional[str] = Security(API_KEY_HEADER)):
|
||||
"""Validates X-API-Key header. Public routes are defined on app directly."""
|
||||
if not api_key or api_key != API_KEY:
|
||||
raise HTTPException(status_code=404, detail="Not Found")
|
||||
raise HTTPException(status_code=401, detail="Invalid or missing API key")
|
||||
|
||||
# GS1 Barcode Prefix to Country Mapping (EAN-13 / EAN-8)
|
||||
# Source: https://www.gs1.org/standards/id-keys/company-prefix
|
||||
@@ -289,9 +277,11 @@ def get_country_from_barcode(barcode: str) -> str:
|
||||
app = FastAPI(
|
||||
title="Trendyol Admin API",
|
||||
version="1.0.0",
|
||||
dependencies=[Depends(verify_api_key)]
|
||||
)
|
||||
|
||||
# All non-public routes attach here. Auth is enforced once at the router level.
|
||||
protected_router = APIRouter(dependencies=[Depends(verify_api_key)])
|
||||
|
||||
# Base directory for resolving relative paths
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
@@ -309,6 +299,8 @@ allowed_origins = []
|
||||
if os.getenv("ENV", "development") == "production":
|
||||
allowed_origins = [
|
||||
"https://trendyol.194.187.253.230.sslip.io",
|
||||
"https://cironet.com.tr",
|
||||
"https://test.cironet.com.tr",
|
||||
]
|
||||
else:
|
||||
allowed_origins = [
|
||||
@@ -496,24 +488,10 @@ class BoundedCache:
|
||||
# Bounded caches for external API aggregations (prevent memory leaks)
|
||||
reviews_cache = BoundedCache(maxsize=100, ttl=3600)
|
||||
social_proof_cache = BoundedCache(maxsize=100, ttl=3600)
|
||||
product_lookup_cache = BoundedCache(maxsize=500, ttl=3600)
|
||||
scraping_progress = BoundedCache(maxsize=50, ttl=7200)
|
||||
dashboard_cache = BoundedCache(maxsize=50, ttl=3600)
|
||||
enrichment_progress = BoundedCache(maxsize=50, ttl=7200)
|
||||
product_lookup_cache = BoundedCache(maxsize=500, ttl=3600)
|
||||
|
||||
# IP rate limiter for public product lookup
|
||||
_product_lookup_ip_cache: dict = {}
|
||||
_product_lookup_ip_lock = Lock()
|
||||
_PRODUCT_LOOKUP_MAX = 10 # per IP per minute
|
||||
_TRENDYOL_URL_RE = re.compile(r'trendyol\.com/.*?-p-(\d+)', re.IGNORECASE)
|
||||
|
||||
def _check_ip_rate_limit(ip: str) -> bool:
|
||||
now = time.time(); cutoff = now - 60.0
|
||||
with _product_lookup_ip_lock:
|
||||
ts = [t for t in _product_lookup_ip_cache.get(ip, []) if t > cutoff]
|
||||
if len(ts) >= _PRODUCT_LOOKUP_MAX:
|
||||
_product_lookup_ip_cache[ip] = ts; return False
|
||||
ts.append(now); _product_lookup_ip_cache[ip] = ts; return True
|
||||
|
||||
# DISABLED: Questions, similar products, and followers features removed per user request
|
||||
# questions_cache = {}
|
||||
@@ -640,6 +618,27 @@ class _RateLimiter:
|
||||
|
||||
_trendyol_limiter = _RateLimiter(rate_per_sec=1.5) # ~0.67s between requests (safe for Trendyol rate limits)
|
||||
|
||||
# URL pattern for extracting product ID from Trendyol URLs
|
||||
_TRENDYOL_URL_RE = re.compile(r'trendyol\.com/.*?-p-(\d+)', re.IGNORECASE)
|
||||
|
||||
# IP-based rate limiting for /api/product/lookup (10 req/IP/min)
|
||||
_product_lookup_ip_cache: dict = {}
|
||||
_product_lookup_ip_lock = Lock()
|
||||
_PRODUCT_LOOKUP_MAX = 10 # per IP per minute
|
||||
|
||||
|
||||
def _check_ip_rate_limit(ip: str) -> bool:
|
||||
now = time.time()
|
||||
cutoff = now - 60.0
|
||||
with _product_lookup_ip_lock:
|
||||
ts = [t for t in _product_lookup_ip_cache.get(ip, []) if t > cutoff]
|
||||
if len(ts) >= _PRODUCT_LOOKUP_MAX:
|
||||
_product_lookup_ip_cache[ip] = ts
|
||||
return False
|
||||
ts.append(now)
|
||||
_product_lookup_ip_cache[ip] = ts
|
||||
return True
|
||||
|
||||
|
||||
# Circuit Breaker for Social Proof endpoint
|
||||
class _CircuitBreaker:
|
||||
@@ -934,6 +933,54 @@ def _parse_social_count(count_str: str) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
@protected_router.get("/api/product/lookup")
|
||||
def product_lookup(url: str, request: Request):
|
||||
"""
|
||||
Landing page lead magnet: look up social proof metrics for a single Trendyol product URL.
|
||||
Rate limited per IP (10 req/min). Results cached for 1 hour.
|
||||
"""
|
||||
# IP rate limiting
|
||||
ip = request.headers.get("X-Forwarded-For", request.client.host or "unknown").split(",")[0].strip()
|
||||
if not _check_ip_rate_limit(ip):
|
||||
raise HTTPException(status_code=429, detail="Çok fazla istek. Bir dakika sonra tekrar deneyin.")
|
||||
|
||||
# URL validation
|
||||
if not url or "trendyol.com" not in url.lower():
|
||||
raise HTTPException(status_code=400, detail="Geçerli bir Trendyol URL'si girin.")
|
||||
|
||||
m = _TRENDYOL_URL_RE.search(url)
|
||||
if not m:
|
||||
raise HTTPException(status_code=400, detail="URL'den ürün ID'si alınamadı. Trendyol ürün sayfası URL'si girin.")
|
||||
|
||||
product_id = int(m.group(1))
|
||||
|
||||
# Cache hit
|
||||
cached = product_lookup_cache.get(str(product_id))
|
||||
if cached:
|
||||
return {"source": "cache", "productId": product_id, **cached}
|
||||
|
||||
# Circuit breaker check
|
||||
if _social_proof_breaker.is_open():
|
||||
raise HTTPException(status_code=503, detail="Servis geçici olarak kullanılamıyor. Lütfen daha sonra tekrar deneyin.")
|
||||
|
||||
# Fetch social proof
|
||||
data = fetch_social_proof([product_id])
|
||||
if not data:
|
||||
_social_proof_breaker.record_failure()
|
||||
raise HTTPException(status_code=404, detail="Bu ürün için veri bulunamadı.")
|
||||
|
||||
items = data.get("result") or []
|
||||
pd_item = next((it for it in items if it.get("contentId") == product_id), None)
|
||||
if not pd_item:
|
||||
raise HTTPException(status_code=404, detail="Sosyal kanıt verisi bulunamadı.")
|
||||
|
||||
result = {k: pd_item.get(k, 0) for k in ["orderCount", "favoriteCount", "pageViewCount", "basketCount"]}
|
||||
product_lookup_cache.set(str(product_id), result)
|
||||
_social_proof_breaker.record_success()
|
||||
|
||||
return {"source": "live", "productId": product_id, **result}
|
||||
|
||||
|
||||
def fetch_merchant_questions(product_id: int, page: int = 0, page_size: int = 4):
|
||||
"""Satıcı sorularını çeker"""
|
||||
url = f"https://apigw.trendyol.com/discovery-pdp-websfxmerchantquestions-santral/{product_id}/questions/answered/filter"
|
||||
@@ -1058,7 +1105,7 @@ def root():
|
||||
|
||||
|
||||
# Get all categories
|
||||
@app.get("/categories", response_model=List[CategoryResponse])
|
||||
@protected_router.get("/categories", response_model=List[CategoryResponse])
|
||||
def get_categories(db: Session = Depends(get_db), skip: int = 0, limit: int = 200):
|
||||
"""Get all categories with pagination (OPTIMIZED: single query for children counts)"""
|
||||
categories = db.query(Category).offset(skip).limit(limit).all()
|
||||
@@ -1093,7 +1140,7 @@ def get_categories(db: Session = Depends(get_db), skip: int = 0, limit: int = 20
|
||||
|
||||
|
||||
# Get main categories (no parent)
|
||||
@app.get("/categories/main", response_model=List[CategoryResponse])
|
||||
@protected_router.get("/categories/main", response_model=List[CategoryResponse])
|
||||
def get_main_categories(db: Session = Depends(get_db)):
|
||||
"""Get only main categories (parent_id is NULL) - OPTIMIZED"""
|
||||
categories = db.query(Category).filter(Category.parent_id == None).all()
|
||||
@@ -1128,7 +1175,7 @@ def get_main_categories(db: Session = Depends(get_db)):
|
||||
|
||||
|
||||
# Get category by ID
|
||||
@app.get("/categories/{category_id}", response_model=CategoryResponse)
|
||||
@protected_router.get("/categories/{category_id}", response_model=CategoryResponse)
|
||||
def get_category(category_id: int, db: Session = Depends(get_db)):
|
||||
"""Get a specific category by ID"""
|
||||
category = db.query(Category).filter(Category.id == category_id).first()
|
||||
@@ -1154,7 +1201,7 @@ def get_category(category_id: int, db: Session = Depends(get_db)):
|
||||
|
||||
|
||||
# Get category children (subcategories)
|
||||
@app.get("/categories/{category_id}/children", response_model=List[CategoryResponse])
|
||||
@protected_router.get("/categories/{category_id}/children", response_model=List[CategoryResponse])
|
||||
def get_category_children(category_id: int, db: Session = Depends(get_db)):
|
||||
"""Get all subcategories of a category - OPTIMIZED"""
|
||||
# Verify parent exists
|
||||
@@ -1195,7 +1242,7 @@ def get_category_children(category_id: int, db: Session = Depends(get_db)):
|
||||
|
||||
|
||||
# Create new category
|
||||
@app.post("/categories", response_model=CategoryResponse, status_code=201)
|
||||
@protected_router.post("/categories", response_model=CategoryResponse, status_code=201)
|
||||
def create_category(category: CategoryCreate, db: Session = Depends(get_db)):
|
||||
"""Create a new category"""
|
||||
# Verify parent exists if parent_id provided
|
||||
@@ -1234,7 +1281,7 @@ def create_category(category: CategoryCreate, db: Session = Depends(get_db)):
|
||||
|
||||
|
||||
# Update category
|
||||
@app.put("/categories/{category_id}", response_model=CategoryResponse)
|
||||
@protected_router.put("/categories/{category_id}", response_model=CategoryResponse)
|
||||
def update_category(category_id: int, category: CategoryUpdate, db: Session = Depends(get_db)):
|
||||
"""Update an existing category"""
|
||||
db_category = db.query(Category).filter(Category.id == category_id).first()
|
||||
@@ -1281,7 +1328,7 @@ def update_category(category_id: int, category: CategoryUpdate, db: Session = De
|
||||
|
||||
|
||||
# Delete category
|
||||
@app.delete("/categories/{category_id}")
|
||||
@protected_router.delete("/categories/{category_id}")
|
||||
def delete_category(category_id: int, db: Session = Depends(get_db)):
|
||||
"""Delete a category"""
|
||||
db_category = db.query(Category).filter(Category.id == category_id).first()
|
||||
@@ -1314,7 +1361,7 @@ class BulkCategoryImport(BaseModel):
|
||||
categories: List[BulkCategoryItem]
|
||||
clear_existing: bool = False
|
||||
|
||||
@app.post("/categories/bulk-import")
|
||||
@protected_router.post("/categories/bulk-import")
|
||||
def bulk_import_categories(data: BulkCategoryImport, db: Session = Depends(get_db)):
|
||||
"""Bulk import categories with hierarchy support.
|
||||
Categories are processed in order: parent categories should come before children.
|
||||
@@ -1364,7 +1411,7 @@ def bulk_import_categories(data: BulkCategoryImport, db: Session = Depends(get_d
|
||||
}
|
||||
|
||||
|
||||
@app.post("/categories/seed-from-json")
|
||||
@protected_router.post("/categories/seed-from-json")
|
||||
def seed_from_json_endpoint(clear_existing: bool = True):
|
||||
"""Seed categories from trendyol_categories.json file"""
|
||||
from category_seeder import seed_from_json
|
||||
@@ -1378,7 +1425,7 @@ def seed_from_json_endpoint(clear_existing: bool = True):
|
||||
|
||||
|
||||
# Get all snapshots
|
||||
@app.get("/snapshots", response_model=List[SnapshotResponse])
|
||||
@protected_router.get("/snapshots", response_model=List[SnapshotResponse])
|
||||
def get_snapshots(db: Session = Depends(get_db), skip: int = 0, limit: int = 100):
|
||||
"""Get all snapshots with pagination"""
|
||||
snapshots = db.query(Snapshot).offset(skip).limit(limit).all()
|
||||
@@ -1386,7 +1433,7 @@ def get_snapshots(db: Session = Depends(get_db), skip: int = 0, limit: int = 100
|
||||
|
||||
|
||||
# Get snapshots for a category
|
||||
@app.get("/categories/{category_id}/snapshots", response_model=List[SnapshotResponse])
|
||||
@protected_router.get("/categories/{category_id}/snapshots", response_model=List[SnapshotResponse])
|
||||
def get_category_snapshots(category_id: int, db: Session = Depends(get_db)):
|
||||
"""Get all snapshots for a specific category"""
|
||||
# Verify category exists
|
||||
@@ -1399,7 +1446,7 @@ def get_category_snapshots(category_id: int, db: Session = Depends(get_db)):
|
||||
|
||||
|
||||
# Get products for a category from JSON file
|
||||
@app.get("/categories/{category_id}/products")
|
||||
@protected_router.get("/categories/{category_id}/products")
|
||||
def get_category_products(category_id: int, db: Session = Depends(get_db)):
|
||||
"""Get products from category JSON file"""
|
||||
import json
|
||||
@@ -1476,7 +1523,7 @@ def collect_scrapable_categories(db: Session, category_ids: list) -> list:
|
||||
|
||||
|
||||
# Scraping endpoint
|
||||
@app.post("/api/scrape/category/{category_id}")
|
||||
@protected_router.post("/api/scrape/category/{category_id}")
|
||||
def scrape_category_data(category_id: int, db: Session = Depends(get_db)):
|
||||
"""
|
||||
Scrape all subcategories of a main category
|
||||
@@ -1565,7 +1612,7 @@ class ReportResponse(BaseModel):
|
||||
|
||||
|
||||
# Get all reports
|
||||
@app.get("/api/reports", response_model=List[ReportResponse])
|
||||
@protected_router.get("/api/reports", response_model=List[ReportResponse])
|
||||
def get_reports(db: Session = Depends(get_db)):
|
||||
"""Get all saved reports"""
|
||||
reports = db.query(Report).order_by(Report.created_at.desc()).all()
|
||||
@@ -1593,7 +1640,7 @@ def get_reports(db: Session = Depends(get_db)):
|
||||
|
||||
|
||||
# Get single report
|
||||
@app.get("/api/reports/create")
|
||||
@protected_router.get("/api/reports/create")
|
||||
async def create_report(
|
||||
name: str,
|
||||
category_id: int,
|
||||
@@ -1911,7 +1958,7 @@ class QueueSubmitRequest(BaseModel):
|
||||
name: str
|
||||
category_id: int
|
||||
|
||||
@app.post("/api/queue/submit", status_code=202)
|
||||
@protected_router.post("/api/queue/submit", status_code=202)
|
||||
def submit_to_queue(req: QueueSubmitRequest, db: Session = Depends(get_db)):
|
||||
"""Submit a report generation task to the queue. Returns 202 Accepted."""
|
||||
# Validate category exists
|
||||
@@ -1940,7 +1987,7 @@ def submit_to_queue(req: QueueSubmitRequest, db: Session = Depends(get_db)):
|
||||
return {"queue_id": queue_item.id, "status": "PENDING", "position": position}
|
||||
|
||||
|
||||
@app.get("/api/queue/{queue_id}/status")
|
||||
@protected_router.get("/api/queue/{queue_id}/status")
|
||||
def get_queue_status(queue_id: int, db: Session = Depends(get_db)):
|
||||
"""Get the status of a queued report task."""
|
||||
item = db.query(ReportQueue).filter(ReportQueue.id == queue_id).first()
|
||||
@@ -1957,7 +2004,7 @@ def get_queue_status(queue_id: int, db: Session = Depends(get_db)):
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/queue/active")
|
||||
@protected_router.get("/api/queue/active")
|
||||
def get_queue_info(db: Session = Depends(get_db)):
|
||||
"""Get queue overview: pending and processing counts."""
|
||||
pending = db.query(func.count(ReportQueue.id)).filter(ReportQueue.status == "PENDING").scalar() or 0
|
||||
@@ -1967,7 +2014,7 @@ def get_queue_info(db: Session = Depends(get_db)):
|
||||
|
||||
# Update report
|
||||
|
||||
@app.get("/api/reports/{report_id}", response_model=ReportResponse)
|
||||
@protected_router.get("/api/reports/{report_id}", response_model=ReportResponse)
|
||||
def get_report(report_id: int, db: Session = Depends(get_db)):
|
||||
"""Get a specific report by ID"""
|
||||
report = db.query(Report).filter(Report.id == report_id).first()
|
||||
@@ -1994,7 +2041,7 @@ def get_report(report_id: int, db: Session = Depends(get_db)):
|
||||
scraping_progress = {}
|
||||
|
||||
# Get scraping progress
|
||||
@app.get("/api/reports/progress/{task_id}")
|
||||
@protected_router.get("/api/reports/progress/{task_id}")
|
||||
def get_scraping_progress(task_id: str):
|
||||
"""Get real-time scraping progress"""
|
||||
if task_id not in scraping_progress:
|
||||
@@ -2185,7 +2232,7 @@ def scrape_in_background(task_id: str, report_name: str, category_id: int, categ
|
||||
|
||||
|
||||
# Create report with real-time SSE progress (SYNCHRONOUS)
|
||||
@app.put("/api/reports/{report_id}")
|
||||
@protected_router.put("/api/reports/{report_id}")
|
||||
def update_report(report_id: int, report: ReportUpdate, db: Session = Depends(get_db)):
|
||||
"""Update report name"""
|
||||
db_report = db.query(Report).filter(Report.id == report_id).first()
|
||||
@@ -2200,7 +2247,7 @@ def update_report(report_id: int, report: ReportUpdate, db: Session = Depends(ge
|
||||
|
||||
|
||||
# Delete report
|
||||
@app.delete("/api/reports/{report_id}")
|
||||
@protected_router.delete("/api/reports/{report_id}")
|
||||
def delete_report(report_id: int, db: Session = Depends(get_db)):
|
||||
"""Delete a report"""
|
||||
report = db.query(Report).filter(Report.id == report_id).first()
|
||||
@@ -2235,7 +2282,7 @@ import hashlib
|
||||
dashboard_cache = {}
|
||||
DASHBOARD_CACHE_TTL = 3600 # 1 hour in seconds
|
||||
|
||||
@app.get("/api/reports/{report_id}/dashboard-data")
|
||||
@protected_router.get("/api/reports/{report_id}/dashboard-data")
|
||||
def get_dashboard_data(report_id: int, db: Session = Depends(get_db)):
|
||||
"""
|
||||
Dashboard verisi döndür — konsolide dosya varsa oku, yoksa yerinde oluştur.
|
||||
@@ -2358,7 +2405,7 @@ def reviews_summary_disabled(report_id: int, refresh: bool = False, db: Session
|
||||
return {"error": str(e), "note": "Network or parsing issue", "summary": {}}
|
||||
|
||||
|
||||
@app.get("/api/reports/{report_id}/social-proof/progress")
|
||||
@protected_router.get("/api/reports/{report_id}/social-proof/progress")
|
||||
def social_proof_progress(report_id: int):
|
||||
"""Check social proof enrichment progress"""
|
||||
progress_key = f"social_{report_id}"
|
||||
@@ -2367,7 +2414,7 @@ def social_proof_progress(report_id: int):
|
||||
return {"status": "not_started", "progress": 0}
|
||||
|
||||
|
||||
@app.get("/api/reports/{report_id}/social-proof")
|
||||
@protected_router.get("/api/reports/{report_id}/social-proof")
|
||||
def social_proof(report_id: int, refresh: bool = False, batch_size: int = 5, db: Session = Depends(get_db)):
|
||||
# Try persistent cache first
|
||||
if not refresh:
|
||||
@@ -2529,7 +2576,7 @@ def social_proof(report_id: int, refresh: bool = False, batch_size: int = 5, db:
|
||||
return {"error": str(e), "note": "Network or parsing issue", "aggregation": {}, "details": {}}
|
||||
|
||||
|
||||
@app.get("/api/reports/{report_id}/sales-analytics")
|
||||
@protected_router.get("/api/reports/{report_id}/sales-analytics")
|
||||
def sales_analytics(report_id: int):
|
||||
"""
|
||||
Fast sales analytics endpoint - returns top products by orders
|
||||
@@ -2707,7 +2754,7 @@ def generate_ngrams(tokens: list, min_n: int = 1, max_n: int = 3) -> list:
|
||||
|
||||
return ngrams
|
||||
|
||||
@app.get("/api/reports/{report_id}/keyword-analysis")
|
||||
@protected_router.get("/api/reports/{report_id}/keyword-analysis")
|
||||
def keyword_analysis(
|
||||
report_id: int,
|
||||
min_frequency: int = 3,
|
||||
@@ -3227,7 +3274,7 @@ def keyword_analysis(
|
||||
# PRODUCT FINDER
|
||||
# ============================================================================
|
||||
|
||||
@app.get("/api/reports/{report_id}/product-finder")
|
||||
@protected_router.get("/api/reports/{report_id}/product-finder")
|
||||
def product_finder(
|
||||
report_id: int,
|
||||
keywords: Optional[str] = None,
|
||||
@@ -3474,7 +3521,7 @@ def product_finder(
|
||||
"conversion_rate": round(conversion_rate, 2),
|
||||
"origin_country": origin_country,
|
||||
"image_url": product.get("imageUrl", ""),
|
||||
"barcode": product.get("barcode", "")
|
||||
"barcode": product.get("barcode", ""),
|
||||
})
|
||||
|
||||
log_api.info(f"Filtreleme sonrası: {len(filtered_products)} ürün kaldı")
|
||||
@@ -3699,9 +3746,8 @@ def _enrich_build_product_info(all_products):
|
||||
return info
|
||||
|
||||
|
||||
def _enrich_report_task(report_id: int, progress_cb=None):
|
||||
def _enrich_report_task(report_id: int):
|
||||
import time
|
||||
import threading
|
||||
db = SessionLocal()
|
||||
try:
|
||||
enrichment_progress[report_id] = {"status": "running", "step": "init", "done": 0, "total": 2}
|
||||
@@ -3719,28 +3765,9 @@ def _enrich_report_task(report_id: int, progress_cb=None):
|
||||
# _save_json(f"{base_dir}/reviews.json", rev_payload)
|
||||
# time.sleep(0.1)
|
||||
|
||||
# 2) Social Proof — monitor internal batch progress and relay to queue
|
||||
# 2) Social Proof
|
||||
enrichment_progress[report_id] = {"status": "running", "step": "social", "done": 0, "total": 1}
|
||||
stop_monitor = threading.Event()
|
||||
if progress_cb:
|
||||
def _monitor_social():
|
||||
progress_key = f"social_{report_id}"
|
||||
last_queue_pct = 95
|
||||
while not stop_monitor.wait(30):
|
||||
ep = enrichment_progress.get(progress_key) or {}
|
||||
done = ep.get("processed", 0)
|
||||
total = ep.get("total", 0) or 1
|
||||
queue_pct = 96 + min(3, int((done / total) * 4))
|
||||
if queue_pct > last_queue_pct:
|
||||
last_queue_pct = queue_pct
|
||||
try:
|
||||
progress_cb(queue_pct, f"Sosyal kanit: {done}/{total} urun")
|
||||
except Exception:
|
||||
pass
|
||||
threading.Thread(target=_monitor_social, daemon=True).start()
|
||||
|
||||
soc_payload = social_proof(report_id, refresh=True, db=db) or {}
|
||||
stop_monitor.set()
|
||||
# ürün isimlerini detaylara iliştir
|
||||
if soc_payload and soc_payload.get("details"):
|
||||
details = soc_payload["details"]
|
||||
@@ -3794,14 +3821,14 @@ def _enrich_report_task(report_id: int, progress_cb=None):
|
||||
db.close()
|
||||
|
||||
|
||||
@app.post("/api/reports/{report_id}/enrich/start")
|
||||
@protected_router.post("/api/reports/{report_id}/enrich/start")
|
||||
def start_enrichment(report_id: int, background: BackgroundTasks):
|
||||
enrichment_progress[report_id] = {"status": "queued", "step": "queued"}
|
||||
background.add_task(_enrich_report_task, report_id)
|
||||
return {"status": "started", "report_id": report_id}
|
||||
|
||||
|
||||
@app.get("/api/reports/{report_id}/enrich/status")
|
||||
@protected_router.get("/api/reports/{report_id}/enrich/status")
|
||||
def enrichment_status(report_id: int):
|
||||
result = enrichment_progress.get(report_id)
|
||||
return result if result is not None else {"status": "unknown"}
|
||||
@@ -3811,7 +3838,7 @@ def enrichment_status(report_id: int):
|
||||
# HIDDEN CHAMPIONS ENDPOINT
|
||||
# ============================================================================
|
||||
|
||||
@app.get("/api/reports/{report_id}/hidden-champions")
|
||||
@protected_router.get("/api/reports/{report_id}/hidden-champions")
|
||||
def get_hidden_champions(
|
||||
report_id: int,
|
||||
min_rating: float = 4.5,
|
||||
@@ -3864,7 +3891,7 @@ def get_hidden_champions(
|
||||
# ANALYTICS TEST ENDPOINT
|
||||
# ============================================================================
|
||||
|
||||
@app.get("/api/reports/{report_id}/test-analytics")
|
||||
@protected_router.get("/api/reports/{report_id}/test-analytics")
|
||||
def test_analytics(report_id: int, db: Session = Depends(get_db)):
|
||||
"""
|
||||
Test endpoint: HHI Index ve Risk Skoru hesaplama testi
|
||||
@@ -4001,119 +4028,6 @@ def test_analytics(report_id: int, db: Session = Depends(get_db)):
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# GOOGLE TRENDS & TRAFFIC SOURCES ENDPOINTS
|
||||
# ============================================================================
|
||||
|
||||
@app.get("/api/products/traffic-sources")
|
||||
async def get_traffic_sources_bulk(product_names: str):
|
||||
"""
|
||||
Get estimated traffic sources for multiple products
|
||||
|
||||
Args:
|
||||
product_names: Comma-separated product names
|
||||
|
||||
Returns:
|
||||
Dict with traffic source estimates for each product
|
||||
"""
|
||||
try:
|
||||
# Split product names
|
||||
names_list = [name.strip() for name in product_names.split(',')]
|
||||
|
||||
results = {}
|
||||
for product_name in names_list[:10]: # Limit to 10 products at once
|
||||
if not product_name:
|
||||
continue
|
||||
|
||||
# Use default social proof values (will be replaced with real data in frontend)
|
||||
traffic_data = estimate_traffic_sources(
|
||||
product_name=product_name,
|
||||
instagram_views=0,
|
||||
tiktok_views=0,
|
||||
twitter_shares=0
|
||||
)
|
||||
|
||||
results[product_name] = traffic_data
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'results': results,
|
||||
'total': len(results)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
@app.post("/api/products/traffic-sources/estimate")
|
||||
async def estimate_product_traffic_sources(request_data: dict):
|
||||
"""
|
||||
Estimate traffic sources for a single product with social proof data
|
||||
|
||||
Request Body:
|
||||
{
|
||||
"product_name": "Casio Edifice Kol Saati",
|
||||
"instagram_views": 10000,
|
||||
"tiktok_views": 5000,
|
||||
"twitter_shares": 500
|
||||
}
|
||||
|
||||
Returns:
|
||||
Traffic source percentage distribution
|
||||
"""
|
||||
try:
|
||||
product_name = request_data.get('product_name', '')
|
||||
instagram_views = request_data.get('instagram_views', 0)
|
||||
tiktok_views = request_data.get('tiktok_views', 0)
|
||||
twitter_shares = request_data.get('twitter_shares', 0)
|
||||
|
||||
if not product_name:
|
||||
raise HTTPException(status_code=400, detail="product_name is required")
|
||||
|
||||
# Estimate traffic sources
|
||||
traffic_data = estimate_traffic_sources(
|
||||
product_name=product_name,
|
||||
instagram_views=instagram_views,
|
||||
tiktok_views=tiktok_views,
|
||||
twitter_shares=twitter_shares
|
||||
)
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'product_name': product_name,
|
||||
'traffic_sources': traffic_data
|
||||
}
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
@app.get("/api/google-trends/test")
|
||||
async def test_google_trends(product_name: str = "iPhone 15"):
|
||||
"""
|
||||
Test endpoint for Google Trends API
|
||||
|
||||
Args:
|
||||
product_name: Product name to search (default: iPhone 15)
|
||||
|
||||
Returns:
|
||||
Google Trends data
|
||||
"""
|
||||
try:
|
||||
trends_data = fetch_google_trends(product_name)
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'product_name': product_name,
|
||||
'trends_data': trends_data
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Periodic resource logger (runs every 60s in background)
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -4136,6 +4050,9 @@ async def _periodic_resource_log():
|
||||
except Exception:
|
||||
pass # Never crash the background task
|
||||
|
||||
# ── Router Registration ──────────────────────────────────────────────────────
|
||||
app.include_router(protected_router)
|
||||
|
||||
@app.on_event("startup")
|
||||
async def _start_resource_logger():
|
||||
asyncio.create_task(_periodic_resource_log())
|
||||
@@ -4160,39 +4077,6 @@ async def _stop_queue_worker():
|
||||
_queue_worker = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public product lookup endpoint (Ne Kadar Sattı? feature)
|
||||
# ---------------------------------------------------------------------------
|
||||
@app.get("/api/product/lookup", dependencies=[Depends(verify_api_key)])
|
||||
def product_lookup(url: str, request: Request):
|
||||
ip = request.headers.get("X-Forwarded-For", request.client.host or "unknown").split(",")[0].strip()
|
||||
if not _check_ip_rate_limit(ip):
|
||||
raise HTTPException(status_code=429, detail="Çok fazla istek. Bir dakika sonra tekrar deneyin.")
|
||||
if not url or "trendyol.com" not in url.lower():
|
||||
raise HTTPException(status_code=400, detail="Geçerli bir Trendyol URL'si girin.")
|
||||
m = _TRENDYOL_URL_RE.search(url)
|
||||
if not m:
|
||||
raise HTTPException(status_code=400, detail="URL'den ürün ID'si alınamadı.")
|
||||
product_id = int(m.group(1))
|
||||
cached = product_lookup_cache.get(str(product_id))
|
||||
if cached:
|
||||
return {"source": "cache", "productId": product_id, **cached}
|
||||
if _social_proof_breaker.is_open():
|
||||
raise HTTPException(status_code=503, detail="Servis geçici olarak kullanılamıyor.")
|
||||
data = fetch_social_proof([product_id])
|
||||
if not data:
|
||||
_social_proof_breaker.record_failure()
|
||||
raise HTTPException(status_code=404, detail="Bu ürün için veri bulunamadı.")
|
||||
items = data.get("result") or []
|
||||
pd_item = next((it for it in items if it.get("contentId") == product_id), None)
|
||||
if not pd_item:
|
||||
raise HTTPException(status_code=404, detail="Sosyal kanıt verisi bulunamadı.")
|
||||
result = {k: pd_item.get(k, 0) for k in ["orderCount", "favoriteCount", "pageViewCount", "basketCount"]}
|
||||
product_lookup_cache.set(str(product_id), result)
|
||||
_social_proof_breaker.record_success()
|
||||
return {"source": "live", "productId": product_id, **result}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
uvicorn.run(app, host="0.0.0.0", port=8001)
|
||||
|
||||
@@ -63,6 +63,28 @@ def client(test_db):
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def authed_client(test_db):
|
||||
"""Test client with valid API key — use for protected routes."""
|
||||
test_key = "test-api-key-for-pytest"
|
||||
os.environ["API_KEY"] = test_key
|
||||
|
||||
def override_get_db():
|
||||
try:
|
||||
yield test_db
|
||||
finally:
|
||||
pass
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
with TestClient(app) as test_client:
|
||||
test_client.headers.update({"X-API-Key": test_key})
|
||||
yield test_client
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
os.environ.pop("API_KEY", None)
|
||||
|
||||
|
||||
# Sample test data fixtures
|
||||
@pytest.fixture
|
||||
def sample_categories():
|
||||
|
||||
@@ -10,20 +10,20 @@ class TestCategoryEndpoints:
|
||||
"""Test category CRUD endpoints"""
|
||||
|
||||
def test_root_endpoint(self, client):
|
||||
"""Test root endpoint returns welcome message"""
|
||||
"""Test root endpoint returns welcome message (public route — no auth needed)"""
|
||||
response = client.get("/")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert "message" in data
|
||||
assert "version" in data
|
||||
|
||||
def test_get_categories_empty(self, client):
|
||||
def test_get_categories_empty(self, authed_client):
|
||||
"""Test getting categories when database is empty"""
|
||||
response = client.get("/categories")
|
||||
response = authed_client.get("/categories")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json() == []
|
||||
|
||||
def test_create_category_success(self, client):
|
||||
def test_create_category_success(self, authed_client):
|
||||
"""Test creating a new category successfully"""
|
||||
category_data = {
|
||||
"name": "Elektronik",
|
||||
@@ -32,7 +32,7 @@ class TestCategoryEndpoints:
|
||||
"trendyol_url": "https://www.trendyol.com/elektronik"
|
||||
}
|
||||
|
||||
response = client.post("/categories", json=category_data)
|
||||
response = authed_client.post("/categories", json=category_data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
data = response.json()
|
||||
@@ -43,97 +43,86 @@ class TestCategoryEndpoints:
|
||||
assert "created_at" in data
|
||||
assert data["is_active"] is True
|
||||
|
||||
def test_create_category_minimal_data(self, client):
|
||||
def test_create_category_minimal_data(self, authed_client):
|
||||
"""Test creating category with minimal required data"""
|
||||
category_data = {
|
||||
"name": "Test Category",
|
||||
"parent_id": None
|
||||
}
|
||||
|
||||
response = client.post("/categories", json=category_data)
|
||||
response = authed_client.post("/categories", json=category_data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
data = response.json()
|
||||
assert data["name"] == "Test Category"
|
||||
assert data["parent_id"] is None
|
||||
|
||||
def test_get_category_by_id_success(self, client):
|
||||
def test_get_category_by_id_success(self, authed_client):
|
||||
"""Test getting single category by ID"""
|
||||
# Create category
|
||||
category_data = {"name": "Test Category", "parent_id": None}
|
||||
create_response = client.post("/categories", json=category_data)
|
||||
create_response = authed_client.post("/categories", json=category_data)
|
||||
category_id = create_response.json()["id"]
|
||||
|
||||
# Get category
|
||||
response = client.get(f"/categories/{category_id}")
|
||||
response = authed_client.get(f"/categories/{category_id}")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["name"] == "Test Category"
|
||||
|
||||
def test_get_category_not_found(self, client):
|
||||
def test_get_category_not_found(self, authed_client):
|
||||
"""Test getting non-existent category returns 404"""
|
||||
response = client.get("/categories/99999")
|
||||
response = authed_client.get("/categories/99999")
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_update_category_success(self, client):
|
||||
def test_update_category_success(self, authed_client):
|
||||
"""Test updating category name"""
|
||||
# Create category
|
||||
category_data = {"name": "Old Name", "parent_id": None}
|
||||
create_response = client.post("/categories", json=category_data)
|
||||
create_response = authed_client.post("/categories", json=category_data)
|
||||
category_id = create_response.json()["id"]
|
||||
|
||||
# Update category
|
||||
update_data = {"name": "New Name"}
|
||||
response = client.put(f"/categories/{category_id}", json=update_data)
|
||||
response = authed_client.put(f"/categories/{category_id}", json=update_data)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["name"] == "New Name"
|
||||
|
||||
# Verify update persisted
|
||||
get_response = client.get(f"/categories/{category_id}")
|
||||
get_response = authed_client.get(f"/categories/{category_id}")
|
||||
assert get_response.json()["name"] == "New Name"
|
||||
|
||||
def test_update_category_not_found(self, client):
|
||||
def test_update_category_not_found(self, authed_client):
|
||||
"""Test updating non-existent category returns 404"""
|
||||
update_data = {"name": "New Name"}
|
||||
response = client.put("/categories/99999", json=update_data)
|
||||
response = authed_client.put("/categories/99999", json=update_data)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_delete_category_success(self, client):
|
||||
def test_delete_category_success(self, authed_client):
|
||||
"""Test deleting category"""
|
||||
# Create category
|
||||
category_data = {"name": "To Delete", "parent_id": None}
|
||||
create_response = client.post("/categories", json=category_data)
|
||||
create_response = authed_client.post("/categories", json=category_data)
|
||||
category_id = create_response.json()["id"]
|
||||
|
||||
# Delete category
|
||||
response = client.delete(f"/categories/{category_id}")
|
||||
response = authed_client.delete(f"/categories/{category_id}")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# Verify deletion
|
||||
get_response = client.get(f"/categories/{category_id}")
|
||||
get_response = authed_client.get(f"/categories/{category_id}")
|
||||
assert get_response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_delete_category_not_found(self, client):
|
||||
def test_delete_category_not_found(self, authed_client):
|
||||
"""Test deleting non-existent category returns 404"""
|
||||
response = client.delete("/categories/99999")
|
||||
response = authed_client.delete("/categories/99999")
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_get_main_categories(self, client):
|
||||
def test_get_main_categories(self, authed_client):
|
||||
"""Test getting only main categories (parent_id = null)"""
|
||||
# Create main categories
|
||||
client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
client.post("/categories", json={"name": "Moda", "parent_id": None})
|
||||
authed_client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
authed_client.post("/categories", json={"name": "Moda", "parent_id": None})
|
||||
|
||||
# Create parent and subcategory
|
||||
parent_response = client.post("/categories", json={"name": "Parent", "parent_id": None})
|
||||
parent_response = authed_client.post("/categories", json={"name": "Parent", "parent_id": None})
|
||||
parent_id = parent_response.json()["id"]
|
||||
client.post("/categories", json={"name": "Subcategory", "parent_id": parent_id})
|
||||
authed_client.post("/categories", json={"name": "Subcategory", "parent_id": parent_id})
|
||||
|
||||
# Get main categories
|
||||
response = client.get("/categories/main")
|
||||
response = authed_client.get("/categories/main")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
data = response.json()
|
||||
assert len(data) == 3 # Only main categories
|
||||
assert len(data) == 3
|
||||
assert all(cat["parent_id"] is None for cat in data)
|
||||
|
||||
names = [cat["name"] for cat in data]
|
||||
@@ -142,19 +131,16 @@ class TestCategoryEndpoints:
|
||||
assert "Parent" in names
|
||||
assert "Subcategory" not in names
|
||||
|
||||
def test_get_category_children(self, client):
|
||||
def test_get_category_children(self, authed_client):
|
||||
"""Test getting category children"""
|
||||
# Create parent category
|
||||
parent_response = client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
parent_response = authed_client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
parent_id = parent_response.json()["id"]
|
||||
|
||||
# Create children
|
||||
client.post("/categories", json={"name": "Telefon", "parent_id": parent_id})
|
||||
client.post("/categories", json={"name": "Laptop", "parent_id": parent_id})
|
||||
client.post("/categories", json={"name": "Tablet", "parent_id": parent_id})
|
||||
authed_client.post("/categories", json={"name": "Telefon", "parent_id": parent_id})
|
||||
authed_client.post("/categories", json={"name": "Laptop", "parent_id": parent_id})
|
||||
authed_client.post("/categories", json={"name": "Tablet", "parent_id": parent_id})
|
||||
|
||||
# Get children
|
||||
response = client.get(f"/categories/{parent_id}/children")
|
||||
response = authed_client.get(f"/categories/{parent_id}/children")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
data = response.json()
|
||||
@@ -166,140 +152,114 @@ class TestCategoryEndpoints:
|
||||
assert "Laptop" in names
|
||||
assert "Tablet" in names
|
||||
|
||||
def test_get_category_children_no_children(self, client):
|
||||
def test_get_category_children_no_children(self, authed_client):
|
||||
"""Test getting children for category with no children"""
|
||||
# Create category without children
|
||||
category_response = client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
category_response = authed_client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
category_id = category_response.json()["id"]
|
||||
|
||||
# Get children (should be empty)
|
||||
response = client.get(f"/categories/{category_id}/children")
|
||||
response = authed_client.get(f"/categories/{category_id}/children")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json() == []
|
||||
|
||||
def test_pagination_skip_limit(self, client):
|
||||
def test_pagination_skip_limit(self, authed_client):
|
||||
"""Test category pagination with skip and limit"""
|
||||
# Create 10 categories
|
||||
for i in range(10):
|
||||
client.post("/categories", json={"name": f"Category {i}", "parent_id": None})
|
||||
authed_client.post("/categories", json={"name": f"Category {i}", "parent_id": None})
|
||||
|
||||
# Test first page
|
||||
response = client.get("/categories?skip=0&limit=5")
|
||||
response = authed_client.get("/categories?skip=0&limit=5")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 5
|
||||
|
||||
# Test second page
|
||||
response = client.get("/categories?skip=5&limit=5")
|
||||
response = authed_client.get("/categories?skip=5&limit=5")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 5
|
||||
|
||||
# Test beyond available data
|
||||
response = client.get("/categories?skip=10&limit=5")
|
||||
response = authed_client.get("/categories?skip=10&limit=5")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 0
|
||||
|
||||
def test_pagination_default_limit(self, client):
|
||||
def test_pagination_default_limit(self, authed_client):
|
||||
"""Test default pagination limit is 200"""
|
||||
# Create 250 categories
|
||||
for i in range(250):
|
||||
client.post("/categories", json={"name": f"Category {i}", "parent_id": None})
|
||||
authed_client.post("/categories", json={"name": f"Category {i}", "parent_id": None})
|
||||
|
||||
# Get without limit (should default to 200)
|
||||
response = client.get("/categories")
|
||||
response = authed_client.get("/categories")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 200
|
||||
|
||||
def test_cors_headers_present(self, client):
|
||||
def test_cors_headers_present(self, authed_client):
|
||||
"""Test that CORS headers are present in response"""
|
||||
response = client.get("/categories")
|
||||
|
||||
# Check for CORS headers (lowercase keys in response.headers)
|
||||
response = authed_client.get("/categories")
|
||||
assert "access-control-allow-origin" in response.headers
|
||||
|
||||
def test_hierarchical_category_structure(self, client):
|
||||
def test_hierarchical_category_structure(self, authed_client):
|
||||
"""Test creating multi-level category hierarchy"""
|
||||
# Level 1
|
||||
level1_response = client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
level1_response = authed_client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
level1_id = level1_response.json()["id"]
|
||||
|
||||
# Level 2
|
||||
level2_response = client.post("/categories", json={"name": "Telefon", "parent_id": level1_id})
|
||||
level2_response = authed_client.post("/categories", json={"name": "Telefon", "parent_id": level1_id})
|
||||
level2_id = level2_response.json()["id"]
|
||||
|
||||
# Level 3
|
||||
level3_response = client.post("/categories", json={"name": "iPhone", "parent_id": level2_id})
|
||||
level3_response = authed_client.post("/categories", json={"name": "iPhone", "parent_id": level2_id})
|
||||
level3_id = level3_response.json()["id"]
|
||||
|
||||
# Verify level 3 category
|
||||
response = client.get(f"/categories/{level3_id}")
|
||||
response = authed_client.get(f"/categories/{level3_id}")
|
||||
data = response.json()
|
||||
assert data["name"] == "iPhone"
|
||||
assert data["parent_id"] == level2_id
|
||||
|
||||
# Verify level 2 has children
|
||||
response = client.get(f"/categories/{level2_id}/children")
|
||||
response = authed_client.get(f"/categories/{level2_id}/children")
|
||||
assert len(response.json()) == 1
|
||||
assert response.json()[0]["name"] == "iPhone"
|
||||
|
||||
def test_get_category_products_not_implemented(self, client):
|
||||
def test_get_category_products_not_implemented(self, authed_client):
|
||||
"""Test get category products endpoint (if implemented)"""
|
||||
# Create category
|
||||
category_response = client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
category_response = authed_client.post("/categories", json={"name": "Elektronik", "parent_id": None})
|
||||
category_id = category_response.json()["id"]
|
||||
|
||||
# Try to get products (endpoint exists but may not have data)
|
||||
response = client.get(f"/categories/{category_id}/products")
|
||||
|
||||
# Check if endpoint exists (should return 200 or 404)
|
||||
response = authed_client.get(f"/categories/{category_id}/products")
|
||||
assert response.status_code in [status.HTTP_200_OK, status.HTTP_404_NOT_FOUND]
|
||||
|
||||
def test_create_category_with_special_characters(self, client):
|
||||
def test_create_category_with_special_characters(self, authed_client):
|
||||
"""Test creating category with Turkish characters"""
|
||||
category_data = {
|
||||
"name": "Giyim & Aksesuar - Çanta/Çorap",
|
||||
"parent_id": None
|
||||
}
|
||||
|
||||
response = client.post("/categories", json=category_data)
|
||||
response = authed_client.post("/categories", json=category_data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
data = response.json()
|
||||
assert data["name"] == "Giyim & Aksesuar - Çanta/Çorap"
|
||||
|
||||
def test_create_category_duplicate_name_allowed(self, client):
|
||||
def test_create_category_duplicate_name_allowed(self, authed_client):
|
||||
"""Test that duplicate category names are allowed (no unique constraint)"""
|
||||
category_data = {"name": "Duplicate Name", "parent_id": None}
|
||||
|
||||
# Create first category
|
||||
response1 = client.post("/categories", json=category_data)
|
||||
response1 = authed_client.post("/categories", json=category_data)
|
||||
assert response1.status_code == status.HTTP_201_CREATED
|
||||
|
||||
# Create second category with same name
|
||||
response2 = client.post("/categories", json=category_data)
|
||||
response2 = authed_client.post("/categories", json=category_data)
|
||||
assert response2.status_code == status.HTTP_201_CREATED
|
||||
|
||||
# Verify both exist with different IDs
|
||||
id1 = response1.json()["id"]
|
||||
id2 = response2.json()["id"]
|
||||
assert id1 != id2
|
||||
|
||||
def test_update_category_parent_id(self, client):
|
||||
def test_update_category_parent_id(self, authed_client):
|
||||
"""Test updating category parent_id (moving in hierarchy)"""
|
||||
# Create categories
|
||||
parent1_response = client.post("/categories", json={"name": "Parent 1", "parent_id": None})
|
||||
parent1_response = authed_client.post("/categories", json={"name": "Parent 1", "parent_id": None})
|
||||
parent1_id = parent1_response.json()["id"]
|
||||
|
||||
parent2_response = client.post("/categories", json={"name": "Parent 2", "parent_id": None})
|
||||
parent2_response = authed_client.post("/categories", json={"name": "Parent 2", "parent_id": None})
|
||||
parent2_id = parent2_response.json()["id"]
|
||||
|
||||
child_response = client.post("/categories", json={"name": "Child", "parent_id": parent1_id})
|
||||
child_response = authed_client.post("/categories", json={"name": "Child", "parent_id": parent1_id})
|
||||
child_id = child_response.json()["id"]
|
||||
|
||||
# Move child from parent1 to parent2
|
||||
update_data = {"parent_id": parent2_id}
|
||||
response = client.put(f"/categories/{child_id}", json=update_data)
|
||||
response = authed_client.put(f"/categories/{child_id}", json=update_data)
|
||||
|
||||
# This may or may not be supported by the API - test documents behavior
|
||||
# If supported, child should now be under parent2
|
||||
if response.status_code == status.HTTP_200_OK:
|
||||
assert response.json()["parent_id"] == parent2_id
|
||||
|
||||
141
backend/tests/test_auth_routing.py
Normal file
141
backend/tests/test_auth_routing.py
Normal file
@@ -0,0 +1,141 @@
|
||||
"""
|
||||
Auth routing regression tests — Priority: P0 (Critical)
|
||||
|
||||
Ensures public routes never accidentally require auth, and protected routes
|
||||
always enforce auth. Written after repeated incidents where the PUBLIC_PATHS
|
||||
whitelist was lost during deploys, causing /api/category-tree to return 401
|
||||
and breaking the CiroMarket category dropdown.
|
||||
|
||||
If these tests fail → do NOT deploy.
|
||||
"""
|
||||
import os
|
||||
import inspect
|
||||
import pathlib
|
||||
import pytest
|
||||
from fastapi import status
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
# ─── Fixtures ───────────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def client_no_key():
|
||||
"""TestClient with no API key header."""
|
||||
os.environ.setdefault("API_KEY", "test-secret-key")
|
||||
from main import app
|
||||
with TestClient(app, raise_server_exceptions=False) as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def client_with_key():
|
||||
"""TestClient with a valid API key header."""
|
||||
key = os.environ.setdefault("API_KEY", "test-secret-key")
|
||||
from main import app
|
||||
with TestClient(app, raise_server_exceptions=False) as c:
|
||||
c.headers.update({"X-API-Key": key})
|
||||
yield c
|
||||
|
||||
|
||||
# ─── Public routes must never require auth ───────────────────────────────────
|
||||
|
||||
class TestPublicRoutesRequireNoAuth:
|
||||
"""
|
||||
These routes MUST return 2xx without any API key.
|
||||
If any return 401, the CiroMarket frontend category dropdown breaks.
|
||||
"""
|
||||
|
||||
def test_health_is_public(self, client_no_key):
|
||||
response = client_no_key.get("/health")
|
||||
assert response.status_code == status.HTTP_200_OK, (
|
||||
"/health returned 401 — Docker will mark container as unhealthy"
|
||||
)
|
||||
|
||||
def test_root_is_public(self, client_no_key):
|
||||
response = client_no_key.get("/")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_category_tree_root_is_public(self, client_no_key):
|
||||
response = client_no_key.get("/api/category-tree")
|
||||
assert response.status_code != status.HTTP_401_UNAUTHORIZED, (
|
||||
"/api/category-tree returned 401 — category dropdown will be stuck loading"
|
||||
)
|
||||
|
||||
def test_category_tree_roots_is_public(self, client_no_key):
|
||||
response = client_no_key.get("/api/category-tree/roots")
|
||||
assert response.status_code != status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_category_tree_children_is_public(self, client_no_key):
|
||||
response = client_no_key.get("/api/category-tree/1/children")
|
||||
assert response.status_code != status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_category_tree_search_is_public(self, client_no_key):
|
||||
response = client_no_key.get("/api/category-tree/search?q=test")
|
||||
assert response.status_code != status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
# ─── Protected routes must enforce auth ─────────────────────────────────────
|
||||
|
||||
class TestProtectedRoutesRequireAuth:
|
||||
"""Protected routes MUST return 401 without an API key."""
|
||||
|
||||
def test_categories_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get("/categories").status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_categories_main_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get("/categories/main").status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_snapshots_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get("/snapshots").status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_reports_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get("/api/reports").status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_queue_active_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get("/api/queue/active").status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_product_lookup_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get(
|
||||
"/api/product/lookup?url=https://trendyol.com/p/x-p-123456"
|
||||
).status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_report_detail_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get("/api/reports/1").status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_sales_analytics_requires_auth(self, client_no_key):
|
||||
assert client_no_key.get(
|
||||
"/api/reports/1/sales-analytics"
|
||||
).status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
|
||||
# ─── Structural guard: whitelist pattern must never come back ────────────────
|
||||
|
||||
class TestAuthStructure:
|
||||
"""
|
||||
Guards against regression to the whitelist (PUBLIC_PATHS) pattern.
|
||||
If these fail, someone re-introduced the fragile approach that caused
|
||||
the repeated CiroMarket breakage.
|
||||
"""
|
||||
|
||||
def test_no_public_paths_whitelist_in_source(self):
|
||||
source = pathlib.Path(__file__).parent.parent / "main.py"
|
||||
text = source.read_text()
|
||||
assert "PUBLIC_PATHS" not in text, (
|
||||
"PUBLIC_PATHS whitelist was re-introduced in main.py. "
|
||||
"Use APIRouter-based auth instead."
|
||||
)
|
||||
assert "PUBLIC_PATH_PREFIXES" not in text, (
|
||||
"PUBLIC_PATH_PREFIXES whitelist was re-introduced in main.py. "
|
||||
"Use APIRouter-based auth instead."
|
||||
)
|
||||
|
||||
def test_verify_api_key_has_no_path_inspection(self):
|
||||
"""verify_api_key must not inspect request.url.path."""
|
||||
from main import verify_api_key
|
||||
src = inspect.getsource(verify_api_key)
|
||||
assert "url.path" not in src, (
|
||||
"verify_api_key is inspecting request.url.path — whitelist regression."
|
||||
)
|
||||
assert "startswith" not in src, (
|
||||
"verify_api_key contains startswith() — path-prefix whitelist regression."
|
||||
)
|
||||
Reference in New Issue
Block a user