test(saas): Phase 1 integration tests — billing + scheduled_task + knowledge (68 tests)

- Fix TIMESTAMPTZ decode errors: add ::TEXT cast to all SELECT queries
  where Row structs use String for TIMESTAMPTZ columns (~22 locations)
- Fix Axum 0.7 route params: {id} → :id in billing/knowledge/scheduled_task routes
- Fix JSONB bind: scheduled_task INSERT uses ::jsonb cast for input_payload
- Add billing_test.rs (14 tests): plans, subscription, usage, payments, invoices
- Add scheduled_task_test.rs (12 tests): CRUD, validation, isolation
- Add knowledge_test.rs (20 tests): categories, items, versions, search, analytics, permissions
- Fix auth test regression: 6 tests were failing due to TIMESTAMPTZ type mismatch

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

crates/zclaw-saas/tests/billing_test.rs

@@ -0,0 +1,344 @@
+//! 计费模块集成测试
+//!
+//! 覆盖 billing 模块的 plan/subscription/usage/payment/invoice 端点。
+
+mod common;
+use common::*;
+use axum::http::StatusCode;
+
+// ── Plans（公开路由，不强制 auth） ──────────────────────────────
+
+#[tokio::test]
+async fn test_list_plans() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_plan_user").await;
+
+    let (status, body) = send(&app, get("/api/v1/billing/plans", &token)).await;
+    assert_eq!(status, StatusCode::OK, "list_plans failed: {body}");
+
+    let arr = body.as_array().expect("plans should be array");
+    assert!(arr.len() >= 3, "expected >= 3 seed plans, got {}", arr.len());
+
+    let names: Vec<&str> = arr.iter().filter_map(|p| p["name"].as_str()).collect();
+    assert!(names.contains(&"free"), "missing free plan");
+    assert!(names.contains(&"pro"), "missing pro plan");
+    assert!(names.contains(&"team"), "missing team plan");
+}
+
+#[tokio::test]
+async fn test_get_plan_by_id() {
+    let (app, pool) = build_test_app().await;
+    let token = register_token(&app, "billing_plan_get").await;
+
+    // 获取 free plan 的真实 ID
+    let plan_id: String = sqlx::query_scalar(
+        "SELECT id FROM billing_plans WHERE name = 'free' LIMIT 1"
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("no free plan seeded");
+
+    let (status, body) = send(&app, get(&format!("/api/v1/billing/plans/{}", plan_id), &token)).await;
+    assert_eq!(status, StatusCode::OK, "get_plan failed: {body}");
+    assert_eq!(body["name"], "free");
+    assert_eq!(body["price_cents"], 0);
+}
+
+#[tokio::test]
+async fn test_get_plan_not_found() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_plan_404").await;
+
+    let (status, _body) = send(&app, get("/api/v1/billing/plans/nonexistent-id", &token)).await;
+    assert_eq!(status, StatusCode::NOT_FOUND);
+}
+
+// ── Subscription / Usage（需认证） ─────────────────────────────
+
+#[tokio::test]
+async fn test_get_subscription() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_sub_user").await;
+
+    let (status, body) = send(&app, get("/api/v1/billing/subscription", &token)).await;
+    assert_eq!(status, StatusCode::OK, "get_subscription failed: {body}");
+
+    // 新用户应获得 free plan
+    assert_eq!(body["plan"]["name"], "free");
+    // 无活跃订阅
+    assert!(body["subscription"].is_null());
+    // 用量应为零
+    assert_eq!(body["usage"]["input_tokens"], 0);
+    assert_eq!(body["usage"]["relay_requests"], 0);
+}
+
+#[tokio::test]
+async fn test_get_usage() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_usage_user").await;
+
+    let (status, body) = send(&app, get("/api/v1/billing/usage", &token)).await;
+    assert_eq!(status, StatusCode::OK, "get_usage failed: {body}");
+
+    // 首次访问自动创建，所有计数为 0
+    assert_eq!(body["input_tokens"], 0);
+    assert_eq!(body["output_tokens"], 0);
+    assert_eq!(body["relay_requests"], 0);
+    assert_eq!(body["hand_executions"], 0);
+    assert_eq!(body["pipeline_runs"], 0);
+    // max 值来自 free plan limits
+    assert!(body["max_relay_requests"].is_number());
+}
+
+#[tokio::test]
+async fn test_increment_usage() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_incr_user").await;
+
+    // 递增 hand_executions
+    let (status, body) = send(&app, post(
+        "/api/v1/billing/usage/increment",
+        &token,
+        serde_json::json!({ "dimension": "hand_executions", "count": 3 }),
+    )).await;
+    assert_eq!(status, StatusCode::OK, "increment hand_executions failed: {body}");
+    assert_eq!(body["dimension"], "hand_executions");
+    assert_eq!(body["incremented"], 3);
+    assert_eq!(body["usage"]["hand_executions"], 3);
+
+    // 递增 relay_requests
+    let (status, body) = send(&app, post(
+        "/api/v1/billing/usage/increment",
+        &token,
+        serde_json::json!({ "dimension": "relay_requests", "count": 10 }),
+    )).await;
+    assert_eq!(status, StatusCode::OK, "increment relay_requests failed: {body}");
+    assert_eq!(body["usage"]["relay_requests"], 10);
+
+    // 递增 pipeline_runs
+    let (status, body) = send(&app, post(
+        "/api/v1/billing/usage/increment",
+        &token,
+        serde_json::json!({ "dimension": "pipeline_runs", "count": 1 }),
+    )).await;
+    assert_eq!(status, StatusCode::OK, "increment pipeline_runs failed: {body}");
+    assert_eq!(body["usage"]["pipeline_runs"], 1);
+}
+
+#[tokio::test]
+async fn test_increment_usage_invalid_dimension() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_incr_invaliddim").await;
+
+    let (status, body) = send(&app, post(
+        "/api/v1/billing/usage/increment",
+        &token,
+        serde_json::json!({ "dimension": "invalid_dim", "count": 1 }),
+    )).await;
+    assert_eq!(status, StatusCode::BAD_REQUEST, "should reject invalid dimension: {body}");
+}
+
+#[tokio::test]
+async fn test_increment_usage_invalid_count() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_incr_invalidcount").await;
+
+    // count = 0
+    let (status, _body) = send(&app, post(
+        "/api/v1/billing/usage/increment",
+        &token,
+        serde_json::json!({ "dimension": "hand_executions", "count": 0 }),
+    )).await;
+    assert_eq!(status, StatusCode::BAD_REQUEST, "should reject count=0");
+
+    // count = 101
+    let (status, _body) = send(&app, post(
+        "/api/v1/billing/usage/increment",
+        &token,
+        serde_json::json!({ "dimension": "hand_executions", "count": 101 }),
+    )).await;
+    assert_eq!(status, StatusCode::BAD_REQUEST, "should reject count=101");
+}
+
+// ── Payments（需认证） ─────────────────────────────────────────
+
+#[tokio::test]
+async fn test_create_payment() {
+    let (app, pool) = build_test_app().await;
+    let token = register_token(&app, "billing_pay_user").await;
+
+    // 获取 pro plan ID
+    let plan_id: String = sqlx::query_scalar(
+        "SELECT id FROM billing_plans WHERE name = 'pro' LIMIT 1"
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("no pro plan seeded");
+
+    let (status, body) = send(&app, post(
+        "/api/v1/billing/payments",
+        &token,
+        serde_json::json!({ "plan_id": plan_id, "payment_method": "alipay" }),
+    )).await;
+    assert_eq!(status, StatusCode::OK, "create_payment failed: {body}");
+
+    // 应返回支付信息
+    assert!(body["payment_id"].is_string(), "missing payment_id");
+    assert!(body["trade_no"].is_string(), "missing trade_no");
+    assert!(body["pay_url"].is_string(), "missing pay_url");
+    assert!(body["amount_cents"].is_number(), "missing amount_cents");
+}
+
+#[tokio::test]
+async fn test_create_payment_invalid_plan() {
+    let (app, _pool) = build_test_app().await;
+    let token = register_token(&app, "billing_pay_invalidplan").await;
+
+    let (status, body) = send(&app, post(
+        "/api/v1/billing/payments",
+        &token,
+        serde_json::json!({ "plan_id": "nonexistent-plan", "payment_method": "alipay" }),
+    )).await;
+    assert_eq!(status, StatusCode::NOT_FOUND, "should 404 for invalid plan: {body}");
+}
+
+#[tokio::test]
+async fn test_get_payment_status() {
+    let (app, pool) = build_test_app().await;
+    let token = register_token(&app, "billing_paystatus_user").await;
+
+    // 先创建支付
+    let plan_id: String = sqlx::query_scalar(
+        "SELECT id FROM billing_plans WHERE name = 'pro' LIMIT 1"
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("no pro plan");
+
+    let (_, create_body) = send(&app, post(
+        "/api/v1/billing/payments",
+        &token,
+        serde_json::json!({ "plan_id": plan_id, "payment_method": "alipay" }),
+    )).await;
+
+    let payment_id = create_body["payment_id"].as_str().expect("missing payment_id");
+
+    // 查询支付状态
+    let (status, body) = send(&app, get(
+        &format!("/api/v1/billing/payments/{}", payment_id),
+        &token,
+    )).await;
+    assert_eq!(status, StatusCode::OK, "get_payment_status failed: {body}");
+    assert_eq!(body["status"], "pending");
+}
+
+#[tokio::test]
+async fn test_mock_pay_flow() {
+    let (app, pool) = build_test_app().await;
+    let token = register_token(&app, "billing_mockpay_user").await;
+
+    let plan_id: String = sqlx::query_scalar(
+        "SELECT id FROM billing_plans WHERE name = 'pro' LIMIT 1"
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("no pro plan");
+
+    // 1. 创建支付
+    let (_, create_body) = send(&app, post(
+        "/api/v1/billing/payments",
+        &token,
+        serde_json::json!({ "plan_id": plan_id, "payment_method": "alipay" }),
+    )).await;
+
+    let trade_no = create_body["trade_no"].as_str().expect("missing trade_no");
+    let amount = create_body["amount_cents"].as_i64().expect("missing amount_cents") as i32;
+
+    // 2. Mock 支付确认（返回 HTML，不能用 JSON 解析）
+    let csrf_token = generate_test_csrf_token(trade_no);
+    let form_body = format!(
+        "trade_no={}&action=success&csrf_token={}",
+        urlencoding::encode(trade_no),
+        urlencoding::encode(&csrf_token),
+    );
+    let req = axum::http::Request::builder()
+        .method("POST")
+        .uri("/api/v1/billing/mock-pay/confirm")
+        .header("Content-Type", "application/x-www-form-urlencoded")
+        .body(axum::body::Body::from(form_body))
+        .unwrap();
+
+    let (status, body) = send_raw(&app, req).await;
+    assert!(status == StatusCode::OK, "mock pay confirm should succeed: status={}, body={}", status, body);
+    assert!(body.contains("支付成功"), "expected success message in HTML: {}", body);
+}
+
+#[tokio::test]
+async fn test_invoice_pdf_requires_paid() {
+    let (app, pool) = build_test_app().await;
+    let token = register_token(&app, "billing_invoice_user").await;
+
+    let plan_id: String = sqlx::query_scalar(
+        "SELECT id FROM billing_plans WHERE name = 'pro' LIMIT 1"
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("no pro plan");
+
+    // 创建支付 → 产生 pending 发票
+    let (_, create_body) = send(&app, post(
+        "/api/v1/billing/payments",
+        &token,
+        serde_json::json!({ "plan_id": plan_id, "payment_method": "alipay" }),
+    )).await;
+
+    let payment_id = create_body["payment_id"].as_str().expect("missing payment_id");
+
+    // 查找关联发票
+    let invoice_id: Option<String> = sqlx::query_scalar(
+        "SELECT invoice_id FROM billing_payments WHERE id = $1"
+    )
+    .bind(payment_id)
+    .fetch_optional(&pool)
+    .await
+    .expect("db error")
+    .flatten();
+
+    if let Some(inv_id) = invoice_id {
+        // 发票未支付，应返回 400
+        let (status, _body) = send(&app, get(
+            &format!("/api/v1/billing/invoices/{}/pdf", inv_id),
+            &token,
+        )).await;
+        assert_eq!(status, StatusCode::BAD_REQUEST, "unpaid invoice should reject PDF download");
+    }
+}
+
+#[tokio::test]
+async fn test_payment_callback() {
+    let (app, _pool) = build_test_app().await;
+
+    // 模拟支付宝回调（开发模式，不验签）
+    let callback_body = "out_trade_no=ZCLAW-INVALID-TEST&trade_status=TRADE_SUCCESS&total_amount=0.01";
+    let req = axum::http::Request::builder()
+        .method("POST")
+        .uri("/api/v1/billing/callback/alipay")
+        .header("Content-Type", "application/x-www-form-urlencoded")
+        .body(axum::body::Body::from(callback_body))
+        .unwrap();
+
+    let (status, body) = send_raw(&app, req).await;
+    // 回调返回纯文本 "success" 或 "fail"，不是 JSON
+    assert!(
+        status == StatusCode::OK || status == StatusCode::BAD_REQUEST,
+        "callback should be processed or rejected gracefully: status={}, body={}", status, body
+    );
+}
+
+/// 生成测试用 CSRF token（复制 handlers.rs 中的逻辑）
+fn generate_test_csrf_token(trade_no: &str) -> String {
+    use sha2::{Sha256, Digest};
+    let message = format!("ZCLAW_MOCK:{}:", trade_no);
+    let hash = Sha256::digest(message.as_bytes());
+    hex::encode(hash)
+}