diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9cc3c4e..54a6279 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,7 +7,6 @@ on: branches: [main] jobs: - # ── 后端 ────────────────────────────────────────────── backend-lint-test: runs-on: ubuntu-latest services: @@ -63,7 +62,39 @@ jobs: cd backend pytest tests/ -v --tb=short - # ── 前端 ────────────────────────────────────────────── + - name: Security scan (bandit) + run: | + pip install bandit + cd backend + bandit -r app/ -f json -o bandit-report.json || true + bandit -r app/ -f txt || true + continue-on-error: true + + - name: Upload bandit report + if: always() + uses: actions/upload-artifact@v4 + with: + name: bandit-report + path: backend/bandit-report.json + retention-days: 7 + + - name: Performance baseline + env: + DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/geo_test + REDIS_URL: redis://localhost:6379/0 + JWT_SECRET: test-secret-key-minimum-32-characters-long + ENVIRONMENT: test + run: | + cd backend + START_TIME=$(date +%s) + pytest tests/ -v --tb=short -q 2>&1 | tee test-output.txt + END_TIME=$(date +%s) + DURATION=$((END_TIME - START_TIME)) + echo "Backend test duration: ${DURATION}s" >> $GITHUB_STEP_SUMMARY + echo "## Performance Baseline" >> $GITHUB_STEP_SUMMARY + echo "- Backend test suite duration: **${DURATION}s**" >> $GITHUB_STEP_SUMMARY + continue-on-error: true + frontend-lint-test: runs-on: ubuntu-latest steps: @@ -87,7 +118,10 @@ jobs: - name: Unit tests run: cd frontend && npm run test:ci - # ── Docker build 验证 ────────────────────────────────── + - name: Security audit (npm audit) + run: cd frontend && npm audit --audit-level=moderate || true + continue-on-error: true + docker-build: runs-on: ubuntu-latest needs: [backend-lint-test, frontend-lint-test] @@ -99,3 +133,95 @@ jobs: - name: Build frontend image run: docker build -t geo-frontend:test ./frontend + + e2e-test: + runs-on: ubuntu-latest + needs: [backend-lint-test, frontend-lint-test] + steps: + - uses: actions/checkout@v4 + + - name: Create .env for Docker Compose + run: | + cat > .env << 'EOF' + POSTGRES_PASSWORD=geo_e2e_test_2026 + REDIS_PASSWORD=geo_redis_e2e_2026 + JWT_SECRET=e2e-test-secret-key-minimum-32-characters-long + ENVIRONMENT=test + DEEPSEEK_API_KEY=${{ secrets.DEEPSEEK_API_KEY || '' }} + EOF + + - name: Start Docker Compose services + run: | + docker compose up -d db redis + echo "Waiting for database..." + until docker compose exec -T db pg_isready -U postgres -d geo_platform; do + sleep 2 + done + echo "Database is ready" + + - name: Run database migrations + run: | + docker compose up -d backend + echo "Waiting for backend to start..." + for i in $(seq 1 30); do + if curl -sf http://localhost:8000/health > /dev/null 2>&1; then + echo "Backend is healthy" + break + fi + echo "Waiting for backend... ($i/30)" + sleep 5 + done + + - name: Start frontend service + run: | + docker compose up -d frontend + echo "Waiting for frontend to start..." + for i in $(seq 1 30); do + if curl -sf http://localhost:3000 > /dev/null 2>&1; then + echo "Frontend is healthy" + break + fi + echo "Waiting for frontend... ($i/30)" + sleep 5 + done + + - uses: actions/setup-node@v4 + with: + node-version: '20' + cache: 'npm' + cache-dependency-path: frontend/package-lock.json + + - name: Install Playwright browsers + run: | + cd frontend + npm ci + npx playwright install --with-deps chromium + + - name: Seed test data + run: | + curl -sf -X POST http://localhost:8000/api/v1/auth/register \ + -H "Content-Type: application/json" \ + -d '{"email":"admin@example.com","password":"admin@123","name":"E2E Test Admin"}' || true + + - name: Run E2E tests + run: | + cd frontend + E2E_TEST_EMAIL=admin@example.com \ + E2E_TEST_PASSWORD=admin@123 \ + npx playwright test --project=chromium --reporter=html + env: + PLAYWRIGHT_BASE_URL: http://localhost:3000 + + - name: Upload E2E test results + if: failure() + uses: actions/upload-artifact@v4 + with: + name: e2e-test-results + path: | + frontend/playwright-report/ + frontend/test-results/ + retention-days: 7 + + - name: Stop Docker Compose + if: always() + run: docker compose down -v diff --git a/backend/app/config.py b/backend/app/config.py index ff2a1d4..56c9341 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -98,6 +98,8 @@ class Settings(BaseSettings): DISTRIBUTION_MODE: str = "mock" + ATTRIBUTION_WINDOW_DAYS: int = 28 + @field_validator("JWT_SECRET") @classmethod def validate_jwt_secret(cls, v: str) -> str: diff --git a/backend/app/services/attribution/attribution_engine.py b/backend/app/services/attribution/attribution_engine.py index 94b0b38..732ee88 100644 --- a/backend/app/services/attribution/attribution_engine.py +++ b/backend/app/services/attribution/attribution_engine.py @@ -5,6 +5,7 @@ from datetime import UTC, datetime, timedelta from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.config import settings from app.models.attribution_record import AttributionRecord from app.models.diagnosis_record import DiagnosisRecord @@ -12,6 +13,9 @@ logger = logging.getLogger(__name__) class AttributionEngine: + def __init__(self): + self.window_days = settings.ATTRIBUTION_WINDOW_DAYS + async def start_tracking( self, db: AsyncSession, @@ -27,8 +31,9 @@ class AttributionEngine: brand_id=brand_id, content_id=content_id, baseline_score=baseline_score, + attribution_window_days=self.window_days, published_at=now, - window_end_at=now + timedelta(days=28), + window_end_at=now + timedelta(days=self.window_days), status="tracking", ) db.add(record) diff --git a/backend/app/services/content/rule_validator.py b/backend/app/services/content/rule_validator.py index 009e0d2..bb25593 100644 --- a/backend/app/services/content/rule_validator.py +++ b/backend/app/services/content/rule_validator.py @@ -290,7 +290,7 @@ class RuleValidator: elif platform in ("baijiahao", "toutiao"): # 标题党检测 - found_clickbait = clickbait_words & set(title) + found_clickbait = {w for w in clickbait_words if w in title} if found_clickbait: issues.append(ValidationIssue( "high", diff --git a/backend/app/services/distribution/platform_rules.py b/backend/app/services/distribution/platform_rules.py index 627bad7..b199c78 100644 --- a/backend/app/services/distribution/platform_rules.py +++ b/backend/app/services/distribution/platform_rules.py @@ -1155,7 +1155,7 @@ class PlatformRuleEngine: elif platform in ("baijiahao", "toutiao"): # 标题党检测 - found_clickbait = _CLICKBAIT_WORDS & set(title) + found_clickbait = {w for w in _CLICKBAIT_WORDS if w in title} if found_clickbait: issues.append({ "severity": "high", diff --git a/backend/app/services/payment/alipay.py b/backend/app/services/payment/alipay.py index 0083d66..3103517 100644 --- a/backend/app/services/payment/alipay.py +++ b/backend/app/services/payment/alipay.py @@ -1,5 +1,10 @@ +import json import logging +import urllib.parse import uuid +from datetime import datetime + +import requests from app.config import settings from app.services.payment.base import PaymentGateway, PaymentOrder, PaymentCallback @@ -7,6 +12,8 @@ from app.services.payment.mock_gateway import MockGateway logger = logging.getLogger(__name__) +ALIPAY_GATEWAY_URL = "https://openapi.alipay.com/gateway.do" + class AlipayGateway(PaymentGateway): def __init__(self): @@ -14,14 +21,92 @@ class AlipayGateway(PaymentGateway): self.private_key_path = settings.ALIPAY_PRIVATE_KEY_PATH self.public_key_path = settings.ALIPAY_PUBLIC_KEY_PATH self.notify_url = settings.ALIPAY_NOTIFY_URL + self._private_key = None + self._alipay_public_key = None def _is_configured(self) -> bool: return bool(self.app_id and self.private_key_path and self.public_key_path) + def _load_private_key(self): + if self._private_key is not None: + return self._private_key + try: + with open(self.private_key_path, "r") as f: + key_data = f.read() + from cryptography.hazmat.primitives.serialization import load_pem_private_key + from cryptography.hazmat.backends import default_backend + self._private_key = load_pem_private_key( + key_data.encode("utf-8"), password=None, backend=default_backend() + ) + return self._private_key + except Exception as e: + logger.error("[Alipay] 加载私钥失败: %s", e) + return None + + def _load_alipay_public_key(self): + if self._alipay_public_key is not None: + return self._alipay_public_key + try: + with open(self.public_key_path, "r") as f: + key_data = f.read() + from cryptography.hazmat.primitives.serialization import load_pem_public_key + from cryptography.hazmat.backends import default_backend + self._alipay_public_key = load_pem_public_key( + key_data.encode("utf-8"), backend=default_backend() + ) + return self._alipay_public_key + except Exception as e: + logger.error("[Alipay] 加载支付宝公钥失败: %s", e) + return None + def _build_sign_content(self, params: dict) -> str: sorted_params = sorted(params.items(), key=lambda x: x[0]) return "&".join(f"{k}={v}" for k, v in sorted_params if v) + def _rsa2_sign(self, sign_content: str) -> str: + private_key = self._load_private_key() + if private_key is None: + raise RuntimeError("Alipay private key not loaded") + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import padding + signature = private_key.sign( + sign_content.encode("utf-8"), + padding.PKCS1v15(), + hashes.SHA256(), + ) + import base64 + return base64.b64encode(signature).decode("utf-8") + + def _rsa2_verify(self, sign_content: str, sign: str) -> bool: + public_key = self._load_alipay_public_key() + if public_key is None: + logger.warning("[Alipay] 支付宝公钥未加载，跳过签名验证") + return False + import base64 + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.asymmetric import padding + try: + public_key.verify( + base64.b64decode(sign), + sign_content.encode("utf-8"), + padding.PKCS1v15(), + hashes.SHA256(), + ) + return True + except Exception: + return False + + def _build_common_params(self, method: str) -> dict: + return { + "app_id": self.app_id, + "method": method, + "charset": "utf-8", + "sign_type": "RSA2", + "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "version": "1.0", + "notify_url": self.notify_url, + } + async def create_order( self, order_id: str, amount: float, description: str, user_id: str, plan: str ) -> PaymentOrder: @@ -29,32 +114,43 @@ class AlipayGateway(PaymentGateway): logger.info("[Alipay] 未配置应用信息，降级为Mock支付") return await MockGateway().create_order(order_id, amount, description, user_id, plan) - biz_content = { + biz_content = json.dumps({ "out_trade_no": order_id, "total_amount": str(amount), "subject": description, "product_code": "QUICK_WAP_WAY", - } + }) - params = { - "app_id": self.app_id, - "method": "alipay.trade.wap.pay", - "charset": "utf-8", - "sign_type": "RSA2", - "timestamp": __import__("datetime").datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "version": "1.0", - "notify_url": self.notify_url, - "biz_content": str(biz_content), - } + params = self._build_common_params("alipay.trade.wap.pay") + params["biz_content"] = biz_content - logger.info(f"[Alipay] 创建WAP支付订单: order_id={order_id}, amount={amount}") + sign_content = self._build_sign_content(params) + params["sign"] = self._rsa2_sign(sign_content) - return PaymentOrder( - order_id=order_id, - pay_url=f"https://openapi.alipay.com/gateway.do?out_trade_no={order_id}", - amount=amount, - status="pending", - ) + try: + resp = requests.post( + ALIPAY_GATEWAY_URL, + data=params, + timeout=10, + ) + resp.raise_for_status() + + if resp.headers.get("content-type", "").startswith("text/html"): + pay_url = f"{ALIPAY_GATEWAY_URL}?{urllib.parse.urlencode(params)}" + else: + data = resp.json() + resp_data = data.get("alipay_trade_wap_pay_response", {}) + pay_url = resp_data.get("pay_url", f"{ALIPAY_GATEWAY_URL}?{urllib.parse.urlencode(params)}") + + return PaymentOrder( + order_id=order_id, + pay_url=pay_url, + amount=amount, + status="pending", + ) + except requests.RequestException as e: + logger.error("[Alipay] 创建订单失败: %s", e) + raise RuntimeError(f"Alipay create_order failed: {e}") from e async def verify_callback(self, request_data: dict) -> PaymentCallback: if not self._is_configured(): @@ -66,14 +162,22 @@ class AlipayGateway(PaymentGateway): params = {k: v for k, v in request_data.items() if k not in ("sign", "sign_type") and v} sign_content = self._build_sign_content(params) - logger.info(f"[Alipay] 验证回调签名: order_id={request_data.get('out_trade_no')}") + if not self._rsa2_verify(sign_content, sign): + logger.warning("[Alipay] 回调签名验证失败: order_id=%s", request_data.get("out_trade_no")) + return PaymentCallback( + order_id=request_data.get("out_trade_no", ""), + payment_id=request_data.get("trade_no", ""), + amount=float(request_data.get("total_amount", 0)), + status="failed", + raw_data=request_data, + ) trade_status = request_data.get("trade_status", "TRADE_CLOSED") return PaymentCallback( order_id=request_data.get("out_trade_no", ""), payment_id=request_data.get("trade_no", ""), amount=float(request_data.get("total_amount", 0)), - status="success" if trade_status == "TRADE_SUCCESS" else "failed", + status="success" if trade_status in ("TRADE_SUCCESS", "TRADE_FINISHED") else "failed", raw_data=request_data, ) @@ -81,17 +185,59 @@ class AlipayGateway(PaymentGateway): if not self._is_configured(): return await MockGateway().query_order(order_id) - logger.info(f"[Alipay] 查询订单: order_id={order_id}") - return PaymentOrder( - order_id=order_id, - pay_url="", - amount=0, - status="pending", - ) + biz_content = json.dumps({"out_trade_no": order_id}) + params = self._build_common_params("alipay.trade.query") + params["biz_content"] = biz_content + + sign_content = self._build_sign_content(params) + params["sign"] = self._rsa2_sign(sign_content) + + try: + resp = requests.post(ALIPAY_GATEWAY_URL, data=params, timeout=10) + resp.raise_for_status() + data = resp.json() + resp_data = data.get("alipay_trade_query_response", {}) + trade_status = resp_data.get("trade_status", "WAIT_BUYER_PAY") + total_amount = float(resp_data.get("total_amount", 0)) + + status_map = { + "TRADE_SUCCESS": "completed", + "TRADE_FINISHED": "completed", + "WAIT_BUYER_PAY": "pending", + "TRADE_CLOSED": "closed", + } + return PaymentOrder( + order_id=order_id, + pay_url="", + amount=total_amount, + status=status_map.get(trade_status, "pending"), + ) + except requests.RequestException as e: + logger.error("[Alipay] 查询订单失败: %s", e) + raise RuntimeError(f"Alipay query_order failed: {e}") from e async def refund(self, order_id: str, amount: float, reason: str = "") -> bool: if not self._is_configured(): return await MockGateway().refund(order_id, amount, reason) - logger.info(f"[Alipay] 申请退款: order_id={order_id}, amount={amount}") - return True + biz_content = json.dumps({ + "out_trade_no": order_id, + "refund_amount": str(amount), + "refund_reason": reason or "用户申请退款", + "out_request_no": f"R_{order_id}_{int(datetime.now().timestamp())}", + }) + params = self._build_common_params("alipay.trade.refund") + params["biz_content"] = biz_content + + sign_content = self._build_sign_content(params) + params["sign"] = self._rsa2_sign(sign_content) + + try: + resp = requests.post(ALIPAY_GATEWAY_URL, data=params, timeout=10) + resp.raise_for_status() + data = resp.json() + resp_data = data.get("alipay_trade_refund_response", {}) + return resp_data.get("code") == "10000" + except requests.RequestException as e: + logger.error("[Alipay] 退款失败: %s", e) + return False diff --git a/backend/app/services/payment/wechat_pay.py b/backend/app/services/payment/wechat_pay.py index 66400bb..b3a2169 100644 --- a/backend/app/services/payment/wechat_pay.py +++ b/backend/app/services/payment/wechat_pay.py @@ -1,14 +1,21 @@ +import base64 import hashlib +import hmac +import json import logging import time import uuid +import requests + from app.config import settings from app.services.payment.base import PaymentGateway, PaymentOrder, PaymentCallback from app.services.payment.mock_gateway import MockGateway logger = logging.getLogger(__name__) +WECHAT_PAY_BASE_URL = "https://api.mch.weixin.qq.com" + class WeChatPayGateway(PaymentGateway): def __init__(self): @@ -17,6 +24,7 @@ class WeChatPayGateway(PaymentGateway): self.app_id = settings.WECHAT_PAY_APP_ID self.cert_path = settings.WECHAT_PAY_CERT_PATH self.notify_url = settings.WECHAT_PAY_NOTIFY_URL + self._serial_no = "" def _is_configured(self) -> bool: return bool(self.mch_id and self.api_key and self.app_id) @@ -26,10 +34,43 @@ class WeChatPayGateway(PaymentGateway): return MockGateway() return self - def _generate_sign(self, params: dict) -> str: - sorted_params = sorted(params.items(), key=lambda x: x[0]) - sign_str = "&".join(f"{k}={v}" for k, v in sorted_params if v) + f"&key={self.api_key}" - return hashlib.md5(sign_str.encode()).hexdigest().upper() + def _generate_nonce(self) -> str: + return uuid.uuid4().hex[:32] + + def _build_auth_header(self, method: str, url_path: str, body: str = "") -> str: + timestamp = str(int(time.time())) + nonce_str = self._generate_nonce() + sign_message = f"{self.mch_id}\n{timestamp}\n{nonce_str}\n{body}\n" + signature = hmac.new( + self.api_key.encode("utf-8"), + sign_message.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return ( + f'WECHATPAY2-SHA256-RSA2048 ' + f'mchid="{self.mch_id}",' + f'nonce_str="{nonce_str}",' + f'timestamp="{timestamp}",' + f'serial_no="{self._serial_no}",' + f'signature="{signature}"' + ) + + def _build_signature(self, method: str, url_path: str, timestamp: str, nonce_str: str, body: str = "") -> str: + sign_message = f"{method}\n{url_path}\n{timestamp}\n{nonce_str}\n{body}\n" + return hmac.new( + self.api_key.encode("utf-8"), + sign_message.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + + def _verify_v3_signature(self, timestamp: str, nonce: str, body: str, signature: str) -> bool: + sign_message = f"{timestamp}\n{nonce}\n{body}\n" + expected = hmac.new( + self.api_key.encode("utf-8"), + sign_message.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return hmac.compare_digest(expected, signature) async def create_order( self, order_id: str, amount: float, description: str, user_id: str, plan: str @@ -38,70 +79,202 @@ class WeChatPayGateway(PaymentGateway): logger.info("[WeChatPay] 未配置商户信息，降级为Mock支付") return await MockGateway().create_order(order_id, amount, description, user_id, plan) - params = { + url_path = "/v3/pay/transactions/native" + url = f"{WECHAT_PAY_BASE_URL}{url_path}" + body = json.dumps({ "appid": self.app_id, - "mch_id": self.mch_id, - "nonce_str": uuid.uuid4().hex[:32], - "body": description, + "mchid": self.mch_id, + "description": description, "out_trade_no": order_id, - "total_fee": str(int(amount * 100)), - "spbill_create_ip": "127.0.0.1", "notify_url": self.notify_url, - "trade_type": "NATIVE", - } - params["sign"] = self._generate_sign(params) + "amount": { + "total": int(amount * 100), + "currency": "CNY", + }, + }) - logger.info(f"[WeChatPay] 创建Native支付订单: order_id={order_id}, amount={amount}") - - return PaymentOrder( - order_id=order_id, - pay_url=f"weixin://wxpay/bizpayurl?pr={order_id}", - amount=amount, - status="pending", + timestamp = str(int(time.time())) + nonce_str = self._generate_nonce() + signature = self._build_signature("POST", url_path, timestamp, nonce_str, body) + auth_header = ( + f'WECHATPAY2-SHA256-RSA2048 ' + f'mchid="{self.mch_id}",' + f'nonce_str="{nonce_str}",' + f'timestamp="{timestamp}",' + f'derial_no="{self._serial_no}",' + f'signature="{signature}"' ) + headers = { + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": auth_header, + } + + try: + resp = requests.post(url, data=body.encode("utf-8"), headers=headers, timeout=10) + resp.raise_for_status() + data = resp.json() + code_url = data.get("code_url", "") + return PaymentOrder( + order_id=order_id, + pay_url=code_url, + amount=amount, + status="pending", + ) + except requests.RequestException as e: + logger.error("[WeChatPay] 创建订单失败: %s", e) + raise RuntimeError(f"WeChat Pay create_order failed: {e}") from e + async def verify_callback(self, request_data: dict) -> PaymentCallback: if not self._is_configured(): return await MockGateway().verify_callback(request_data) - received_sign = request_data.get("sign", "") - params = {k: v for k, v in request_data.items() if k != "sign" and v} - expected_sign = self._generate_sign(params) + timestamp = request_data.get("timestamp", "") + nonce = request_data.get("nonce", "") + body = request_data.get("body", "") + signature = request_data.get("signature", "") - if received_sign != expected_sign: - logger.warning(f"[WeChatPay] 回调签名验证失败: order_id={request_data.get('out_trade_no')}") + if not self._verify_v3_signature(timestamp, nonce, body, signature): + logger.warning("[WeChatPay] V3回调签名验证失败") return PaymentCallback( - order_id=request_data.get("out_trade_no", ""), - payment_id=request_data.get("transaction_id", ""), - amount=float(request_data.get("total_fee", 0)) / 100, + order_id="", + payment_id="", + amount=0, status="failed", raw_data=request_data, ) - result_code = request_data.get("result_code", "FAIL") - return PaymentCallback( - order_id=request_data.get("out_trade_no", ""), - payment_id=request_data.get("transaction_id", ""), - amount=float(request_data.get("total_fee", 0)) / 100, - status="success" if result_code == "SUCCESS" else "failed", - raw_data=request_data, - ) + try: + resource = json.loads(body).get("resource", {}) if body.startswith("{") else {} + if not resource and "resource" in request_data: + resource = request_data["resource"] + + ciphertext = resource.get("ciphertext", "") + nonce_val = resource.get("nonce", "") + associated_data = resource.get("associated_data", "") + + decrypted = self._decrypt_resource(ciphertext, nonce_val, associated_data) + order_id = decrypted.get("out_trade_no", "") + transaction_id = decrypted.get("transaction_id", "") + total = decrypted.get("amount", {}).get("total", 0) + trade_state = decrypted.get("trade_state", "") + + return PaymentCallback( + order_id=order_id, + payment_id=transaction_id, + amount=total / 100, + status="success" if trade_state == "SUCCESS" else "failed", + raw_data=request_data, + ) + except Exception as e: + logger.error("[WeChatPay] 回调解密失败: %s", e) + return PaymentCallback( + order_id="", + payment_id="", + amount=0, + status="failed", + raw_data=request_data, + ) + + def _decrypt_resource(self, ciphertext_b64: str, nonce: str, associated_data: str) -> dict: + key = self.api_key.encode("utf-8") + ct = base64.b64decode(ciphertext_b64) + tag = ct[-16:] + ct_bytes = ct[:-16] + + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + aesgcm = AESGCM(key) + plaintext = aesgcm.decrypt(nonce.encode("utf-8"), ct_bytes + tag, associated_data.encode("utf-8")) + return json.loads(plaintext.decode("utf-8")) async def query_order(self, order_id: str) -> PaymentOrder: if not self._is_configured(): return await MockGateway().query_order(order_id) - logger.info(f"[WeChatPay] 查询订单: order_id={order_id}") - return PaymentOrder( - order_id=order_id, - pay_url="", - amount=0, - status="pending", + url_path = f"/v3/pay/transactions/out-trade-no/{order_id}?mchid={self.mch_id}" + url = f"{WECHAT_PAY_BASE_URL}{url_path}" + + timestamp = str(int(time.time())) + nonce_str = self._generate_nonce() + signature = self._build_signature("GET", url_path, timestamp, nonce_str, "") + auth_header = ( + f'WECHATPAY2-SHA256-RSA2048 ' + f'mchid="{self.mch_id}",' + f'nonce_str="{nonce_str}",' + f'timestamp="{timestamp}",' + f'derial_no="{self._serial_no}",' + f'signature="{signature}"' ) + headers = { + "Accept": "application/json", + "Authorization": auth_header, + } + + try: + resp = requests.get(url, headers=headers, timeout=10) + resp.raise_for_status() + data = resp.json() + trade_state = data.get("trade_state", "NOTPAY") + total = data.get("amount", {}).get("total", 0) + status_map = { + "SUCCESS": "completed", + "NOTPAY": "pending", + "CLOSED": "closed", + "REFUND": "refunded", + } + return PaymentOrder( + order_id=order_id, + pay_url="", + amount=total / 100, + status=status_map.get(trade_state, "pending"), + ) + except requests.RequestException as e: + logger.error("[WeChatPay] 查询订单失败: %s", e) + raise RuntimeError(f"WeChat Pay query_order failed: {e}") from e + async def refund(self, order_id: str, amount: float, reason: str = "") -> bool: if not self._is_configured(): return await MockGateway().refund(order_id, amount, reason) - logger.info(f"[WeChatPay] 申请退款: order_id={order_id}, amount={amount}") - return True + url_path = "/v3/refund/domestic/refunds" + url = f"{WECHAT_PAY_BASE_URL}{url_path}" + refund_id = f"R_{order_id}_{int(time.time())}" + body = json.dumps({ + "out_trade_no": order_id, + "out_refund_no": refund_id, + "reason": reason or "用户申请退款", + "amount": { + "refund": int(amount * 100), + "total": int(amount * 100), + "currency": "CNY", + }, + }) + + timestamp = str(int(time.time())) + nonce_str = self._generate_nonce() + signature = self._build_signature("POST", url_path, timestamp, nonce_str, body) + auth_header = ( + f'WECHATPAY2-SHA256-RSA2048 ' + f'mchid="{self.mch_id}",' + f'nonce_str="{nonce_str}",' + f'timestamp="{timestamp}",' + f'derial_no="{self._serial_no}",' + f'signature="{signature}"' + ) + + headers = { + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": auth_header, + } + + try: + resp = requests.post(url, data=body.encode("utf-8"), headers=headers, timeout=10) + resp.raise_for_status() + data = resp.json() + return data.get("status") in ("PROCESSING", "SUCCESS") + except requests.RequestException as e: + logger.error("[WeChatPay] 退款失败: %s", e) + return False diff --git a/backend/tests/test_content_pipeline/test_content_generation.py b/backend/tests/test_content_pipeline/test_content_generation.py index c48fd23..3001109 100644 --- a/backend/tests/test_content_pipeline/test_content_generation.py +++ b/backend/tests/test_content_pipeline/test_content_generation.py @@ -99,7 +99,7 @@ class TestRuleValidator: validator = RuleValidator() result = validator.validate( - content="A" * 50000, # 超过限制 + content="A" * 50001, title="测试标题", platform="zhihu" ) @@ -113,13 +113,13 @@ class TestRuleValidator: result = validator.validate( content="这是一篇正常的文章内容，包含足够的文字来通过校验。", - title="这是一个有效的标题", + title="这是一个有效的标题啊", platform="zhihu" ) # 验证通过检查 assert len(result.passed) > 0 - assert "标题长度" in result.passed[0] + assert any("标题长度" in p for p in result.passed) def test_validate_unsupported_platform(self): """测试不支持的平台""" diff --git a/backend/tests/test_content_pipeline/test_rule_validator.py b/backend/tests/test_content_pipeline/test_rule_validator.py index a1a1bf4..5b28d51 100644 --- a/backend/tests/test_content_pipeline/test_rule_validator.py +++ b/backend/tests/test_content_pipeline/test_rule_validator.py @@ -203,6 +203,69 @@ class TestRuleValidator: assert hasattr(result, 'passed') assert isinstance(result.passed, list) + def test_validate_wechat_external_link(self): + validator = RuleValidator() + result = validator.validate( + content="请访问 https://example.com 了解更多", + title="测试标题", + platform="wechat" + ) + assert any("外部链接" in i.message for i in result.issues) + + def test_validate_wechat_consecutive_symbols(self): + validator = RuleValidator() + result = validator.validate( + content="正常内容", + title="限时优惠！！！", + platform="wechat" + ) + assert any("连续特殊符号" in i.message for i in result.issues) + + def test_validate_baijiahao_clickbait(self): + validator = RuleValidator() + result = validator.validate( + content="这是一篇正常的文章内容", + title="震惊！这个产品竟然如此好用", + platform="baijiahao" + ) + assert any("标题党" in i.message for i in result.issues) + + def test_validate_toutiao_clickbait(self): + validator = RuleValidator() + result = validator.validate( + content="这是一篇正常的文章内容", + title="惊呆了这个消息刷屏了", + platform="toutiao" + ) + assert any("标题党" in i.message for i in result.issues) + + def test_validate_douyin_watermark(self): + validator = RuleValidator() + result = validator.validate( + content="视频来自抖音水印", + title="测试标题", + platform="douyin" + ) + assert any("水印" in i.message for i in result.issues) + + def test_validate_zhihu_marketing_words(self): + validator = RuleValidator() + result = validator.validate( + content="点击购买，限时折扣，优惠价", + title="产品推荐", + platform="zhihu" + ) + assert any("营销" in i.message for i in result.issues) + + def test_validate_xiaohongshu_content_too_long(self): + validator = RuleValidator() + result = validator.validate( + content="A" * 900, + title="笔记标题", + platform="xiaohongshu" + ) + assert any("300" in i.message or "800" in i.message for i in result.issues) + def test_validation_score_calculation(self): """验证分数计算逻辑""" validator = RuleValidator() diff --git a/backend/tests/test_services/test_attribution.py b/backend/tests/test_services/test_attribution.py new file mode 100644 index 0000000..c014630 --- /dev/null +++ b/backend/tests/test_services/test_attribution.py @@ -0,0 +1,146 @@ +import uuid +from datetime import UTC, datetime, timedelta + +import pytest +import pytest_asyncio +from unittest.mock import patch, AsyncMock, MagicMock +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker +from sqlalchemy.pool import StaticPool + +from app.database import Base +from app.models.attribution_record import AttributionRecord +from app.services.attribution.attribution_engine import AttributionEngine + + +@pytest.fixture +def engine(): + return create_async_engine( + "sqlite+aiosqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + + +@pytest_asyncio.fixture +async def session(engine): + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + async with session_maker() as s: + yield s + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.drop_all) + await engine.dispose() + + +class TestAttributionWindowDefault: + def test_default_window_days_is_28(self): + with patch("app.services.attribution.attribution_engine.settings") as mock_settings: + mock_settings.ATTRIBUTION_WINDOW_DAYS = 28 + engine = AttributionEngine() + assert engine.window_days == 28 + + @pytest.mark.asyncio + async def test_start_tracking_uses_default_28_days(self, session): + with patch("app.services.attribution.attribution_engine.settings") as mock_settings: + mock_settings.ATTRIBUTION_WINDOW_DAYS = 28 + eng = AttributionEngine() + + with patch.object(eng, "_get_latest_score", return_value=50.0): + record = await eng.start_tracking( + db=session, + brand_id=uuid.uuid4(), + content_id=uuid.uuid4(), + user_id="test_user", + ) + + assert record.attribution_window_days == 28 + expected_end = record.published_at + timedelta(days=28) + assert record.window_end_at is not None + time_diff = abs((record.window_end_at - expected_end).total_seconds()) + assert time_diff < 2 + + +class TestAttributionWindowCustom: + def test_custom_window_days_14(self): + with patch("app.services.attribution.attribution_engine.settings") as mock_settings: + mock_settings.ATTRIBUTION_WINDOW_DAYS = 14 + eng = AttributionEngine() + assert eng.window_days == 14 + + @pytest.mark.asyncio + async def test_start_tracking_uses_custom_14_days(self, session): + with patch("app.services.attribution.attribution_engine.settings") as mock_settings: + mock_settings.ATTRIBUTION_WINDOW_DAYS = 14 + eng = AttributionEngine() + + with patch.object(eng, "_get_latest_score", return_value=50.0): + record = await eng.start_tracking( + db=session, + brand_id=uuid.uuid4(), + content_id=uuid.uuid4(), + user_id="test_user", + ) + + assert record.attribution_window_days == 14 + expected_end = record.published_at + timedelta(days=14) + assert record.window_end_at is not None + time_diff = abs((record.window_end_at - expected_end).total_seconds()) + assert time_diff < 2 + + @pytest.mark.asyncio + async def test_start_tracking_uses_custom_7_days(self, session): + with patch("app.services.attribution.attribution_engine.settings") as mock_settings: + mock_settings.ATTRIBUTION_WINDOW_DAYS = 7 + eng = AttributionEngine() + + with patch.object(eng, "_get_latest_score", return_value=50.0): + record = await eng.start_tracking( + db=session, + brand_id=uuid.uuid4(), + content_id=uuid.uuid4(), + user_id="test_user", + ) + + assert record.attribution_window_days == 7 + expected_end = record.published_at + timedelta(days=7) + assert record.window_end_at is not None + time_diff = abs((record.window_end_at - expected_end).total_seconds()) + assert time_diff < 2 + + +class TestAttributionWindowConfigDriven: + def test_config_attribute_window_days_default(self): + from app.config import Settings + test_settings = Settings( + JWT_SECRET="a" * 32, + ATTRIBUTION_WINDOW_DAYS=28, + ) + assert test_settings.ATTRIBUTION_WINDOW_DAYS == 28 + + def test_config_attribute_window_days_custom(self): + from app.config import Settings + test_settings = Settings( + JWT_SECRET="a" * 32, + ATTRIBUTION_WINDOW_DAYS=14, + ) + assert test_settings.ATTRIBUTION_WINDOW_DAYS == 14 + + @pytest.mark.asyncio + async def test_window_end_matches_window_days(self, session): + with patch("app.services.attribution.attribution_engine.settings") as mock_settings: + mock_settings.ATTRIBUTION_WINDOW_DAYS = 21 + eng = AttributionEngine() + + with patch.object(eng, "_get_latest_score", return_value=50.0): + record = await eng.start_tracking( + db=session, + brand_id=uuid.uuid4(), + content_id=uuid.uuid4(), + user_id="test_user", + ) + + assert record.attribution_window_days == 21 + delta = record.window_end_at - record.published_at + assert delta.days == 21 diff --git a/backend/tests/test_services/test_payment.py b/backend/tests/test_services/test_payment.py new file mode 100644 index 0000000..5cb3a39 --- /dev/null +++ b/backend/tests/test_services/test_payment.py @@ -0,0 +1,365 @@ +import pytest +from unittest.mock import patch, MagicMock, AsyncMock + +from app.services.payment import get_payment_gateway +from app.services.payment.base import PaymentGateway, PaymentOrder, PaymentCallback +from app.services.payment.mock_gateway import MockGateway +from app.services.payment.wechat_pay import WeChatPayGateway +from app.services.payment.alipay import AlipayGateway + + +class TestPaymentGatewaySelection: + def test_mock_mode_returns_mock_gateway(self): + with patch("app.config.settings") as mock_settings: + mock_settings.PAYMENT_MODE = "mock" + mock_settings.WECHAT_PAY_MCH_ID = "" + gateway = get_payment_gateway("wechat") + assert isinstance(gateway, MockGateway) + + def test_unconfigured_wechat_returns_mock_gateway(self): + with patch("app.config.settings") as mock_settings: + mock_settings.PAYMENT_MODE = "live" + mock_settings.WECHAT_PAY_MCH_ID = "" + gateway = get_payment_gateway("wechat") + assert isinstance(gateway, MockGateway) + + def test_configured_wechat_returns_wechat_gateway(self): + with patch("app.config.settings") as mock_settings: + mock_settings.PAYMENT_MODE = "live" + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_key" + mock_settings.WECHAT_PAY_APP_ID = "wx123456" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + mock_settings.ALIPAY_APP_ID = "" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "" + mock_settings.ALIPAY_NOTIFY_URL = "" + gateway = get_payment_gateway("wechat") + assert isinstance(gateway, WeChatPayGateway) + + def test_configured_alipay_returns_alipay_gateway(self): + with patch("app.config.settings") as mock_settings: + mock_settings.PAYMENT_MODE = "live" + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_key" + mock_settings.WECHAT_PAY_APP_ID = "wx123456" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + mock_settings.ALIPAY_APP_ID = "alipay_app_123" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/key" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/pub" + mock_settings.ALIPAY_NOTIFY_URL = "" + gateway = get_payment_gateway("alipay") + assert isinstance(gateway, AlipayGateway) + + def test_unknown_provider_returns_mock_gateway(self): + with patch("app.config.settings") as mock_settings: + mock_settings.PAYMENT_MODE = "live" + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_key" + mock_settings.WECHAT_PAY_APP_ID = "wx123456" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + mock_settings.ALIPAY_APP_ID = "" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "" + mock_settings.ALIPAY_NOTIFY_URL = "" + gateway = get_payment_gateway("unknown") + assert isinstance(gateway, MockGateway) + + +class TestMockGateway: + @pytest.mark.asyncio + async def test_create_order(self): + gw = MockGateway() + result = await gw.create_order("order1", 99.9, "test", "user1", "pro") + assert isinstance(result, PaymentOrder) + assert result.order_id == "order1" + assert result.amount == 99.9 + assert result.status == "pending" + assert "mock://pay/order1" == result.pay_url + + @pytest.mark.asyncio + async def test_verify_callback(self): + gw = MockGateway() + result = await gw.verify_callback({"order_id": "order1", "amount": "99.9"}) + assert isinstance(result, PaymentCallback) + assert result.order_id == "order1" + assert result.status == "success" + + @pytest.mark.asyncio + async def test_query_order(self): + gw = MockGateway() + result = await gw.query_order("order1") + assert isinstance(result, PaymentOrder) + assert result.order_id == "order1" + assert result.status == "completed" + + @pytest.mark.asyncio + async def test_refund(self): + gw = MockGateway() + result = await gw.refund("order1", 99.9, "test") + assert result is True + + +class TestWeChatPayGatewayUnconfigured: + def test_is_configured_false(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "" + mock_settings.WECHAT_PAY_API_KEY = "" + mock_settings.WECHAT_PAY_APP_ID = "" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + gw = WeChatPayGateway() + assert gw._is_configured() is False + + @pytest.mark.asyncio + async def test_create_order_fallback_to_mock(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "" + mock_settings.WECHAT_PAY_API_KEY = "" + mock_settings.WECHAT_PAY_APP_ID = "" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + gw = WeChatPayGateway() + result = await gw.create_order("order1", 100.0, "test", "user1", "pro") + assert isinstance(result, PaymentOrder) + assert result.order_id == "order1" + assert "mock://pay/" in result.pay_url + + @pytest.mark.asyncio + async def test_verify_callback_fallback_to_mock(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "" + mock_settings.WECHAT_PAY_API_KEY = "" + mock_settings.WECHAT_PAY_APP_ID = "" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + gw = WeChatPayGateway() + result = await gw.verify_callback({"order_id": "order1", "amount": "100"}) + assert isinstance(result, PaymentCallback) + assert result.status == "success" + + +class TestWeChatPayGatewayConfigured: + def test_is_configured_true(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough" + mock_settings.WECHAT_PAY_APP_ID = "wx1234567890" + mock_settings.WECHAT_PAY_CERT_PATH = "/certs/apiclient_cert.pem" + mock_settings.WECHAT_PAY_NOTIFY_URL = "https://example.com/callback" + gw = WeChatPayGateway() + assert gw._is_configured() is True + + def test_build_signature(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough" + mock_settings.WECHAT_PAY_APP_ID = "wx1234567890" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + gw = WeChatPayGateway() + sig = gw._build_signature("POST", "/v3/pay/transactions/native", "1234567890", "abc123", '{"test":1}') + assert isinstance(sig, str) + assert len(sig) > 0 + + @pytest.mark.asyncio + async def test_create_order_calls_api(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough" + mock_settings.WECHAT_PAY_APP_ID = "wx1234567890" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "https://example.com/callback" + gw = WeChatPayGateway() + + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"code_url": "weixin://wxpay/bizpayurl?pr=test123"} + mock_resp.raise_for_status = MagicMock() + + with patch("app.services.payment.wechat_pay.requests.post", return_value=mock_resp): + result = await gw.create_order("order1", 100.0, "test plan", "user1", "pro") + assert isinstance(result, PaymentOrder) + assert result.order_id == "order1" + assert result.pay_url == "weixin://wxpay/bizpayurl?pr=test123" + assert result.amount == 100.0 + assert result.status == "pending" + + @pytest.mark.asyncio + async def test_create_order_api_failure(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough" + mock_settings.WECHAT_PAY_APP_ID = "wx1234567890" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "https://example.com/callback" + gw = WeChatPayGateway() + + import requests as req + with patch("app.services.payment.wechat_pay.requests.post", side_effect=req.RequestException("timeout")): + with pytest.raises(RuntimeError, match="WeChat Pay create_order failed"): + await gw.create_order("order1", 100.0, "test", "user1", "pro") + + @pytest.mark.asyncio + async def test_query_order_calls_api(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough" + mock_settings.WECHAT_PAY_APP_ID = "wx1234567890" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + gw = WeChatPayGateway() + + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = { + "trade_state": "SUCCESS", + "amount": {"total": 10000}, + } + mock_resp.raise_for_status = MagicMock() + + with patch("app.services.payment.wechat_pay.requests.get", return_value=mock_resp): + result = await gw.query_order("order1") + assert result.status == "completed" + assert result.amount == 100.0 + + @pytest.mark.asyncio + async def test_refund_calls_api(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough" + mock_settings.WECHAT_PAY_APP_ID = "wx1234567890" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + gw = WeChatPayGateway() + + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"status": "SUCCESS"} + mock_resp.raise_for_status = MagicMock() + + with patch("app.services.payment.wechat_pay.requests.post", return_value=mock_resp): + result = await gw.refund("order1", 100.0, "reason") + assert result is True + + @pytest.mark.asyncio + async def test_verify_callback_bad_signature(self): + with patch("app.services.payment.wechat_pay.settings") as mock_settings: + mock_settings.WECHAT_PAY_MCH_ID = "1234567890" + mock_settings.WECHAT_PAY_API_KEY = "test_api_key_32chars_long_enough" + mock_settings.WECHAT_PAY_APP_ID = "wx1234567890" + mock_settings.WECHAT_PAY_CERT_PATH = "" + mock_settings.WECHAT_PAY_NOTIFY_URL = "" + gw = WeChatPayGateway() + + result = await gw.verify_callback({ + "timestamp": "1234567890", + "nonce": "abc", + "body": "test", + "signature": "wrong_signature", + }) + assert result.status == "failed" + + +class TestAlipayGatewayUnconfigured: + def test_is_configured_false(self): + with patch("app.services.payment.alipay.settings") as mock_settings: + mock_settings.ALIPAY_APP_ID = "" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "" + mock_settings.ALIPAY_NOTIFY_URL = "" + gw = AlipayGateway() + assert gw._is_configured() is False + + @pytest.mark.asyncio + async def test_create_order_fallback_to_mock(self): + with patch("app.services.payment.alipay.settings") as mock_settings: + mock_settings.ALIPAY_APP_ID = "" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "" + mock_settings.ALIPAY_NOTIFY_URL = "" + gw = AlipayGateway() + result = await gw.create_order("order1", 100.0, "test", "user1", "pro") + assert isinstance(result, PaymentOrder) + assert "mock://pay/" in result.pay_url + + +class TestAlipayGatewayConfigured: + def test_is_configured_true(self): + with patch("app.services.payment.alipay.settings") as mock_settings: + mock_settings.ALIPAY_APP_ID = "alipay_2021" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem" + mock_settings.ALIPAY_NOTIFY_URL = "https://example.com/callback" + gw = AlipayGateway() + assert gw._is_configured() is True + + def test_build_sign_content(self): + with patch("app.services.payment.alipay.settings") as mock_settings: + mock_settings.ALIPAY_APP_ID = "alipay_2021" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem" + mock_settings.ALIPAY_NOTIFY_URL = "" + gw = AlipayGateway() + content = gw._build_sign_content({"b": "2", "a": "1", "c": "3"}) + assert content == "a=1&b=2&c=3" + + def test_build_common_params(self): + with patch("app.services.payment.alipay.settings") as mock_settings: + mock_settings.ALIPAY_APP_ID = "alipay_2021" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem" + mock_settings.ALIPAY_NOTIFY_URL = "https://example.com/notify" + gw = AlipayGateway() + params = gw._build_common_params("alipay.trade.wap.pay") + assert params["app_id"] == "alipay_2021" + assert params["method"] == "alipay.trade.wap.pay" + assert params["sign_type"] == "RSA2" + assert params["charset"] == "utf-8" + assert params["version"] == "1.0" + + @pytest.mark.asyncio + async def test_verify_callback_bad_signature(self): + with patch("app.services.payment.alipay.settings") as mock_settings: + mock_settings.ALIPAY_APP_ID = "alipay_2021" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem" + mock_settings.ALIPAY_NOTIFY_URL = "" + gw = AlipayGateway() + + with patch.object(gw, "_rsa2_verify", return_value=False): + result = await gw.verify_callback({ + "sign": "bad_sign", + "sign_type": "RSA2", + "out_trade_no": "order1", + "trade_no": "trade1", + "total_amount": "100.00", + "trade_status": "TRADE_SUCCESS", + }) + assert result.status == "failed" + + @pytest.mark.asyncio + async def test_verify_callback_good_signature(self): + with patch("app.services.payment.alipay.settings") as mock_settings: + mock_settings.ALIPAY_APP_ID = "alipay_2021" + mock_settings.ALIPAY_PRIVATE_KEY_PATH = "/path/to/private.pem" + mock_settings.ALIPAY_PUBLIC_KEY_PATH = "/path/to/public.pem" + mock_settings.ALIPAY_NOTIFY_URL = "" + gw = AlipayGateway() + + with patch.object(gw, "_rsa2_verify", return_value=True): + result = await gw.verify_callback({ + "sign": "good_sign", + "sign_type": "RSA2", + "out_trade_no": "order1", + "trade_no": "trade1", + "total_amount": "100.00", + "trade_status": "TRADE_SUCCESS", + }) + assert result.status == "success" + assert result.order_id == "order1" + assert result.payment_id == "trade1" + assert result.amount == 100.0 diff --git a/frontend/e2e/tests/content-monitoring.spec.ts b/frontend/e2e/tests/content-monitoring.spec.ts new file mode 100644 index 0000000..eea35f5 --- /dev/null +++ b/frontend/e2e/tests/content-monitoring.spec.ts @@ -0,0 +1,328 @@ +import { test, expect, describe } from "@playwright/test"; +import { LoginPage } from "../pages/login.page"; +import { DashboardPage } from "../pages/dashboard.page"; + +const TEST_USER = { + email: process.env.E2E_TEST_EMAIL || "admin@example.com", + password: process.env.E2E_TEST_PASSWORD || "admin@123", +}; + +async function loginAndWait(page: import("@playwright/test").Page) { + const loginPage = new LoginPage(page); + await loginPage.goto(); + await loginPage.login(TEST_USER.email, TEST_USER.password); + try { + await page.waitForURL(/\/dashboard/, { timeout: 60000 }); + } catch { + const currentUrl = page.url(); + if (!currentUrl.includes("/dashboard")) { + await loginPage.goto(); + await loginPage.login(TEST_USER.email, TEST_USER.password); + await page.waitForURL(/\/dashboard/, { timeout: 60000 }); + } + } + await page.waitForLoadState("networkidle"); +} + +async function hasProjects(page: import("@playwright/test").Page): Promise { + const dashboardPage = new DashboardPage(page); + await dashboardPage.waitForDashboardLoad(); + const emptyMsg = page.getByText("开始优化您的AI可见性"); + const errorTitle = page.getByText("数据加载失败"); + const isEmpty = await emptyMsg.isVisible().catch(() => false); + const isError = await errorTitle.isVisible().catch(() => false); + return !isEmpty && !isError; +} + +describe("内容工坊 - 页面加载测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("内容工坊页面标题正确显示", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByRole("heading", { name: "内容工坊" })).toBeVisible({ timeout: 15000 }); + }); + + test("内容工坊页面副标题正确显示", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByText("AI驱动的内容生产流水线")).toBeVisible({ timeout: 15000 }); + }); + + test("内容工坊页面显示AI生成新内容按钮", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const generateBtn = page.getByRole("button", { name: /AI生成新内容/ }); + await expect(generateBtn).toBeVisible({ timeout: 15000 }); + }); +}); + +describe("内容工坊 - 内容列表测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("有内容时显示内容卡片列表", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("还没有内容"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + + if (hasEmpty) { test.skip(); return; } + + const contentCards = page.locator(".bg-white.rounded-xl.border"); + const count = await contentCards.count(); + expect(count).toBeGreaterThanOrEqual(1); + }); + + test("内容卡片显示状态标签", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("还没有内容"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + + if (hasEmpty) { test.skip(); return; } + + const statusBadge = page.getByText(/草稿|待审核|已审核|已发布|已归档/).first(); + await expect(statusBadge).toBeVisible({ timeout: 10000 }); + }); + + test("内容卡片显示类型标签", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("还没有内容"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + + if (hasEmpty) { test.skip(); return; } + + const typeBadge = page.getByText(/文章|问答|知识库|社媒/).first(); + await expect(typeBadge).toBeVisible({ timeout: 10000 }); + }); + + test("空状态时显示引导文案和AI生成按钮", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("还没有内容"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasEmpty) { test.skip(); return; } + + await expect(emptyState).toBeVisible(); + const generateBtn = page.getByRole("button", { name: /AI生成新内容/ }); + await expect(generateBtn).toBeVisible(); + }); +}); + +describe("内容工坊 - 内容生成测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("点击AI生成新内容打开生成对话框", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const generateBtn = page.getByRole("button", { name: /AI生成新内容/ }).first(); + await generateBtn.click(); + + await expect(page.getByRole("dialog")).toBeVisible({ timeout: 10000 }); + await expect(page.getByText("AI生成新内容")).toBeVisible(); + }); + + test("生成对话框包含目标关键词输入框", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const generateBtn = page.getByRole("button", { name: /AI生成新内容/ }).first(); + await generateBtn.click(); + + await expect(page.getByLabel("目标关键词")).toBeVisible({ timeout: 10000 }); + }); + + test("生成对话框包含目标平台选择器", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const generateBtn = page.getByRole("button", { name: /AI生成新内容/ }).first(); + await generateBtn.click(); + + await expect(page.getByLabel("目标平台")).toBeVisible({ timeout: 10000 }); + }); + + test("未填写必填项时开始AI生成按钮禁用", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const generateBtn = page.getByRole("button", { name: /AI生成新内容/ }).first(); + await generateBtn.click(); + + const submitBtn = page.getByRole("button", { name: /开始AI生成/ }); + await expect(submitBtn).toBeDisabled({ timeout: 10000 }); + }); + + test("填写关键词和平台后开始AI生成按钮启用", async ({ page }) => { + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + + const generateBtn = page.getByRole("button", { name: /AI生成新内容/ }).first(); + await generateBtn.click(); + + const keywordInput = page.getByLabel("目标关键词"); + await keywordInput.fill("AI营销"); + + const platformSelect = page.getByLabel("目标平台"); + await platformSelect.click(); + await page.getByText("通用").click(); + + const submitBtn = page.getByRole("button", { name: /开始AI生成/ }); + await expect(submitBtn).toBeEnabled({ timeout: 10000 }); + }); +}); + +describe("监测优化 - 页面加载测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("监测优化页面标题正确显示", async ({ page }) => { + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByRole("heading", { name: "监测优化" })).toBeVisible({ timeout: 15000 }); + }); + + test("监测优化页面副标题正确显示", async ({ page }) => { + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByText("实时监控品牌AI可见性，及时响应告警通知")).toBeVisible({ timeout: 15000 }); + }); + + test("监测优化页面显示告警配置按钮", async ({ page }) => { + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + const settingsBtn = page.getByRole("button", { name: /告警配置/ }); + await expect(settingsBtn).toBeVisible({ timeout: 15000 }); + }); + + test("监测优化页面显示监测记录和告警通知Tab", async ({ page }) => { + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByRole("tab", { name: /监测记录/ })).toBeVisible({ timeout: 15000 }); + await expect(page.getByRole("tab", { name: /告警通知/ })).toBeVisible(); + }); +}); + +describe("监测优化 - 监测记录测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("无监测记录时显示空状态", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("暂无监测记录"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasEmpty) { test.skip(); return; } + + await expect(emptyState).toBeVisible(); + }); + + test("有监测记录时显示记录卡片", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("暂无监测记录"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + if (hasEmpty) { test.skip(); return; } + + const recordCards = page.locator("div.flex.items-center.gap-4"); + const count = await recordCards.count(); + expect(count).toBeGreaterThanOrEqual(1); + }); + + test("监测记录卡片包含立即检测按钮", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("暂无监测记录"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + if (hasEmpty) { test.skip(); return; } + + const checkBtn = page.getByRole("button", { name: /立即检测/ }).first(); + await expect(checkBtn).toBeVisible({ timeout: 10000 }); + }); + + test("监测记录卡片包含暂停/启用按钮", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + const emptyState = page.getByText("暂无监测记录"); + const hasEmpty = await emptyState.isVisible({ timeout: 10000 }).catch(() => false); + if (hasEmpty) { test.skip(); return; } + + const toggleBtn = page.getByRole("button", { name: /暂停|启用/ }).first(); + await expect(toggleBtn).toBeVisible({ timeout: 10000 }); + }); +}); + +describe("监测优化 - 告警通知测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("点击告警通知Tab切换到告警列表", async ({ page }) => { + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + const alertsTab = page.getByRole("tab", { name: /告警通知/ }); + await alertsTab.click(); + + await expect(page.getByText("告警列表").or(page.getByText("暂无告警"))).toBeVisible({ timeout: 10000 }); + }); + + test("告警通知Tab显示统计卡片", async ({ page }) => { + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + + const alertsTab = page.getByRole("tab", { name: /告警通知/ }); + await alertsTab.click(); + + const statCards = page.getByText(/未读告警|严重告警|今日新增|已处理/); + const count = await statCards.count(); + expect(count).toBeGreaterThanOrEqual(1); + }); +}); + +describe("内容→监测完整流程测试", () => { + test("从内容工坊导航到监测优化页面", async ({ page }) => { + await loginAndWait(page); + + await page.goto("/dashboard/content"); + await page.waitForLoadState("networkidle"); + await expect(page.getByRole("heading", { name: "内容工坊" })).toBeVisible({ timeout: 15000 }); + + await page.goto("/dashboard/monitoring"); + await page.waitForLoadState("networkidle"); + await expect(page.getByRole("heading", { name: "监测优化" })).toBeVisible({ timeout: 15000 }); + }); +}); diff --git a/frontend/e2e/tests/diagnosis-strategy.spec.ts b/frontend/e2e/tests/diagnosis-strategy.spec.ts new file mode 100644 index 0000000..9d7cd2e --- /dev/null +++ b/frontend/e2e/tests/diagnosis-strategy.spec.ts @@ -0,0 +1,324 @@ +import { test, expect, describe } from "@playwright/test"; +import { LoginPage } from "../pages/login.page"; +import { DashboardPage } from "../pages/dashboard.page"; + +const TEST_USER = { + email: process.env.E2E_TEST_EMAIL || "admin@example.com", + password: process.env.E2E_TEST_PASSWORD || "admin@123", +}; + +async function loginAndWait(page: import("@playwright/test").Page) { + const loginPage = new LoginPage(page); + await loginPage.goto(); + await loginPage.login(TEST_USER.email, TEST_USER.password); + try { + await page.waitForURL(/\/dashboard/, { timeout: 60000 }); + } catch { + const currentUrl = page.url(); + if (!currentUrl.includes("/dashboard")) { + await loginPage.goto(); + await loginPage.login(TEST_USER.email, TEST_USER.password); + await page.waitForURL(/\/dashboard/, { timeout: 60000 }); + } + } + await page.waitForLoadState("networkidle"); +} + +async function hasProjects(page: import("@playwright/test").Page): Promise { + const dashboardPage = new DashboardPage(page); + await dashboardPage.waitForDashboardLoad(); + const emptyMsg = page.getByText("开始优化您的AI可见性"); + const errorTitle = page.getByText("数据加载失败"); + const isEmpty = await emptyMsg.isVisible().catch(() => false); + const isError = await errorTitle.isVisible().catch(() => false); + return !isEmpty && !isError; +} + +describe("诊断分析 - 页面加载测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("诊断分析页面标题正确显示", async ({ page }) => { + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByRole("heading", { name: "诊断分析" })).toBeVisible({ timeout: 15000 }); + }); + + test("诊断分析页面副标题正确显示", async ({ page }) => { + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByText("全面评估品牌在搜索引擎和AI生成式引擎中的可见性")).toBeVisible({ timeout: 15000 }); + }); + + test("无品牌时显示空状态", async ({ page }) => { + if (await hasProjects(page)) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByText("暂无品牌数据")).toBeVisible({ timeout: 15000 }); + await expect(page.getByText("请先创建品牌，然后进行诊断分析")).toBeVisible(); + }); +}); + +describe("诊断分析 - 诊断流程测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("有品牌时显示重新诊断按钮", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const refreshBtn = page.getByRole("button", { name: /重新诊断/ }); + await expect(refreshBtn).toBeVisible({ timeout: 15000 }); + }); + + test("诊断页面显示三个评分卡片", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByText("综合评分")).toBeVisible({ timeout: 15000 }); + await expect(page.getByText("SEO诊断评分")).toBeVisible(); + await expect(page.getByText("GEO诊断评分")).toBeVisible(); + }); + + test("诊断页面显示三个Tab标签", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByRole("tab", { name: "综合诊断" })).toBeVisible({ timeout: 15000 }); + await expect(page.getByRole("tab", { name: "SEO诊断" })).toBeVisible(); + await expect(page.getByRole("tab", { name: "GEO诊断" })).toBeVisible(); + }); + + test("点击SEO诊断Tab显示SEO诊断详情", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const seoTab = page.getByRole("tab", { name: "SEO诊断" }); + await seoTab.click(); + + await expect(page.getByText("SEO诊断详情")).toBeVisible({ timeout: 10000 }); + }); + + test("点击GEO诊断Tab显示GEO诊断详情", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const geoTab = page.getByRole("tab", { name: "GEO诊断" }); + await geoTab.click(); + + await expect(page.getByText("GEO诊断详情")).toBeVisible({ timeout: 10000 }); + }); + + test("诊断结果显示5维度评分", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const geoTab = page.getByRole("tab", { name: "GEO诊断" }); + await geoTab.click(); + await page.waitForLoadState("networkidle"); + + const dimensionLabels = page.getByText(/内容可提取性|实体清晰度|E-E-A-T信号|Schema标记|主题权威|引用就绪度/); + const count = await dimensionLabels.count(); + expect(count).toBeGreaterThanOrEqual(1); + }); + + test("点击重新诊断按钮触发刷新", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const refreshBtn = page.getByRole("button", { name: /重新诊断/ }); + if (await refreshBtn.isVisible()) { + await refreshBtn.click(); + await page.waitForLoadState("networkidle"); + } + }); +}); + +describe("诊断分析 - 优先优化建议测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("诊断结果包含优先优化建议", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const recommendations = page.getByText("优先优化建议"); + const hasRecommendations = await recommendations.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasRecommendations) { test.skip(); return; } + + await expect(recommendations).toBeVisible(); + }); + + test("建议列表包含基于诊断制定GEO方案按钮", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const geoPlanBtn = page.getByRole("button", { name: /基于诊断制定GEO方案/ }); + const hasBtn = await geoPlanBtn.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasBtn) { test.skip(); return; } + + await expect(geoPlanBtn).toBeVisible(); + }); + + test("点击基于诊断制定GEO方案跳转到策略页面", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const geoPlanBtn = page.getByRole("button", { name: /基于诊断制定GEO方案/ }); + const hasBtn = await geoPlanBtn.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasBtn) { test.skip(); return; } + + await geoPlanBtn.click(); + await expect(page).toHaveURL(/\/dashboard\/strategy/, { timeout: 15000 }); + }); +}); + +describe("策略制定 - 页面加载测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("策略制定页面标题正确显示", async ({ page }) => { + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByRole("heading", { name: "策略制定" })).toBeVisible({ timeout: 15000 }); + }); + + test("策略制定页面副标题正确显示", async ({ page }) => { + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + await expect(page.getByText("制定GEO优化策略、关键词规划与目标设定")).toBeVisible({ timeout: 15000 }); + }); + + test("无方案时显示生成新方案按钮", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + const generateBtn = page.getByRole("button", { name: /生成新方案|生成优化方案/ }); + const hasBtn = await generateBtn.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasBtn) { test.skip(); return; } + + await expect(generateBtn.first()).toBeVisible(); + }); +}); + +describe("策略制定 - 方案详情测试", () => { + test.beforeEach(async ({ page }) => { + await loginAndWait(page); + }); + + test("有方案时显示诊断分数和目标分数", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + const scoreLabel = page.getByText("诊断分数 → 目标分数"); + const hasScore = await scoreLabel.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasScore) { test.skip(); return; } + + await expect(scoreLabel).toBeVisible(); + }); + + test("有方案时显示预计周数", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + const weeksLabel = page.getByText("预计周数"); + const hasWeeks = await weeksLabel.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasWeeks) { test.skip(); return; } + + await expect(weeksLabel).toBeVisible(); + }); + + test("有方案时显示行动项进度", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + const progressLabel = page.getByText("行动项进度"); + const hasProgress = await progressLabel.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasProgress) { test.skip(); return; } + + await expect(progressLabel).toBeVisible(); + }); + + test("有方案时显示行动项列表", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + const actionList = page.getByText("行动项列表"); + const hasActions = await actionList.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasActions) { test.skip(); return; } + + await expect(actionList).toBeVisible(); + }); + + test("行动项包含AI生成内容按钮", async ({ page }) => { + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/strategy"); + await page.waitForLoadState("networkidle"); + + const aiGenBtn = page.getByRole("button", { name: /AI生成内容/ }); + const hasBtn = await aiGenBtn.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasBtn) { test.skip(); return; } + + await expect(aiGenBtn.first()).toBeVisible(); + }); +}); + +describe("诊断→策略完整流程测试", () => { + test("从诊断页面导航到策略制定页面", async ({ page }) => { + await loginAndWait(page); + + if (!(await hasProjects(page))) { test.skip(); return; } + + await page.goto("/dashboard/diagnosis"); + await page.waitForLoadState("networkidle"); + + const geoPlanBtn = page.getByRole("button", { name: /基于诊断制定GEO方案/ }); + const hasBtn = await geoPlanBtn.isVisible({ timeout: 10000 }).catch(() => false); + if (!hasBtn) { test.skip(); return; } + + await geoPlanBtn.click(); + await expect(page).toHaveURL(/\/dashboard\/strategy/, { timeout: 15000 }); + await expect(page.getByRole("heading", { name: "策略制定" })).toBeVisible({ timeout: 15000 }); + }); +});