测试验收用例（非功能性与端到端）

文档版本：1.0（正式版） 最后更新：2026-04-23 维护者：Reflekt Health 产品团队

文档依据：functional_requirements.md、non_functional_requirements.md、business_architecture.md、user_stories.md、technical_architecture.md、technical_debt_and_risks.md、software_endpoints.md

使用说明

本文档覆盖 Reflekt Health MVP 的全部测试验收用例，按类型分为四大类：

类别	测试内容	测试方法	负责人
性能压测	API 响应时间、告警端到端延迟	JMeter / k6	妙锋
安全合规	SQL 注入、XSS、数据加密、HIPAA 合规	自动化脚本 + 人工审计	妙锋 + 龙
端到端	跌倒告警闭环、升级链路、离线兜底	Playwright / 手动物料测试	妙锋
接口自动化	D1-D5 微服务 API	pytest + requests	妙锋

1. 性能压测用例

1.1 API 性能压测

场景	并发	验收标准	脚本路径
登录接口压测	100 并发用户	P95 < 200ms，错误率 < 0.1%	`tests/performance/jmeter/login.jmx`
首页状态聚合查询	100 并发用户	P95 < 200ms，错误率 < 0.1%	`tests/performance/jmeter/home_status.jmx`
设备绑定接口	50 并发用户	P95 < 500ms，错误率 < 1%	`tests/performance/jmeter/device_bind.jmx`
预警确认接口	50 并发用户	P95 < 500ms，错误率 < 1%	`tests/performance/jmeter/alert_ack.jmx`
提醒创建接口	50 并发用户	P95 < 500ms，错误率 < 1%	`tests/performance/jmeter/reminder_create.jmx`
设备心跳上报	100 并发	P95 < 200ms，成功率 ≥ 99.5%	`tests/performance/jmeter/heartbeat.jmx`
核心交易吞吐	持续 10 分钟	≥ 100 TPS	`tests/performance/k6/throughput.js`

1.1.1 JMeter 脚本配置参考

# tests/performance/jmeter/login.jmx 配置片段
thread_groups:
  - name: "登录压测"
    num_threads: 100
    ramp_up: 30s
    duration: 300s
    sampler:
      method: POST
      url: "{{API_BASE}}/api/v1/auth/login"
      body:
        phone: "13800138000"
        password: "Test@123"
    assertions:
      - response_time_p95: 200ms
      - error_rate: 0.1%
      - status_code: 200

1.1.2 k6 吞吐测试脚本

// tests/performance/k6/throughput.js
import http from 'k6/http';
import { check, sleep } from 'k6';

export const options = {
  stages: [
    { duration: '2m', target: 100 },  // 预热
    { duration: '10m', target: 100 },  // 稳态压测
  ],
  thresholds: {
    http_req_duration: ['p(95)<200'],
    http_reqs: ['rate>100'],           // ≥ 100 TPS
    errors: ['rate<0.001'],            // 错误率 < 0.1%
  },
};

export default function () {
  // 设备心跳上报
  const res = http.post(
    `${__ENV.API_BASE}/api/v1/device/heartbeat`,
    JSON.stringify({ device_id: `device_${Math.floor(Math.random() * 10)}` }),
    { headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${__ENV.TOKEN}` } }
  );
  check(res, {
    'heartbeat status 200': (r) => r.status === 200,
    'heartbeat p95 < 200ms': (r) => r.timings.duration < 200,
  });
  sleep(0.1);
}

1.2 红色预警端到端延迟压测

核心指标：告警触发 → 家属知晓 < 5 分钟（300 秒）

子场景	测试方法	验收标准	脚本路径
雷达跌倒 → Family App 推送收到	JMeter + Mock 雷达事件	全链路 ≤ 300s，P99 ≤ 280s	`tests/performance/jmeter/fall_to_alert_e2e.jmx`
升级链路 T=30/50/70/90s 准确性	Mock 联系人未 ACK 场景	每级升级延迟误差 ≤ ±2s	`tests/performance/jmeter/escalation_timing.jmx`
APNs / FCM 推送送达率	100 台设备同时推送	送达率 ≥ 99%，P95 < 60s	`tests/performance/k6/push_delivery.js`
Twilio 语音直拨接通率	Mock 跌倒触发语音呼叫	接通率 ≥ 95%	人工测试 + Twilio 日志审计

1.2.1 端到端延迟测试场景

# tests/performance/jmeter/fall_to_alert_e2e.jmx
# 测试场景：模拟雷达跌倒事件 → AI大脑 → Luma询问无应答 → 升级告警 → Family App推送
test_sequence:
  - name: "跌倒事件触发"
    step: 1
    action: "Mock 雷达通过 MQTT 上报 fall_detected 事件"
    expected: "设备接入服务消费事件，延迟 < 1s"

  - name: "AI大脑综合判断"
    step: 2
    action: "D4 服务接收跌倒事件，延迟 < 2s"
    expected: "下发 TTS 询问指令至 Luma"

  - name: "Luma询问播报"
    step: 3
    action: "Luma 屏幕显示 ALERT 状态，播放 TTS"
    expected: "延迟 < 5s"

  - name: "30秒确认窗口"
    step: 4
    action: "30秒内无老人语音应答"
    expected: "超时后触发 RED_ALERT"

  - name: "T=30s 通知第1联系人"
    step: 5
    action: "D5 服务同时触发 APNs + Twilio SMS"
    expected: "延迟 < 60s（含 D5 处理时间）"

  - name: "家属知晓"
    step: 6
    action: "Family App 收到推送或 Twilio 语音被接听"
    expected: "总延迟 < 300s（5分钟）"

1.3 设备心跳监控压测

场景	并发	验收标准	脚本路径
50 台设备同时心跳上报	50 QPS	P95 < 200ms，成功率 ≥ 99.5%	`tests/performance/jmeter/heartbeat_50_devices.jmx`
手表数据上报（埃微 API 模拟）	20 QPS	P95 < 500ms，错误率 < 1%	`tests/performance/jmeter/watch_data.jmx`
Redis 心跳缓存写入	100 QPS	P95 < 50ms，内存稳定	`tests/performance/redis/heartbeat_cache.lua`
Luma WebSocket 心跳（30s间隔）	100 长连接	断连率 < 0.5%	`tests/performance/k6/ws_heartbeat.js`

2. 安全合规测试用例

2.1 认证与授权安全

测试项	测试方法	输入payload	预期结果	自动化脚本
SQL 注入（登录）	手动注入	`' OR '1'='1`	返回参数错误，不认证成功	`tests/security/test_sql_injection.py`
SQL 注入（设备绑定）	自动化	`device_id=1;DROP TABLE devices;--`	参数校验失败，返回 400	`tests/security/test_sql_injection.py`
XSS（语音消息内容）	自动化	`<script>alert(1)</script>`	输出转义，不执行脚本	`tests/security/test_xss.py`
越权访问（跨家庭）	自动化	家庭A的 Token 访问家庭B数据	返回 403 Forbidden	`tests/security/test_authz.py`
Token 伪造	自动化	`Authorization: Bearer fake_token`	返回 401 Unauthorized	`tests/security/test_authz.py`
JWT 过期验证	自动化	过期 2 小时的 Token	返回 401，提示重新登录	`tests/security/test_token_expiry.py`
多端登录会话管理	自动化	同一账号在 3 台设备登录	各端均可使用，密码重置后其他端失效	`tests/security/test_session.py`

2.1.1 SQL 注入测试脚本

# tests/security/test_sql_injection.py
import pytest
import requests

API_BASE = "https://api.reflekt.health/api/v1"

INJECTION_PAYLOADS = [
    "' OR '1'='1",
    "'; DROP TABLE users; --",
    "1; DELETE FROM devices WHERE 1=1;--",
    "<script>alert(1)</script>",
    "../../../etc/passwd",
]

@pytest.mark.parametrize("payload", INJECTION_PAYLOADS)
def test_sql_injection_login(payload):
    """测试登录接口 SQL 注入防护"""
    resp = requests.post(
        f"{API_BASE}/auth/login",
        json={"phone": payload, "password": "anything"}
    )
    # 期望：返回 400 参数错误，而非 200 登录成功
    assert resp.status_code in [400, 422], \
        f"SQL注入 payload {payload} 未被拦截！响应: {resp.text}"
    data = resp.json()
    assert data.get("code") != 0, "SQL注入未被防护"

@pytest.mark.parametrize("payload", INJECTION_PAYLOADS)
def test_sql_injection_device_id(payload):
    """测试设备 ID 参数 SQL 注入防护"""
    resp = requests.get(
        f"{API_BASE}/device/{payload}/status",
        headers={"Authorization": f"Bearer {TOKEN}"}
    )
    assert resp.status_code in [400, 404, 422], \
        f"SQL注入 payload {payload} 未被拦截！响应: {resp.text}"

2.1.2 跨家庭越权访问测试脚本

# tests/security/test_authz.py
def test_cross_family_data_isolation():
    """测试多租户数据隔离：家庭A不能访问家庭B的数据"""
    # 家庭A的Token
    token_a = login("family_a_phone", "password_a")
    # 家庭B的设备ID
    device_id_b = get_device_id("family_b_phone", "password_b")

    # 尝试用家庭A的Token访问家庭B的设备详情
    resp = requests.get(
        f"{API_BASE}/device/{device_id_b}/detail",
        headers={"Authorization": f"Bearer {token_a}"}
    )
    assert resp.status_code == 403, \
        f"跨家庭访问未被阻止！响应: {resp.text}"
    data = resp.json()
    assert "权限" in str(data) or "forbidden" in str(data).lower()

2.2 数据安全与 HIPAA 合规

测试项	测试方法	验收标准	自动化脚本
敏感字段 AES-256 加密	DB 直接查询	联系方式/疾病史在 DB 中为密文	`tests/security/test_encryption.py`
传输加密（TLS 1.2+）	Wireshark 抓包	所有 HTTP 请求强制跳转 HTTPS	人工审计
密码 bcrypt 加盐存储	DB 查询	密码字段非明文，含盐值	`tests/security/test_password_hash.py`
日志保留期 ≥ 12 个月	审计日志表查询	历史日志不被物理删除	`tests/security/test_log_retention.py`
同意记录完整存储	API + DB 验证	撤回同意后数据停止采集	`tests/security/test_consent.py`
CSRF Token 校验	自动化	无 Token 的 POST 请求被拒绝	`tests/security/test_csrf.py`
HIPAA 审计覆盖	日志查询	所有敏感操作均有 traceId	人工审计

2.2.1 敏感字段加密验证脚本

# tests/security/test_encryption.py
import psycopg2
import base64
import re

def test_phone_encrypted_in_db():
    """验证手机号在数据库中为 AES-256 密文"""
    conn = psycopg2.connect(DATABASE_URL)
    cur = conn.cursor()
    cur.execute("SELECT phone FROM ref_caregiver LIMIT 5")
    rows = cur.fetchall()
    conn.close()

    for (phone_value,) in rows:
        phone_str = str(phone_value)
        # 明文手机号特征：11位纯数字，如 13812345678
        is_plain = bool(re.match(r'^\d{11}$', phone_str))
        assert not is_plain, \
            f"手机号未加密！发现明文: {phone_str[:3]}****{phone_str[-4:]}"
        # AES密文特征：Base64编码，长度 > 32
        is_cipher = len(phone_str) > 32 and bool(re.match(r'^[A-Za-z0-9+/=]+$', phone_str))
        assert is_cipher, \
            f"手机号字段不是标准AES密文格式: {phone_str}"

def test_health_profile_encrypted():
    """验证老人健康档案（疾病史/用药史）在数据库中加密"""
    conn = psycopg2.connect(DATABASE_URL)
    cur = conn.cursor()
    cur.execute("SELECT health_profile FROM ref_elder LIMIT 5")
    rows = cur.fetchall()
    conn.close()

    for (profile,) in rows:
        profile_str = str(profile)
        # JSON明文关键词
        dangerous_keywords = ["高血压", "心脏病", "糖尿病", "insulin", "heart disease"]
        for kw in dangerous_keywords:
            assert kw not in profile_str, \
                f"健康档案包含明文敏感词 '{kw}'！内容: {profile_str[:50]}..."

2.2.2 HIPAA 合规审计覆盖测试

# tests/security/test_hipaa_audit.py
import requests
from datetime import datetime, timedelta

def test_audit_log_for_sensitive_operations():
    """验证所有敏感操作均记录审计日志（HIPAA 要求）"""
    sensitive_ops = [
        ("DELETE", "/api/v1/user/withdraw", "账号注销"),
        ("PUT", "/api/v1/device/unbind", "设备解绑"),
        ("PUT", "/api/v1/caregiver/role", "权限变更"),
        ("POST", "/api/v1/data/export", "数据导出"),
        ("DELETE", "/api/v1/data/delete", "数据删除"),
    ]

    for method, path, op_name in sensitive_ops:
        # 执行敏感操作
        resp = requests.request(
            method, f"{API_BASE}{path}",
            headers={"Authorization": f"Bearer {ADMIN_TOKEN}"},
            json={"confirm": "DELETE"}
        )
        # 验证审计日志记录
        audit_resp = requests.get(
            f"{API_BASE}/admin/audit/logs",
            params={"operation": op_name, "start_time": (datetime.now() - timedelta(minutes=5)).isoformat()},
            headers={"Authorization": f"Bearer {ADMIN_TOKEN}"}
        )
        audit_data = audit_resp.json()
        assert audit_data.get("total") >= 1, \
            f"操作 '{op_name}' 未记录审计日志！HIPAA 合规失败"
        log = audit_data["list"][0]
        assert "traceId" in log, "审计日志缺少 traceId，无法串联"
        assert "operator" in log, "审计日志缺少操作人"
        assert "timestamp" in log, "审计日志缺少时间戳"

2.3 API 安全测试

测试项	输入	预期	自动化脚本
短信验证码暴力猜解	同一手机号 100 次错误验证码	第 6 次后触发验证码锁定	`tests/security/test_sms_bruteforce.py`
验证码有效期校验	过期 5 分钟的验证码	提示"验证码已过期"	`tests/security/test_otp_expiry.py`
请求频率限制	1 秒内 100 次登录请求	返回 429 Too Many Requests	`tests/security/test_rate_limit.py`
文件上传类型限制	上传 `.exe` / `.php` 文件	拒绝上传，返回 400	`tests/security/test_upload.py`
OpenAI API Key 隔离	模拟 OpenAI 服务不可用	Fallback 到固定话术，不暴露 Key	`tests/security/test_ai_fallback.py`
WebSocket Token 鉴权	无效 Token 建立 WebSocket	连接拒绝	`tests/security/test_ws_auth.py`

2.3.1 Sa-Token 认证机制专项测试

后端使用 Sa-Token + JWT，会话超时 2 小时（参照 software_endpoints.md §2.4）。以下为 Sa-Token 框架特定行为测试。

测试项	技术细节	预期	自动化脚本
Token 自动续期	接近 2h 过期时自动 refresh	Token 静默续期，无感知	`tests/security/test_satoken_refresh.py`
密码重置后所有会话失效	修改密码后旧 Token 仍有效？	旧 Token 被强制失效（Sa-Token 的 `logout` 行为）	`tests/security/test_satoken_session.py`
多端登录态独立管理	同一账号 3 台设备登录，密码重置	仅目标设备失效，其他设备保持	`tests/security/test_satoken_session.py`
分布式 Token 验证	D1、D2、D3 各验证同一 Token	各服务验证一致（Redis 共享 session）	`tests/security/test_satoken_distributed.py`
Token 刷新机制	refresh_token 换取新 access_token	新 Token 有效，旧 Token 失效	`tests/security/test_satoken_refresh.py`

# tests/security/test_satoken_refresh.py
import pytest
import requests
import time

class TestSaTokenAuth:
    """Sa-Token JWT 认证机制测试"""

    def test_token_auto_refresh_before_expiry(self):
        """Token 接近过期（2h）时自动续期"""
        # 登录获取 Token
        resp = requests.post(
            f"{API_BASE}/auth/login",
            json={"phone": "13800000002", "password": "Family@123"}
        )
        data = resp.json()
        access_token = data["data"]["access_token"]
        refresh_token = data["data"]["refresh_token"]

        # 验证 Token 有效
        me_resp = requests.get(
            f"{API_BASE}/auth/me",
            headers={"Authorization": f"Bearer {access_token}"}
        )
        assert me_resp.status_code == 200

        # 模拟 Token 即将过期（直接调用刷新接口）
        refresh_resp = requests.post(
            f"{API_BASE}/auth/refresh",
            json={"refresh_token": refresh_token}
        )
        assert refresh_resp.status_code == 200
        new_data = refresh_resp.json()
        assert "access_token" in new_data["data"]
        assert new_data["data"]["access_token"] != access_token, "刷新后 Token 应不同"

    def test_password_change_invalidates_all_tokens(self):
        """密码修改后所有设备 Token 失效（Sa-Token logoutAll）"""
        # 设备 A 登录
        resp_a = requests.post(f"{API_BASE}/auth/login", json={"phone": "13800000002", "password": "Family@123"})
        token_a = resp_a.json()["data"]["access_token"]

        # 设备 B 登录
        resp_b = requests.post(f"{API_BASE}/auth/login", json={"phone": "13800000002", "password": "Family@123"})
        token_b = resp_b.json()["data"]["access_token"]

        # 修改密码
        requests.post(
            f"{API_BASE}/auth/change-password",
            headers={"Authorization": f"Bearer {token_a}"},
            json={"old_password": "Family@123", "new_password": "Family@456"}
        )

        # 设备 A 的旧 Token 应失效
        resp_a_check = requests.get(
            f"{API_BASE}/auth/me",
            headers={"Authorization": f"Bearer {token_a}"}
        )
        assert resp_a_check.status_code in [401, 403], "密码修改后旧 Token 应失效"

        # 设备 B 的 Token 也应失效（Sa-Token logoutAll 行为）
        resp_b_check = requests.get(
            f"{API_BASE}/auth/me",
            headers={"Authorization": f"Bearer {token_b}"}
        )
        assert resp_b_check.status_code in [401, 403], "logoutAll 后其他设备 Token 也应失效"

    def test_distributed_token_validation_across_services(self):
        """跨服务（D1/D2/D3）验证同一 Token 结果一致"""
        resp = requests.post(f"{API_BASE}/auth/login", json={"phone": "13800000002", "password": "Family@123"})
        token = resp.json()["data"]["access_token"]

        # D1 验证
        d1_resp = requests.get(
            f"{API_BASE}/device/list",
            headers={"Authorization": f"Bearer {token}"}
        )
        # D2 验证
        d2_resp = requests.get(
            f"{API_BASE}/family/devices",
            headers={"Authorization": f"Bearer {token}"}
        )
        # D3 验证
        d3_resp = requests.get(
            f"{ADMIN_API_BASE}/admin/family/list",
            headers={"Authorization": f"Bearer {token}"}
        )

        # 任一服务拒绝则 Token 验证不一致
        assert d1_resp.status_code == 200, "D1 Token 验证失败"
        assert d2_resp.status_code == 200, "D2 Token 验证失败"
        # 注：D3 需要 admin 权限，此处验证需使用 admin Token

3. 端到端测试用例（Playwright）

端到端测试使用 Playwright，覆盖 Family App、Luma App 双端的核心业务流程。

3.1 跌倒告警闭环（US-09，核心链路）

// tests/e2e/fall_to_alert.spec.ts
import { test, expect, Page } from '@playwright/test';

test.describe('US-09 跌倒告警闭环端到端', () => {

  test('TC-E2E-09-01: 跌倒检测 → 30秒无应答 → 红色预警 → FAP推送', async ({ page }) => {
    // Step 1: Mock 雷达上报跌倒事件（通过 D1 HTTP 接口）
    await page.request.post(`${API_BASE}/api/v1/device/event`, {
      headers: { 'Authorization': `Bearer ${DEVICE_TOKEN}` },
      data: {
        device_id: 'radar_001',
        event_type: 'fall_detected',
        payload: { fall_confidence: 0.92, distance: 1.8 }
      }
    });

    // Step 2: 等待 Luma 询问 ALERT 状态（延迟约 5s）
    await page.waitForTimeout(6000);
    await page.goto(`${FAP_URL}/pages/home`);
    await page.waitForLoadState('networkidle');

    // Step 3: 验证 Family App 首页显示红色预警状态（30秒后）
    const alertBanner = page.locator('.alert-red-banner');
    await expect(alertBanner).toBeVisible({ timeout: 35000 });
    await expect(alertBanner).toContainText(/关注|Alert|跌倒/);

    // Step 4: 验证预警卡片包含操作入口
    const alertCard = page.locator('.alert-card');
    await expect(alertCard).toBeVisible();
    await expect(page.locator('text=立即联系')).toBeVisible();

    // Step 5: 家属点击"标记为已处理"
    await page.click('button:has-text("标记为已处理")');
    await page.waitForTimeout(2000);

    // Step 6: 验证事件状态更新为已处理
    const resolvedBadge = page.locator('.status-resolved');
    await expect(resolvedBadge).toBeVisible();
  });

  test('TC-E2E-09-02: 跌倒检测 → 老人应答"Im okay" → 无预警生成', async ({ page }) => {
    // Step 1: Mock 雷达跌倒事件
    await page.request.post(`${API_BASE}/api/v1/device/event`, {
      data: { device_id: 'radar_001', event_type: 'fall_detected', payload: { fall_confidence: 0.92 } }
    });

    // Step 2: 模拟老人在 30 秒内应答（通过 WebSocket 上报 USER_OK）
    await page.waitForTimeout(5000);
    await page.evaluate(() => {
      // 通过内部 API 模拟 Luma App 上报 USER_OK
      fetch(`${API_BASE}/api/v1/lsp/voice-event`, {
        method: 'POST',
        headers: { 'Authorization': `Bearer ${LSP_TOKEN}`, 'Content-Type': 'application/json' },
        body: JSON.stringify({ event_type: 'user_ok', device_id: 'luma_001' })
      });
    });

    // Step 3: 等待 30 秒确认窗口结束
    await page.waitForTimeout(30000);

    // Step 4: 验证 Family App 首页不显示红色预警（状态为 CALM）
    await page.goto(`${FAP_URL}/pages/home`);
    await page.waitForLoadState('networkidle');
    const alertBanner = page.locator('.alert-red-banner');
    await expect(alertBanner).not.toBeVisible();

    // Step 5: 验证系统生成了低级日志而非告警
    const resp = await page.request.get(`${API_BASE}/api/v1/alerts`, {
      headers: { 'Authorization': `Bearer ${FAP_TOKEN}` }
    });
    const alerts = resp.json().list;
    const redAlerts = alerts.filter((a: any) => a.level === 'red');
    expect(redAlerts).toHaveLength(0);
  });

  test('TC-E2E-09-03: 升级链路 T=30/50/70/90s 准确触发', async ({ page: _page }) => {
    // Mock 雷达跌倒 + 老人无应答，验证升级链路时间准确性
    const startTime = Date.now();
    await page.request.post(`${API_BASE}/api/v1/device/event`, {
      data: { device_id: 'radar_001', event_type: 'fall_detected', payload: { fall_confidence: 0.92 } }
    });

    const checkEscalation = async (expectedSecond: number) => {
      await page.waitForTimeout(expectedSecond * 1000 + 2000); // 允许 ±2s 误差
      const resp = await page.request.get(`${API_BASE}/api/v1/alerts/latest/escalations`, {
        headers: { 'Authorization': `Bearer ${ADMIN_TOKEN}` }
      });
      const escalations = resp.json().list;
      const relevant = escalations.filter((e: any) =>
        e.scheduled_offset_s === expectedSecond ||
        Math.abs(e.scheduled_offset_s - expectedSecond) <= 2
      );
      return { expected: expectedSecond, found: relevant.length > 0 };
    };

    const r30 = await checkEscalation(30);
    const r50 = await checkEscalation(50);
    const r70 = await checkEscalation(70);
    const r90 = await checkEscalation(90);

    expect(r30.found).toBe(true);
    expect(r50.found).toBe(true);
    expect(r70.found).toBe(true);
    expect(r90.found).toBe(true);
  });

  test('TC-E2E-09-04: 家属标记误报 → 数据上报用于算法优化', async ({ page }) => {
    // 先触发一次真实红色预警（复用 TC-E2E-09-01 步骤）
    await triggerFakeAlert(page);

    // 家属进入预警详情页
    await page.goto(`${FAP_URL}/pages/home`);
    await page.click('.alert-card');
    await page.waitForTimeout(1000);

    // 点击"标记为误报"
    await page.click('button:has-text("标记为误报")');

    // 选择误报原因
    await page.click('text=捡东西');

    // 提交
    await page.click('button:has-text("确认提交")');
    await page.waitForTimeout(2000);

    // 验证后端存储了误报原因
    const resp = await page.request.get(`${API_BASE}/api/v1/alerts/latest`, {
      headers: { 'Authorization': `Bearer ${FAP_TOKEN}` }
    });
    const alert = resp.json();
    expect(alert.status).toBe('false_alarm');
    expect(alert.false_alarm_reason).toBe('捡东西');
  });

  // 辅助函数：触发一次虚假红色预警（供测试复用）
  async function triggerFakeAlert(page: Page) {
    await page.request.post(`${API_BASE}/api/v1/device/event`, {
      data: { device_id: 'radar_001', event_type: 'fall_detected', payload: { fall_confidence: 0.92 } }
    });
    await page.waitForTimeout(35000); // 等待升级链路完成
  }
});

3.2 设备心跳与离线告警（US-21）

// tests/e2e/device_heartbeat.spec.ts

test.describe('US-21 设备心跳监控与离线告警', () => {

  test('TC-E2E-21-01: 设备心跳正常上报 → 保持在线', async ({ page }) => {
    // 模拟设备持续心跳上报
    for (let i = 0; i < 5; i++) {
      await page.request.post(`${API_BASE}/api/v1/device/heartbeat`, {
        data: { device_id: 'luma_001', timestamp: Date.now() }
      });
      await page.waitForTimeout(1000);
    }

    // 验证设备状态为在线
    const resp = await page.request.get(`${API_BASE}/api/v1/device/luma_001/status`, {
      headers: { 'Authorization': `Bearer ${FAP_TOKEN}` }
    });
    const status = resp.json();
    expect(status.status).toBe('online');
    expect(status.last_heartbeat).toBeDefined();
  });

  test('TC-E2E-21-02: 设备离线 > 12小时 → 黄色预警', async ({ page }) => {
    // 模拟设备心跳中断 12 小时（通过后端测试工具直接修改 Redis）
    await page.evaluate(async () => {
      // 直接设置 Redis 心跳过期（仅测试环境）
      await fetch(`${API_BASE}/test/mock-heartbeat-expire?device_id=luma_001&hours=13`);
    });

    // 触发离线检测任务
    await page.request.post(`${API_BASE}/api/v1/cron/check-offline`);

    // 等待告警生成
    await page.waitForTimeout(5000);

    // 验证 Family App 收到黄色预警
    await page.goto(`${FAP_URL}/pages/home`);
    await page.waitForLoadState('networkidle');
    const amberAlert = page.locator('.alert-amber');
    await expect(amberAlert).toBeVisible();
    await expect(amberAlert).toContainText(/离线|12小时/);
  });

  test('TC-E2E-21-03: 设备离线 > 48小时 → 红色预警 + Admin 高风险标记', async ({ page }) => {
    // 模拟设备心跳中断 48 小时
    await page.evaluate(async () => {
      await fetch(`${API_BASE}/test/mock-heartbeat-expire?device_id=luma_001&hours=49`);
    });
    await page.request.post(`${API_BASE}/api/v1/cron/check-offline`);
    await page.waitForTimeout(5000);

    // 验证红色预警生成
    const alertResp = await page.request.get(`${API_BASE}/api/v1/alerts/latest`, {
      headers: { 'Authorization': `Bearer ${FAP_TOKEN}` }
    });
    const alert = alertResp.json();
    expect(alert.level).toBe('red');
    expect(alert.reason).toContain('离线');

    // 验证 Admin 后台高风险标记
    await page.goto(`${ADMIN_URL}/family/high-risk`);
    await page.waitForLoadState('networkidle');
    const highRiskBadge = page.locator('.badge-high-risk');
    await expect(highRiskBadge).toBeVisible();
  });
});

3.3 离线 SOS 兜底（US-22）

// tests/e2e/offline_sos.spec.ts

test.describe('US-22 离线状态安全兜底', () => {

  test('TC-E2E-22-01: 网络断开 → Luma 温和告知离线状态', async ({ page }) => {
    // 模拟 Luma 网络断开
    await page.evaluate(async () => {
      await fetch(`${API_BASE}/test/mock-device-offline?device_id=luma_001`);
    });

    // 等待 Luma 屏幕更新
    await page.waitForTimeout(3000);

    // 验证 Luma 屏幕显示离线状态
    // 注意：此场景需要 Luma App 真机或 Android 模拟器
    // 自动化测试通过 WebSocket 连接状态判断
    const wsResp = await page.evaluate(async () => {
      return new Promise((resolve) => {
        const ws = new WebSocket(`${WS_BASE}/device/luma_001`);
        ws.onclose = () => resolve({ connected: false, code: ws.readyState });
        ws.onerror = () => resolve({ connected: false, error: true });
        setTimeout(() => resolve({ connected: false, timeout: true }), 5000);
      });
    });

    // 验证 WebSocket 断开
    expect(wsResp.connected).toBe(false);
  });

  test('TC-E2E-22-02: 离线触发 SOS → 播放 911 引导提示', async ({ page }) => {
    // 模拟 Luma 离线
    await page.evaluate(async () => {
      await fetch(`${API_BASE}/test/mock-device-offline?device_id=luma_001`);
    });

    // 模拟老人触发 SOS（通过后端 API 模拟）
    await page.request.post(`${API_BASE}/api/v1/device/sos`, {
      data: { device_id: 'luma_001', trigger: 'button_offline' }
    });

    // 验证后端不发送 Twilio 告警（离线状态不联网）
    const alertResp = await page.request.get(`${API_BASE}/api/v1/alerts/latest`, {
      headers: { 'Authorization': `Bearer ${ADMIN_TOKEN}` }
    });
    const alert = alertResp.json();
    // 离线 SOS 不生成 RED 告警，而是触发本地 911 提示逻辑
    expect(alert.status).toBe('offline_sos_local');
  });

  test('TC-E2E-22-03: 网络恢复 → 补发离线期间缓存事件', async ({ page }) => {
    // Step 1: 模拟离线期间老人触发 SOS，事件缓存
    await page.evaluate(async () => {
      await fetch(`${API_BASE}/test/mock-device-offline?device_id=luma_001`);
      await fetch(`${API_BASE}/api/v1/device/sos`, {
        data: { device_id: 'luma_001', trigger: 'button_offline', cached: true }
      });
    });

    // Step 2: 模拟网络恢复
    await page.evaluate(async () => {
      await fetch(`${API_BASE}/test/mock-device-online?device_id=luma_001`);
    });

    // Step 3: 等待补发任务执行
    await page.waitForTimeout(10000);

    // Step 4: 验证离线事件已补发
    const eventResp = await page.request.get(`${API_BASE}/api/v1/device/events`, {
      params: { device_id: 'luma_001', type: 'offline_sos', from_cache: true },
      headers: { 'Authorization': `Bearer ${ADMIN_TOKEN}` }
    });
    const events = eventResp.json().list;
    expect(events.length).toBeGreaterThanOrEqual(1);
    expect(events[0].from_cache).toBe(true);
  });
});

3.4 Family App 核心页面流程

技术栈：UniApp + Vue 3 + TypeScript（health.reflekt.app），一套代码编译至 iOS / Android 双端（参照 software_endpoints.md §2.1）

Playwright 测试运行于 H5 预览模式（pnpm dev:h5），真实双端编译测试通过 pnpm build:app-ios / pnpm build:app-android 输出的 .ipa/.apk 进行真机验证。

// tests/e2e/fap_core_flows.spec.ts

test.describe('Family App 核心页面流程', () => {

  test('TC-E2E-FAP-01: 新用户注册 → 同意隐私政策 → 登录 → 首页', async ({ page }) => {
    await page.goto(`${FAP_URL}/pages/auth/login`);
    await page.click('button:has-text("注册账号")');

    // 填写注册信息
    await page.fill('input[type="tel"]', '13800138001');
    await page.click('button:has-text("获取验证码")');
    await page.waitForTimeout(2000);

    // Mock 验证码（测试环境）
    await page.evaluate(() => {
      localStorage.setItem('test_otp', '123456');
    });
    await page.fill('input[placeholder*="验证码"]', '123456');
    await page.fill('input[type="password"]', 'Test@123456');
    await page.click('button:has-text("注册")');

    // 隐私政策弹窗
    await page.waitForSelector('.privacy-modal', { timeout: 5000 });
    await page.click('.privacy-modal button:has-text("同意")');

    // 验证跳转至首页
    await page.waitForURL(`${FAP_URL}/pages/home`, { timeout: 10000 });
    await expect(page.locator('.home-header')).toBeVisible();
  });

  test('TC-E2E-FAP-02: 首页状态切换 → CALM / ALERT / OFFLINE', async ({ page }) => {
    await page.goto(`${FAP_URL}/pages/home`);
    await page.waitForLoadState('networkidle');

    // Mock 不同状态
    const states = ['calm', 'alert', 'offline', 'reminder'];
    for (const state of states) {
      await page.evaluate((s) => {
        localStorage.setItem('mock_home_state', s);
      }, state);
      await page.reload();
      await page.waitForLoadState('networkidle');

      // 验证对应状态卡片显示
      const stateCard = page.locator(`.state-${state}`);
      await expect(stateCard).toBeVisible({ timeout: 5000 });
    }
  });

  test('TC-E2E-FAP-03: 发送语音消息 → Luma 接收并播报', async ({ page }) => {
    await page.goto(`${FAP_URL}/pages/home`);
    await page.click('.quick-send-btn');

    // 选择语音消息
    await page.click('text=语音消息');

    // 模拟录制语音（Mock WebRTC）
    await page.evaluate(() => {
      // 测试环境直接设置 Mock 语音文件
      (window as any).__mockVoiceBlob = new Blob(['mock-audio'], { type: 'audio/mp3' });
    });

    await page.click('button:has-text("发送")');
    await page.waitForTimeout(3000);

    // 验证消息出现在消息列表
    const voiceMsg = page.locator('.message-voice');
    await expect(voiceMsg).toBeVisible();
    await expect(voiceMsg).toContainText('语音消息');
  });

  test('TC-E2E-FAP-04: 提醒创建 → 定时播报 → 老人确认 → 状态同步', async ({ page }) => {
    await page.goto(`${FAP_URL}/pages/reminder/create`);
    await page.waitForLoadState('networkidle');

    // 选择模板
    await page.click('text=用药提醒');

    // 设置提醒时间（未来 1 分钟）
    const futureTime = new Date(Date.now() + 60000);
    await page.fill('input[type="time"]', futureTime.toTimeString().slice(0, 5));

    // 保存
    await page.click('button:has-text("保存提醒")');
    await page.waitForTimeout(2000);

    // 验证提醒出现在提醒列表
    await page.goto(`${FAP_URL}/pages/reminder/list`);
    const reminderItem = page.locator('.reminder-item').first();
    await expect(reminderItem).toBeVisible();

    // Mock 老人确认提醒
    await page.evaluate(async () => {
      await fetch(`${API_BASE}/api/v1/reminder/confirm`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ reminder_id: 'test_reminder_001', status: 'confirmed' })
      });
    });

    // 验证状态同步为已确认
    await page.reload();
    await page.waitForLoadState('networkidle');
    const confirmedBadge = page.locator('.status-confirmed');
    await expect(confirmedBadge).toBeVisible();
  });
});

3.5 Luma App 核心页面流程

技术栈：原生 Android（Java + Kotlin），包名 health.reflekt.luma，1280×800px 触摸屏（参照 software_endpoints.md §2.2）

Luma App 为音箱端，测试分两层：

协议层：HTTP POST / WebSocket（通过 tests/device/test_ws_luma.py 验证，§4.5.3）

界面层：通过 Android 模拟器或真机运行 Web 前端预览（static/ui/lsp/index.html），验证屏幕状态和 TTS 播报

适老化验收：字号 ≥ 18pt、触控 ≥ 44dp×44dp、色彩对比度 WCAG AA（4.5:1）

// tests/e2e/lsp_core_flows.spec.ts

test.describe('Luma App 核心页面流程（老人端）', () => {

  test.beforeEach(async ({ page }) => {
    // Luma App 为音箱端，通过模拟 HTTP 接口测试屏幕状态
    await page.goto(`${LSP_URL}/index.html`);
    await page.waitForLoadState('networkidle');
  });

  test('TC-E2E-LSP-01: 首页 CALM 状态显示', async ({ page }) => {
    // Mock 设备在线且状态平稳
    await page.evaluate(async () => {
      await fetch(`${LSP_API}/api/v1/lsp/state`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ state: 'calm', device_id: 'luma_001' })
      });
    });
    await page.reload();
    await page.waitForTimeout(2000);

    const timeDisplay = page.locator('.time-header');
    await expect(timeDisplay).toBeVisible();

    const statusText = page.locator('.status-text');
    await expect(statusText).toContainText(/I'm right here|Calming/);
  });

  test('TC-E2E-LSP-02: ALERT 状态询问"Are you okay?"', async ({ page }) => {
    // Mock ALERT 状态触发
    await page.evaluate(async () => {
      await fetch(`${LSP_API}/api/v1/lsp/state`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ state: 'alert', device_id: 'luma_001' })
      });
    });
    await page.reload();
    await page.waitForTimeout(3000);

    const alertBanner = page.locator('.alert-banner');
    await expect(alertBanner).toBeVisible();

    const questionText = page.locator('.question-text');
    // 禁止词验证：不能出现 detected / emergency / SOS 等词
    const text = await questionText.textContent();
    expect(text).not.toMatch(/detected|emergency|sos|报警/i);
    await expect(questionText).toContainText(/are you okay|okay\?/i);

    // 验证适老化：字号 ≥ 18pt
    const fontSize = await questionText.evaluate((el) =>
      parseFloat(getComputedStyle(el).fontSize)
    );
    expect(fontSize).toBeGreaterThanOrEqual(18);
  });

  test('TC-E2E-LSP-03: OFFLINE 状态显示 + SOS 可用', async ({ page }) => {
    // Mock 设备离线
    await page.evaluate(async () => {
      await fetch(`${LSP_API}/api/v1/lsp/state`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ state: 'offline', device_id: 'luma_001' })
      });
    });
    await page.reload();
    await page.waitForTimeout(2000);

    const offlineBanner = page.locator('.offline-banner');
    await expect(offlineBanner).toBeVisible();
    await expect(offlineBanner).toContainText(/offline|离线|I'm still here/);

    // 验证 SOS 按钮仍可用（离线安全兜底）
    const sosButton = page.locator('.sos-button');
    await expect(sosButton).toBeVisible();
    await expect(sosButton).toBeEnabled();
  });

  test('TC-E2E-LSP-04: TTS 语速验证（慢 15-20%）', async ({ page }) => {
    // Mock TTS 播报请求
    const ttsRequest = await page.evaluate(async () => {
      const resp = await fetch(`${LSP_API}/api/v1/tts/synthesize`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ text: "It's almost eleven. I'll remind you when it's time." })
      });
      return resp.json();
    });

    // 验证 TTS 请求中包含语速参数
    // OpenAI TTS 的 speed 参数应 < 1.0（慢速）
    expect(ttsRequest.speed).toBeLessThanOrEqual(0.85);
    expect(ttsRequest.speed).toBeGreaterThan(0.80);
  });
});

4. 接口自动化测试（pytest）

后端技术栈：Dromara RuoYi-Vue-Plus（Spring Boot 3 + JDK 17 + MyBatis-Plus + Sa-Token），包名 health.reflekt.api，部署名 reflekt-health-api（参照 software_endpoints.md §2.4）

Admin 技术栈：Vue 3 + TypeScript + ElementPlus，构建命令 pnpm build:admin，输出目录 dist/admin/（参照 software_endpoints.md §2.3）

4.1 D1 设备接入服务

# tests/api/test_device_access.py
import pytest
import requests
from datetime import datetime

API_BASE = "https://api.reflekt.health/api/v1"

class TestDeviceAccessService:
    """D1 设备接入服务接口测试"""

    @pytest.fixture(autouse=True)
    def setup(self):
        self.token = get_test_token("device_access")
        self.headers = {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}

    def test_heartbeat_success(self):
        """设备心跳上报成功"""
        resp = requests.post(
            f"{API_BASE}/device/heartbeat",
            headers=self.headers,
            json={"device_id": "luma_001", "timestamp": int(datetime.now().timestamp() * 1000)}
        )
        assert resp.status_code == 200, f"心跳失败: {resp.text}"
        data = resp.json()
        assert data["code"] == 0
        assert data["data"]["status"] == "online"

    def test_heartbeat_device_not_found(self):
        """心跳上报：设备不存在"""
        resp = requests.post(
            f"{API_BASE}/device/heartbeat",
            headers=self.headers,
            json={"device_id": "nonexistent_device", "timestamp": int(datetime.now().timestamp() * 1000)}
        )
        assert resp.status_code == 404

    def test_watch_health_data_upload(self):
        """手表健康数据上传（埃微 API Mock）"""
        resp = requests.post(
            f"{API_BASE}/device/watch/data",
            headers=self.headers,
            json={
                "device_id": "watch_001",
                "timestamp": int(datetime.now().timestamp() * 1000),
                "heart_rate": 72,
                "blood_oxygen": 98,
                "steps": 3500,
            }
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        # 验证数据入库
        query_resp = requests.get(
            f"{API_BASE}/device/watch_001/health/latest",
            headers=self.headers
        )
        assert query_resp.json()["data"]["heart_rate"] == 72

    def test_radar_fall_event_mqtt_mock(self):
        """雷达跌倒事件上报（MQTT → D1 消费验证）"""
        resp = requests.post(
            f"{API_BASE}/test/inject-radar-event",
            headers=self.headers,
            json={
                "device_id": "radar_001",
                "event_type": "fall_detected",
                "payload": {"fall_confidence": 0.92, "distance": 1.8}
            }
        )
        assert resp.status_code == 200
        # 验证 D1 消费了事件（通过事件表查询）
        event_resp = requests.get(
            f"{API_BASE}/device/events/latest",
            headers=self.headers,
            params={"device_id": "radar_001", "type": "fall"}
        )
        events = event_resp.json()["list"]
        assert len(events) >= 1
        assert events[0]["payload"]["fall_confidence"] == 0.92

    def test_offline_detection_after_12h(self):
        """设备离线检测（>12h 触发）"""
        # Mock 心跳超时
        requests.post(
            f"{API_BASE}/test/mock-heartbeat-expire",
            headers=self.headers,
            json={"device_id": "luma_001", "hours": 13}
        )
        # 触发检测任务
        resp = requests.post(
            f"{API_BASE}/cron/check-device-offline",
            headers=self.headers
        )
        assert resp.status_code == 200
        # 验证生成了 Amber 预警
        alert_resp = requests.get(
            f"{API_BASE}/alerts/latest",
            headers=self.headers
        )
        latest = alert_resp.json()
        assert latest["level"] in ["amber", "red"]

    def test_ws_heartbeat_maintains_connection(self):
        """WebSocket 心跳维持连接"""
        import websocket

        ws = websocket.create_connection(
            f"wss://api.reflekt.health/ws/device/luma_001",
            header={"Authorization": f"Bearer {self.token}"}
        )

        # 发送心跳
        ws.send(json.dumps({"type": "ping", "timestamp": int(datetime.now().timestamp() * 1000)}))
        result = ws.recv()
        ws.close()

        data = json.loads(result)
        assert data["type"] == "pong"

4.2 D2 家属端服务

# tests/api/test_family_api.py
import pytest
import requests

API_BASE = "https://api.reflekt.health/api/v1"

class TestFamilyAPIService:
    """D2 家属端服务接口测试"""

    @pytest.fixture(autouse=True)
    def setup(self):
        self.family_token = get_test_token("family_user")
        self.admin_token = get_test_token("admin")
        self.headers = {"Authorization": f"Bearer {self.family_token}", "Content-Type": "application/json"}

    # === 认证模块 ===

    def test_login_success(self):
        """正常登录"""
        resp = requests.post(
            f"{API_BASE}/auth/login",
            json={"phone": "13800138000", "password": "Test@123"}
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        assert "access_token" in data["data"]
        assert "refresh_token" in data["data"]

    def test_login_wrong_password(self):
        """密码错误"""
        resp = requests.post(
            f"{API_BASE}/auth/login",
            json={"phone": "13800138000", "password": "WrongPassword"}
        )
        assert resp.status_code in [200, 400]
        data = resp.json()
        assert data["code"] != 0
        assert "密码" in str(data) or "password" in str(data).lower()

    def test_sms_otp_rate_limit(self):
        """短信验证码频率限制（1 分钟内不重复发送）"""
        phone = f"138{random.randint(10000000, 99999999)}"
        # 第一次请求
        resp1 = requests.post(
            f"{API_BASE}/auth/send-otp",
            json={"phone": phone, "purpose": "register"}
        )
        assert resp1.status_code == 200
        # 30 秒后再次请求（应被限流）
        resp2 = requests.post(
            f"{API_BASE}/auth/send-otp",
            json={"phone": phone, "purpose": "register"}
        )
        assert resp2.status_code == 429 or resp2.json().get("code") == 429

    def test_consent_record_stored(self):
        """隐私同意记录存储"""
        resp = requests.post(
            f"{API_BASE}/consent",
            headers=self.headers,
            json={"policy_version": "v2.0", "action": "agree"}
        )
        assert resp.status_code == 200
        # 验证数据库记录
        db_resp = requests.get(
            f"{API_BASE}/consent/check",
            headers=self.headers
        )
        consent = db_resp.json()
        assert consent["data"]["status"] == "agreed"
        assert consent["data"]["policy_version"] == "v2.0"

    # === 家庭与设备 ===

    def test_family_device_list(self):
        """获取家庭设备列表"""
        resp = requests.get(
            f"{API_BASE}/family/devices",
            headers=self.headers
        )
        assert resp.status_code == 200
        data = resp.json()
        assert "list" in data["data"]
        for device in data["data"]["list"]:
            assert "device_id" in device
            assert "status" in device  # online / offline
            assert "last_heartbeat" in device

    def test_reminder_create_and_list(self):
        """提醒创建 → 列表查询"""
        # 创建提醒
        create_resp = requests.post(
            f"{API_BASE}/reminder",
            headers=self.headers,
            json={
                "content_type": "text",
                "content": "该喝水了",
                "schedule_time": "09:00",
                "frequency": "daily",
                "need_confirm": True
            }
        )
        assert create_resp.status_code == 200
        reminder_id = create_resp.json()["data"]["id"]

        # 查询提醒列表
        list_resp = requests.get(
            f"{API_BASE}/reminder/list",
            headers=self.headers
        )
        reminders = list_resp.json()["data"]["list"]
        ids = [r["id"] for r in reminders]
        assert reminder_id in ids

    # === 预警处理 ===

    def test_alert_ack_interrupt_escalation(self):
        """预警确认 → 中断升级链路"""
        # 先触发红色预警
        alert_id = trigger_test_alert()
        # 立即确认
        resp = requests.post(
            f"{API_BASE}/alert/{alert_id}/ack",
            headers=self.headers,
            json={"action": "acknowledged"}
        )
        assert resp.status_code == 200
        # 验证升级链路状态
        esc_resp = requests.get(
            f"{API_BASE}/alert/{alert_id}/escalations",
            headers=self.headers
        )
        escalations = esc_resp.json()["data"]["list"]
        # 已确认的升级应停止，不再有新升级触发
        for esc in escalations:
            if esc["status"] == "pending":
                assert esc["scheduled_at"] < datetime.now().timestamp() + 20  # 不应再安排新升级

4.3 D4 AI 大脑服务

# tests/api/test_ai_brain.py
import pytest
import requests

class TestAIBrainService:
    """D4 AI 大脑服务接口测试"""

    def test_fall_verdict_from_radar_event(self):
        """雷达跌倒事件 → AI综合判断"""
        resp = requests.post(
            f"{API_BASE}/ai/fall-verdict",
            headers={"Authorization": f"Bearer {AI_TOKEN}"},
            json={
                "device_id": "radar_001",
                "fall_confidence": 0.92,
                "heart_rate": 85,
                "elder_id": "elder_001"
            }
        )
        assert resp.status_code == 200
        data = resp.json()
        assert "verdict" in data["data"]
        assert data["data"]["verdict"] in ["escalate", "dismiss", "pending"]
        # 跌倒置信度 0.92 属于高置信，应进入 pending（等待 Luma 询问结果）
        if data["data"]["fall_confidence_input"] == 0.92:
            assert data["data"]["verdict"] in ["escalate", "pending"]

    def test_tts_speech_slow_pace(self):
        """TTS 合成：语速慢 15-20%"""
        resp = requests.post(
            f"{API_BASE}/ai/tts",
            headers={"Authorization": f"Bearer {AI_TOKEN}"},
            json={
                "text": "Are you okay? I'm here with you.",
                "elder_id": "elder_001"
            }
        )
        assert resp.status_code == 200
        data = resp.json()
        # 验证 TTS 参数中的 speed 参数
        assert "audio_url" in data["data"]
        # speed 应在 0.80-0.85 范围（慢 15-20%）
        assert data["data"].get("speed", 1.0) <= 0.85

    def test_tts_forbidden_words_check(self):
        """TTS 禁用词检测"""
        forbidden_phrases = [
            "I detected a fall",
            "Emergency SOS triggered",
            "Abnormal activity detected",
        ]
        for phrase in forbidden_phrases:
            resp = requests.post(
                f"{API_BASE}/ai/tts",
                headers={"Authorization": f"Bearer {AI_TOKEN}"},
                json={"text": phrase, "elder_id": "elder_001"}
            )
            # 应被拒绝或静默替换禁用词
            data = resp.json()
            if resp.status_code == 200:
                # 如果生成了音频，验证内容不含禁用词
                generated = data["data"].get("text", "")
                for fw in ["detected", "emergency", "sos", "abnormal"]:
                    assert fw.lower() not in generated.lower()

    def test_gpt_response_not_medical_advice(self):
        """GPT 对话：医疗问题返回免责话术"""
        medical_questions = [
            "我血压高怎么办",
            "I think I might have heart disease, what should I do?",
        ]
        for q in medical_questions:
            resp = requests.post(
                f"{API_BASE}/ai/chat",
                headers={"Authorization": f"Bearer {AI_TOKEN}"},
                json={"elder_id": "elder_001", "message": q}
            )
            data = resp.json()
            response_text = data["data"]["response"]
            # 必须包含免责关键词
            assert any(kw in response_text.lower() for kw in ["我不是医生", "not a doctor", "consult", "professional", "建议"]), \
                f"医疗问题未返回免责话术: {response_text}"
            # 不能包含诊断词
            assert not any(kw in response_text.lower() for kw in ["诊断", "diagnosis", "你应该", "you should take"]), \
                f"回复包含诊断建议: {response_text}"

    def test_user_ok_interrupts_alert(self):
        """USER_OK 信号 → 中断告警升级"""
        # 模拟一个进行中的告警
        alert_id = "test_alert_001"
        # 模拟 Luma 上报 USER_OK
        resp = requests.post(
            f"{API_BASE}/ai/voice-event",
            headers={"Authorization": f"Bearer {AI_TOKEN}"},
            json={
                "device_id": "luma_001",
                "event_type": "user_ok",
                "alert_id": alert_id
            }
        )
        assert resp.status_code == 200
        # 验证告警状态更新
        alert_resp = requests.get(
            f"{API_BASE}/alert/{alert_id}",
            headers={"Authorization": f"Bearer {ADMIN_TOKEN}"}
        )
        alert = alert_resp.json()
        assert alert["status"] in ["resolved", "dismissed"]

4.4 D5 通知告警服务

# tests/api/test_notification.py
import pytest
import requests
import time
from datetime import datetime

class TestNotificationService:
    """D5 通知告警服务接口测试"""

    def test_escalation_chain_timing(self):
        """升级链路时间准确性（T=30/50/70/90s）"""
        alert_id = trigger_test_red_alert()
        timings = [30, 50, 70, 90]
        tolerance = 2  # 允许 ±2s 误差

        for expected in timings:
            # 等待预期触发时间
            time.sleep(expected + tolerance + 1)
            resp = requests.get(
                f"{API_BASE}/alert/{alert_id}/escalations",
                headers={"Authorization": f"Bearer {ADMIN_TOKEN}"}
            )
            escalations = resp.json()["data"]["list"]
            found = False
            for esc in escalations:
                actual = esc.get("scheduled_offset_s", 0)
                if abs(actual - expected) <= tolerance:
                    found = True
                    assert esc["status"] in ["sent", "delivered"], f"T={expected}s 升级状态异常: {esc['status']}"
                    break
            assert found, f"T={expected}s 升级未触发！当前升级列表: {escalations}"

    def test_red_alert_silent_override(self):
        """红色预警静默穿透（无视静默时段）"""
        # 模拟设置了静默时段的家属
        requests.post(
            f"{API_BASE}/notification/quiet-hours",
            headers={"Authorization": f"Bearer {FAMILY_TOKEN}"},
            json={"start": "22:00", "end": "07:00", "enabled": True}
        )

        # 触发红色预警
        alert_id = trigger_test_red_alert()
        resp = requests.get(
            f"{API_BASE}/alert/{alert_id}/notifications",
            headers={"Authorization": f"Bearer {ADMIN_TOKEN}"}
        )
        notifications = resp.json()["data"]["list"]
        # 红色预警应强制推送，无视静默时段
        red_notifications = [n for n in notifications if n["priority"] == "high"]
        assert len(red_notifications) > 0, "红色预警未强制穿透静默设置"

    def test_duplicate_alert_deduplication(self):
        """重复告警去重（短时间内同一设备相同类型告警）"""
        # 短时间内连续触发两次跌倒告警
        trigger_test_alert()
        time.sleep(2)
        resp2 = trigger_test_alert()

        # 验证只有一条告警（去重生效）
        list_resp = requests.get(
            f"{API_BASE}/alerts",
            headers={"Authorization": f"Bearer {FAMILY_TOKEN}"},
            params={"device_id": "radar_001", "type": "fall"}
        )
        fall_alerts = list_resp.json()["data"]["list"]
        # 同一设备 5 分钟内的跌倒告警应合并为一条
        assert len(fall_alerts) <= 2  # 允许 1-2 条（去重不完美但大幅减少）

4.5 设备端协议测试（MQTT / HTTP / WebSocket）

参照 software_endpoints.md §3，设备端通过三种协议接入后端 D1 服务：MQTT（雷达）、HTTP（手表）、HTTP+WebSocket（音箱）。

4.5.1 MQTT 雷达接入（Amazon MQ / RabbitMQ）

测试项	测试方法	验收标准	脚本
MQTT 连接建立（TLS）	paho-mqtt 客户端连接 Amazon MQ	连接成功，收到 CONNACK	`tests/device/test_mqtt_radar.py`
跌倒事件上报（正确 Topic）	模拟雷达上报 `/sys/{pk}/{dn}/thing/event/property/post`	事件消费成功，D4 收到事件	`tests/device/test_mqtt_radar.py`
雷达存在感数据上报	模拟 5Hz 原始数据	后端过滤，事件表无冗余记录	`tests/device/test_mqtt_radar.py`
MQTT 心跳维持（60s interval）	长时间连接保持	60s 内无断连，断连后自动重连	`tests/device/test_mqtt_reconnect.py`
MQTT 断连重连（RISK-01 缓解）	主动断开连接，等待重连	重连成功，离线事件缓存补发	`tests/device/test_mqtt_reconnect.py`

# tests/device/test_mqtt_radar.py
import paho.mqtt.client as mqtt
import ssl
import json
import pytest

AMAZON_MQ_ENDPOINT = "b-xxxx.mq.us-east-1.amazonaws.com"
AMAZON_MQ_PORT = 8883  # TLS
AMAZON_MQ_USERNAME = "test_radar"
AMAZON_MQ_PASSWORD = "test_password"
PRODUCT_KEY = "reflekt_radar"
DEVICE_NAME = "radar_001"

class TestMQTTRadarIntegration:
    """毫米波雷达 MQTT 接入测试（参照 software_endpoints.md §3.1）"""

    def test_mqtt_tls_connection(self):
        """MQTT TLS 连接建立"""
        client = mqtt.Client(client_id=f"test_{DEVICE_NAME}")
        client.tls_set(
            cert_reqs=ssl.CERT_REQUIRED,
            tls_version=ssl.PROTOCOL_TLS_CLIENT
        )
        client.username_pw_set(AMAZON_MQ_USERNAME, AMAZON_MQ_PASSWORD)

        connected = [False]
        def on_connect(client, userdata, flags, rc):
            connected[0] = True

        client.on_connect = on_connect
        client.connect(AMAZON_MQ_ENDPOINT, AMAZON_MQ_PORT, keepalive=60)
        client.loop_start()
        client.loop_stop()

        assert connected[0], "MQTT TLS 连接失败"
        assert client._sock is not None, "TLS 握手未完成"

    def test_fall_event_publish_and_consume(self):
        """跌倒事件上报 → D1 消费 → D4 收到"""
        # 雷达上报 Topic
        topic = f"/sys/{PRODUCT_KEY}/{DEVICE_NAME}/thing/event/property/post"
        payload = json.dumps({
            "deviceId": DEVICE_NAME,
            "timestamp": 1713600000000,
            "eventType": "fall_detected",
            "data": {
                "fallConfidence": 0.92,
                "inRoom": True,
                "distance": 1.8,
                "angle": 30,
                "height": 0.5
            }
        })

        # 通过 HTTP Mock API 注入 MQTT 消息（测试环境）
        resp = requests.post(
            f"{API_BASE}/test/inject-mqtt",
            headers={"Authorization": f"Bearer {DEVICE_TOKEN}"},
            json={"topic": topic, "payload": payload}
        )
        assert resp.status_code == 200

        # 等待 D1 消费（延迟约 1-2s）
        time.sleep(3)

        # 验证 D1 事件入库
        event_resp = requests.get(
            f"{API_BASE}/device/events/latest",
            headers={"Authorization": f"Bearer {DEVICE_TOKEN}"},
            params={"device_id": DEVICE_NAME, "type": "fall"}
        )
        events = event_resp.json()["list"]
        fall_events = [e for e in events if e["event_type"] == "fall_detected"]
        assert len(fall_events) >= 1, "MQTT 跌倒事件未被 D1 消费"
        assert fall_events[0]["payload"]["fallConfidence"] == 0.92

    def test_mqtt_reconnect_on_disconnect(self):
        """MQTT 断连自动重连（RISK-01 缓解验证）"""
        client = mqtt.Client(client_id=f"test_reconnect_{DEVICE_NAME}")
        client.username_pw_set(AMAZON_MQ_USERNAME, AMAZON_MQ_PASSWORD)

        connected_events = []
        disconnected_events = []

        def on_connect(client, userdata, flags, rc):
            connected_events.append(time.time())
        def on_disconnect(client, userdata, rc):
            disconnected_events.append(time.time())

        client.on_connect = on_connect
        client.on_disconnect = on_disconnect
        client.connect(AMAZON_MQ_ENDPOINT, AMAZON_MQ_PORT, keepalive=60)
        client.loop_start()

        time.sleep(2)
        assert len(connected_events) >= 1

        # 主动断开
        client.disconnect()
        time.sleep(1)
        assert len(disconnected_events) >= 1

        # 等待自动重连（paho-mqtt 内置自动重连）
        time.sleep(5)
        assert len(connected_events) >= 2, "MQTT 未自动重连"

        client.loop_stop()

4.5.2 HTTP 手表接入（埃微 API）

测试项	技术栈	测试方法	验收标准
手表健康数据上报（HTTP POST）	HTTP	模拟手表 POST 心率/血氧/步数	数据入库，API 限流 10次/小时触发退避
手表实时测量指令下发	HTTP	模拟后端下发测量指令	手表收到指令并回传结果
手表 HTTP 鉴权	HTTP	Token 鉴权 + 设备 ID 校验	无效 Token 返回 401
埃微 API 超限退避（RISK-02）	HTTP	Mock 触发 10次/小时限流	指数退避重试，不丢失数据

# tests/device/test_http_watch.py
import requests
import time

class TestWatchHTTPIntegration:
    """智能手表 HTTP 接入测试（参照 software_endpoints.md §3.2）"""

    @pytest.fixture(autouse=True)
    def setup(self):
        self.watch_token = get_test_token("watch_device")
        self.headers = {
            "Authorization": f"Bearer {self.watch_token}",
            "Content-Type": "application/json",
            "X-Device-Id": "watch_001"
        }

    def test_health_data_upload_normal_mode(self):
        """手表定时上报健康数据（正常模式，15-30min间隔）"""
        resp = requests.post(
            f"{API_BASE}/device/watch/data",
            headers=self.headers,
            json={
                "device_id": "watch_001",
                "timestamp": int(time.time() * 1000),
                "heart_rate": 72,
                "blood_oxygen": 98,
                "steps": 3500,
                "sleep_minutes": 420,
                "battery_level": 65
            }
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        assert data["data"]["stored"] is True

        # 验证入库
        query = requests.get(
            f"{API_BASE}/device/watch_001/health/latest",
            headers=self.headers
        )
        health = query.json()["data"]
        assert health["heart_rate"] == 72
        assert health["blood_oxygen"] == 98

    def test_watch_realtime_measurement_command(self):
        """后端下发实时测量指令 → 手表回传结果"""
        # 模拟后端下发测量指令
        cmd_resp = requests.post(
            f"{API_BASE}/device/watch/watch_001/measure",
            headers={"Authorization": f"Bearer {ADMIN_TOKEN}"},
            json={"measure_type": "heart_rate"}
        )
        assert cmd_resp.status_code == 200
        cmd_id = cmd_resp.json()["data"]["command_id"]

        # 模拟手表回传测量结果
        result_resp = requests.post(
            f"{API_BASE}/device/watch/result",
            headers=self.headers,
            json={
                "command_id": cmd_id,
                "device_id": "watch_001",
                "heart_rate": 88,  # 实时测量值
                "measurement_time": int(time.time() * 1000)
            }
        )
        assert result_resp.status_code == 200

    def test_ewatris_api_rate_limit_exponential_backoff(self):
        """埃微 API 限流 → 指数退避重试（RISK-02 缓解）"""
        # Mock：连续 10 次请求，第 11 次触发限流
        responses = []
        for i in range(12):
            resp = requests.post(
                f"{API_BASE}/device/watch/data",
                headers=self.headers,
                json={
                    "device_id": "watch_001",
                    "timestamp": int(time.time() * 1000),
                    "heart_rate": 70 + i,
                }
            )
            responses.append(resp.status_code)
            time.sleep(0.1)

        # 前 10 次应成功，第 11 次触发限流
        assert responses[:10].count(200) >= 9, "前 10 次请求不应失败"
        # 第 11 次之后应有退避行为（429 或 503）
        late_failures = [r for r in responses[10:] if r >= 400]
        assert len(late_failures) >= 1, "超限后未触发限流响应"

4.5.3 WebSocket 音箱接入（Luma HTTP + WebSocket）

测试项	技术栈	测试方法	验收标准
WebSocket 连接建立（Token 鉴权）	WebSocket	带 Token 建立 WSS 连接	连接成功，收到认证 ACK
WebSocket 心跳维持（30s 间隔）	WebSocket	30s 内 ping/pong	连接不断开
TTS 指令下发（HTTP POST）	HTTP POST	Mock D4 下发 TTS 指令	Luma 收到指令，音频 URL 有效
SOS 事件上报（WebSocket）	WebSocket	模拟老人触发 SOS	D1 收到事件，D5 触发升级链路
语音对话请求（HTTP POST）	HTTP POST	模拟 ASR 文本上传	D4 收到请求，返回 GPT 对话
语音留言上传（HTTP POST → S3）	HTTP + S3	Mock Family App 发送语音留言	文件上传 S3，Luma 可下载播放

# tests/device/test_ws_luma.py
import websocket
import json
import time
import requests

class TestLumaWebSocketIntegration:
    """Luma 智能音箱 HTTP + WebSocket 接入测试（参照 software_endpoints.md §3.3）"""

    LUMA_WS_URL = "wss://api.reflekt.health/ws/device/luma_001"
    LUMA_HTTP_BASE = "https://api.reflekt.health/api/v1"

    def test_ws_connection_with_token_auth(self):
        """WebSocket 带 Token 鉴权连接"""
        ws = websocket.create_connection(
            self.LUMA_WS_URL,
            header={"Authorization": f"Bearer {LUMA_TOKEN}"}
        )

        # 接收认证结果
        result = ws.recv()
        data = json.loads(result)
        ws.close()

        assert data["type"] == "auth_success", f"WebSocket 认证失败: {data}"
        assert data["device_id"] == "luma_001"

    def test_ws_invalid_token_rejected(self):
        """无效 Token 拒绝 WebSocket 连接"""
        with pytest.raises(websocket.WebSocketBadStatusException) as exc_info:
            ws = websocket.create_connection(
                self.LUMA_WS_URL,
                header={"Authorization": "Bearer invalid_token"}
            )
            ws.close()

        assert exc_info.value.status_code in [401, 403]

    def test_ws_heartbeat_30s_interval(self):
        """WebSocket 心跳 30s 间隔维持连接"""
        ws = websocket.create_connection(
            self.LUMA_WS_URL,
            header={"Authorization": f"Bearer {LUMA_TOKEN}"}
        )
        ws.settimeout(35)  # 略大于 30s 心跳间隔

        # 等待约 30s
        time.sleep(32)

        # 发送 ping
        ws.send(json.dumps({"type": "ping", "timestamp": int(time.time() * 1000)}))
        result = ws.recv()
        pong = json.loads(result)

        ws.close()
        assert pong["type"] == "pong", "WebSocket 心跳响应异常"
        assert pong["connection_status"] == "active"

    def test_tts_command_via_http(self):
        """D4 AI 大脑通过 HTTP 下发 TTS 指令至 Luma"""
        # Mock D4 生成 TTS 请求
        resp = requests.post(
            f"{self.LUMA_HTTP_BASE}/ai/tts",
            headers={"Authorization": f"Bearer {AI_TOKEN}"},
            json={
                "text": "Good morning, Mary. It's a beautiful day.",
                "elder_id": "elder_001",
                "device_id": "luma_001",
                "priority": "normal"
            }
        )
        assert resp.status_code == 200
        data = resp.json()
        assert "audio_url" in data["data"]
        assert data["data"]["speed"] <= 0.85, "TTS 语速未按规范慢 15-20%"

        # 验证音频 URL 可访问
        audio_resp = requests.head(data["data"]["audio_url"])
        assert audio_resp.status_code == 200

    def test_sos_event_via_ws(self):
        """老人通过 Luma 触发 SOS → WebSocket 上报 → D1 → D5"""
        ws = websocket.create_connection(
            self.LUMA_WS_URL,
            header={"Authorization": f"Bearer {LUMA_TOKEN}"}
        )

        # 模拟老人触发 SOS（通过 WebSocket 上报）
        ws.send(json.dumps({
            "type": "sos_trigger",
            "trigger": "voice",  # voice or button
            "timestamp": int(time.time() * 1000)
        }))

        # 接收后端 ACK
        ack = ws.recv()
        ack_data = json.loads(ack)
        ws.close()

        assert ack_data["type"] == "sos_acknowledged"
        assert ack_data["sos_id"] is not None

        # 验证 D5 触发升级链路
        time.sleep(5)
        alert_resp = requests.get(
            f"{self.LUMA_HTTP_BASE}/alerts/latest",
            headers={"Authorization": f"Bearer {ADMIN_TOKEN}"}
        )
        latest = alert_resp.json()
        assert latest["level"] == "red"
        assert latest["trigger"] in ["sos_voice", "sos_button"]

    def test_voice_message_upload_to_s3(self):
        """语音留言上传：Family App → HTTP POST → S3 → Luma 下载"""
        # Mock Family App 发送语音留言（multipart/form-data）
        files = {
            "audio": ("voice_msg.wav", open("tests/fixtures/voice_msg.wav", "rb"), "audio/wav"),
        }
        data = {
            "elder_id": "elder_001",
            "sender_id": "caregiver_001",
            "family_id": "family_001"
        }
        resp = requests.post(
            f"{self.LUMA_HTTP_BASE}/message/voice",
            headers={"Authorization": f"Bearer {FAMILY_TOKEN}"},
            files=files,
            data=data
        )
        assert resp.status_code == 200
        upload = resp.json()["data"]
        assert "s3_url" in upload

        # 验证 Luma 可通过 S3 URL 下载
        luma_download_resp = requests.get(
            f"{self.LUMA_HTTP_BASE}/device/luma_001/message/latest",
            headers={"Authorization": f"Bearer {LUMA_TOKEN}"}
        )
        luma_msg = luma_download_resp.json()["data"]
        assert luma_msg["type"] == "voice"
        assert luma_msg["audio_url"] == upload["s3_url"]

5. 验收矩阵：功能 × 方法 × 责任人

5.1 P0 硬底线功能测试覆盖

P0 功能	功能描述	性能测试	安全测试	E2E测试	API自动化	验收负责人
P0-1 跌倒告警闭环	雷达检测 → AI判断 → Luma询问 → 升级链路 → FAP推送	✅ 端到端延迟压测	✅ 升级链路幂等	✅ TC-E2E-09 系列	✅ D1+D4+D5	妙锋
P0-2 通知分级与静默穿透	Red/Amber/Normal 三级，红色强制穿透	✅ 推送送达率	✅ Token鉴权	✅ 静默时段E2E	✅ D5 API	妙锋
P0-3 设备心跳监控	心跳上报成功率 ≥ 99.5%	✅ 心跳压测	✅ 越权访问	✅ TC-E2E-21	✅ D1 API	妙锋
P0-4 离线SOS兜底	断网 → 本地缓存 → 补发 → 911引导	✅ 网络切换测试	✅ 数据完整性	✅ TC-E2E-22	✅ LSP离线API	妙锋
P0-5 多照护人权限	管理员/查看员权限分级，紧急联系顺序	✅ 并发通知	✅ 越权访问	✅ 成员管理E2E	✅ D2 API	妙锋
P0-6 审计日志	所有敏感操作全记录，保留 ≥ 12个月	—	✅ HIPAA审计覆盖	✅ Admin操作审计	✅ D3 API	妙锋
P0-7 高龄化UI标准	字号/触控/对比度适老化	—	—	✅ TC-E2E-LSP-02	—	佩宜（目测）

5.2 非功能指标验收表

指标	目标值	测试方法	验收标准	测试结果记录
API 查询响应时间 P95	< 200ms	JMeter 压测	100 并发，P95 < 200ms	`tests/performance/report/api_p95.md`
API 写入响应时间 P95	< 500ms	JMeter 压测	50 并发，P95 < 500ms	`tests/performance/report/write_p95.md`
红色预警端到端延迟	< 5 分钟（300s）	JMeter E2E 压测	全链路 P99 ≤ 280s	`tests/performance/report/e2e_latency.md`
设备心跳上报成功率	≥ 99.5%	JMeter 心跳压测	50 QPS × 10min，成功率 ≥ 99.5%	`tests/performance/report/heartbeat_success.md`
核心交易吞吐量	≥ 100 TPS	k6 持续压测	10min 稳态 ≥ 100 TPS	`tests/performance/report/throughput.md`
系统可用性 SLA	99.9%	持续监控	月度 uptime ≥ 99.9%	CloudWatch Dashboard
误报率	< 15%	真实数据统计	月度误报 / 总告警 < 15%	Admin 试点看板

5.3 测试环境配置

环境	用途	数据库	第三方 Mock
`dev`	开发自测	本地 PostgreSQL	Mock Server（本地）
`staging`	功能 + 集成测试	RDS dev（单AZ）	Twilio 沙箱 / OpenAI 测试配额
`prod`	投资人演示 / UAT	RDS prod（Multi-AZ）	正式第三方服务

# 测试环境启动命令
cd reflekt-health-api

# Mock Server（模拟雷达/手表/Luma 设备）
python tests/mock/mock_server.py --port 9999 &

# 运行全部 API 测试
pytest tests/api/ -v --tb=short

# 运行安全测试
pytest tests/security/ -v --tb=short

# 运行 E2E 测试
npx playwright test tests/e2e/ --reporter=list

# 运行性能测试
k6 run tests/performance/k6/throughput.js
jmeter -n -t tests/performance/jmeter/fall_to_alert_e2e.jmx -l results.jtl

6. 适老化专项验收

参照 ux_design_specification.md 规范和 user_stories.md US-16 适老化要求。

测试项	测试方法	验收标准	结果
Luma App 字号最小值	Playwright 截图 + 字体大小计算	≥ 18pt	待测试
Family App 字号最小值	Playwright 截图 + 字体大小计算	≥ 16pt	待测试
Luma App 触摸区域尺寸	Playwright `boundingBox()`	≥ 44dp × 44dp	待测试
颜色对比度	Lighthouse CI Accessibility Audit	WCAG AA（4.5:1）	待测试
TTS 语速	音频文件元数据读取	慢 15-20%（speed ≤ 0.85）	✅ TC-E2E-LSP-04
TTS 禁用词	API 响应内容审查	无 emergency/SOS/detected/abnormal	✅ test_tts_forbidden_words
文案禁用词（Family App）	页面文案正则扫描	无"监测你"/"跌倒检测"/"Missed"	待测试

7. 风险缓解专项测试

针对 technical_debt_and_risks.md 中识别的高风险项。

风险 ID	风险描述	缓解测试方法	验收标准
RISK-01	Amazon MQ 单连接心跳 60s	Mock MQTT 断连，验证本地缓存补发	断连 → 恢复后事件补发 100%
RISK-02	埃微 API 限流 10次/小时	模拟 10 次/小时限流，验证退避重试	超限后指数退避，不丢失数据
RISK-03	Twilio SMS 限流 10条/日	Mock SMS 超限，验证直拨降级通道	直拨通道优先，送达率 ≥ 95%
RISK-04	OpenAI 配额共用	Mock 429 错误，验证 TTS 降级	降级到固定话术，不崩溃
RISK-05	APNs/FCM 证书过期	Mock Push 失败，验证短信兜底	短信降级成功率 ≥ 99%
RISK-08	红色预警端到端 < 5min	`TC-E2E-09-03` 升级链路压测	P99 ≤ 280s

8. 测试数据规范

8.1 测试账号

账号类型	手机号	密码	用途
管理员	`13800000001`	`Admin@123`	Admin 后台操作
家属（管理员）	`13800000002`	`Family@123`	Family App 功能测试
家属（查看者）	`13800000003`	`Member@123`	权限测试
测试设备 Token	—	—	Mock 设备事件

8.2 Mock 数据工厂

// tests/utils/mock-data.ts
export function createMockRadarFallEvent(confidence = 0.92) {
  return {
    device_id: `radar_${Date.now()}`,
    event_type: 'fall_detected',
    timestamp: Date.now(),
    payload: {
      fall_confidence: confidence,
      distance: 1.8,
      angle: 30,
      height: 0.5,
      room: 'living_room',
    }
  };
}

export function createMockHeartRateData(bpm = 72) {
  return {
    device_id: `watch_${Date.now()}`,
    event_type: 'health_data',
    timestamp: Date.now(),
    payload: {
      heart_rate: bpm,
      blood_oxygen: 97 + Math.floor(Math.random() * 3),
      steps: 3000 + Math.floor(Math.random() * 2000),
      battery: 50 + Math.floor(Math.random() * 50),
    }
  };
}

9. 参考文档索引

参考文档	用途
`functional_requirements.md`	功能清单，Family App / Luma App / Admin / API 模块测试覆盖依据
`non_functional_requirements.md`	性能指标（API P95 < 200ms、E2E < 5min、SLA 99.9%）
`business_architecture.md`	跌倒告警闭环、升级链路 T=30/50/70/90s、六状态体系
`user_stories.md`	US-01~US-30 用户故事及内置测试用例
`technical_architecture.md`	D1-D5 微服务接口规格、Redis/PostgreSQL/S3 配置
`technical_debt_and_risks.md`	RISK-01~RISK-12 及风险缓解测试计划
`constraints_and_dependencies.md`	HIPAA 合规要求，审计日志保留规范
`ux_design_specification.md`	适老化硬指标（字号/对比度/禁用词）

本文档由 Reflekt Health 产品团队维护，最后更新于 2026-04-23。

使用说明​

1. 性能压测用例​

1.1 API 性能压测​

1.1.1 JMeter 脚本配置参考​

1.1.2 k6 吞吐测试脚本​

1.2 红色预警端到端延迟压测​

1.2.1 端到端延迟测试场景​

1.3 设备心跳监控压测​

2. 安全合规测试用例​

2.1 认证与授权安全​

2.1.1 SQL 注入测试脚本​

2.1.2 跨家庭越权访问测试脚本​

2.2 数据安全与 HIPAA 合规​

2.2.1 敏感字段加密验证脚本​

2.2.2 HIPAA 合规审计覆盖测试​

2.3 API 安全测试​

2.3.1 Sa-Token 认证机制专项测试​

3. 端到端测试用例（Playwright）​

3.1 跌倒告警闭环（US-09，核心链路）​

3.2 设备心跳与离线告警（US-21）​

3.3 离线 SOS 兜底（US-22）​

3.4 Family App 核心页面流程​

3.5 Luma App 核心页面流程​

4. 接口自动化测试（pytest）​

4.1 D1 设备接入服务​

4.2 D2 家属端服务​

4.3 D4 AI 大脑服务​

4.4 D5 通知告警服务​

4.5 设备端协议测试（MQTT / HTTP / WebSocket）​

4.5.1 MQTT 雷达接入（Amazon MQ / RabbitMQ）​

4.5.2 HTTP 手表接入（埃微 API）​

4.5.3 WebSocket 音箱接入（Luma HTTP + WebSocket）​

5. 验收矩阵：功能 × 方法 × 责任人​

5.1 P0 硬底线功能测试覆盖​

5.2 非功能指标验收表​

5.3 测试环境配置​

6. 适老化专项验收​

7. 风险缓解专项测试​

8. 测试数据规范​

8.1 测试账号​

8.2 Mock 数据工厂​

9. 参考文档索引​

使用说明

1. 性能压测用例

1.1 API 性能压测

1.1.1 JMeter 脚本配置参考

1.1.2 k6 吞吐测试脚本

1.2 红色预警端到端延迟压测

1.2.1 端到端延迟测试场景

1.3 设备心跳监控压测

2. 安全合规测试用例

2.1 认证与授权安全

2.1.1 SQL 注入测试脚本

2.1.2 跨家庭越权访问测试脚本

2.2 数据安全与 HIPAA 合规

2.2.1 敏感字段加密验证脚本

2.2.2 HIPAA 合规审计覆盖测试

2.3 API 安全测试

2.3.1 Sa-Token 认证机制专项测试

3. 端到端测试用例（Playwright）

3.1 跌倒告警闭环（US-09，核心链路）

3.2 设备心跳与离线告警（US-21）

3.3 离线 SOS 兜底（US-22）

3.4 Family App 核心页面流程

3.5 Luma App 核心页面流程

4. 接口自动化测试（pytest）

4.1 D1 设备接入服务

4.2 D2 家属端服务

4.3 D4 AI 大脑服务

4.4 D5 通知告警服务

4.5 设备端协议测试（MQTT / HTTP / WebSocket）

4.5.1 MQTT 雷达接入（Amazon MQ / RabbitMQ）

4.5.2 HTTP 手表接入（埃微 API）

4.5.3 WebSocket 音箱接入（Luma HTTP + WebSocket）

5. 验收矩阵：功能 × 方法 × 责任人

5.1 P0 硬底线功能测试覆盖

5.2 非功能指标验收表

5.3 测试环境配置

6. 适老化专项验收

7. 风险缓解专项测试

8. 测试数据规范

8.1 测试账号

8.2 Mock 数据工厂

9. 参考文档索引