import asyncio import json import os import sys import urllib.error import urllib.request import asyncpg BASE = os.getenv("BASE_URL", "http://localhost:8000") ADMIN_EMAIL = os.getenv("EMAIL", "admin@example.com") ADMIN_PASSWORD = os.getenv("PASSWORD", "admin123") STUDY_ID = os.getenv("STUDY_ID", "").strip() DATABASE_URL = (os.getenv("DATABASE_URL") or "postgresql://ctms_user:secret_password@db/ctms_db").replace("+asyncpg", "") def request_json(path: str, *, method: str = "GET", token: str | None = None, payload: dict | None = None) -> tuple[int, dict]: url = f"{BASE}{path}" headers = {} body = None if token: headers["Authorization"] = f"Bearer {token}" if payload is not None: headers["Content-Type"] = "application/json" body = json.dumps(payload).encode() req = urllib.request.Request(url, method=method, headers=headers, data=body) try: with urllib.request.urlopen(req, timeout=20) as resp: text = resp.read().decode() or "{}" return resp.getcode(), json.loads(text) except urllib.error.HTTPError as exc: text = exc.read().decode() or "{}" try: body_json = json.loads(text) except Exception: body_json = {"raw": text} return exc.code, body_json def assert_or_exit(condition: bool, message: str) -> None: if not condition: print(f"[FAIL] {message}") sys.exit(1) def extract_items(payload: dict) -> list[dict]: if isinstance(payload, list): return payload items = payload.get("items") if isinstance(items, list): return items return [] async def _db_fetch(sql: str, *args): conn = await asyncpg.connect(DATABASE_URL) try: rows = await conn.fetch(sql, *args) return [dict(row) for row in rows] finally: await conn.close() def db_fetch(sql: str, *args) -> list[dict]: return asyncio.run(_db_fetch(sql, *args)) def main() -> int: print(f"[config] BASE={BASE} EMAIL={ADMIN_EMAIL}") status, login = request_json( "/api/v1/auth/login", method="POST", payload={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD}, ) assert_or_exit(status == 200, f"登录失败 status={status} body={login}") token = login["access_token"] if STUDY_ID: study_id = STUDY_ID else: status, studies = request_json("/api/v1/studies/?skip=0&limit=1", token=token) assert_or_exit(status == 200, f"获取项目列表失败 status={status} body={studies}") items = studies.get("items") or [] assert_or_exit(bool(items), "未找到可用项目，请设置环境变量 STUDY_ID") study_id = items[0]["id"] print(f"[1/12] study_id={study_id}") status, study_before = request_json(f"/api/v1/studies/{study_id}", token=token) assert_or_exit(status == 200, f"获取项目详情失败 status={status} body={study_before}") status, sites_resp = request_json(f"/api/v1/studies/{study_id}/sites/?include_inactive=true&skip=0&limit=200", token=token) assert_or_exit(status == 200, f"获取中心列表失败 status={status} body={sites_resp}") sites = extract_items(sites_resp if isinstance(sites_resp, dict) else {"items": sites_resp}) active_sites = [item for item in sites if item.get("is_active") is True] assert_or_exit(len(active_sites) >= 1, "至少需要1个活跃中心以验证中心目标联动") status, cfg = request_json(f"/api/v1/studies/{study_id}/setup-config", token=token) assert_or_exit(status == 200, f"获取配置失败 status={status} body={cfg}") version = cfg["version"] print(f"[2/12] get_ok version={version}") payload = {"expected_version": version, "data": cfg["data"]} payload["data"]["projectMilestones"] = [ { "id": "smoke-project-ms-1", "name": "立项启动", "planDate": "2026-02-20", "owner": ADMIN_EMAIL, "remark": "project-ms-remark", "status": "进行中", } ] payload["data"]["enrollmentPlan"]["monthlyGoalNote"] = f"smoke-test-note-v{version}" payload["data"]["enrollmentPlan"]["stageBreakdown"] = f"smoke-stage-v{version}" payload["data"]["enrollmentPlan"]["totalTarget"] = max(1, int(study_before.get("planned_enrollment_count") or 0) + 7) payload["data"]["enrollmentPlan"]["startDate"] = "2026-03-01" payload["data"]["enrollmentPlan"]["endDate"] = "2026-09-30" payload["data"]["siteMilestones"] = [ { "id": "smoke-site-ms-1", "siteId": active_sites[0]["id"], "siteName": active_sites[0].get("name") or "", "milestone": "中心启动", "planDate": "2026-03-05", "owner": ADMIN_EMAIL, "remark": "site-ms-remark", } ] payload["data"]["siteEnrollmentPlans"] = [ { "id": f"smoke-plan-{idx+1}", "siteId": site["id"], "siteName": site.get("name") or "", "target": 10 if idx == 0 else 8, "startDate": "2026-03-01", "endDate": "2026-09-30", "note": "smoke-site-plan", } for idx, site in enumerate(active_sites[:2]) ] payload["data"]["monitoringStrategies"] = [ { "id": "smoke-monitoring-1", "strategyType": "风险监查", "detail": "基于关键风险触发", "frequency": "按触发", "updatedAt": "", "enabled": True, } ] payload["data"]["centerConfirm"] = [ { "id": "smoke-center-confirm-1", "siteId": active_sites[0]["id"], "siteName": active_sites[0].get("name") or "", "confirmer": "System Admin", "confirmStatus": "已确认", "confirmDate": "2026-03-15", "note": "smoke-center-confirm", } ] targeted_site_ids = {item["siteId"] for item in payload["data"]["siteEnrollmentPlans"]} before_site_map = {str(item.get("id")): item for item in sites if str(item.get("id")) in targeted_site_ids} status, updated = request_json(f"/api/v1/studies/{study_id}/setup-config", token=token, method="PUT", payload=payload) assert_or_exit(status == 200, f"保存草稿失败 status={status} body={updated}") latest_version = updated["version"] print(f"[3/12] put_ok version={latest_version}") status, study_after_save = request_json(f"/api/v1/studies/{study_id}", token=token) assert_or_exit(status == 200, f"保存后获取项目详情失败 status={status} body={study_after_save}") assert_or_exit( study_after_save.get("planned_enrollment_count") == study_before.get("planned_enrollment_count") and study_after_save.get("plan_start_date") == study_before.get("plan_start_date") and study_after_save.get("plan_end_date") == study_before.get("plan_end_date") and study_after_save.get("enrollment_monthly_goal_note") == study_before.get("enrollment_monthly_goal_note") and study_after_save.get("enrollment_stage_breakdown") == study_before.get("enrollment_stage_breakdown"), "保存草稿不应写入项目业务表字段", ) status, sites_after_save_resp = request_json( f"/api/v1/studies/{study_id}/sites/?include_inactive=true&skip=0&limit=200", token=token ) assert_or_exit(status == 200, f"保存后获取中心失败 status={status} body={sites_after_save_resp}") sites_after_save = extract_items(sites_after_save_resp if isinstance(sites_after_save_resp, dict) else {"items": sites_after_save_resp}) after_save_site_map = {str(item.get("id")): item for item in sites_after_save if str(item.get("id")) in targeted_site_ids} for site_id in targeted_site_ids: before_item = before_site_map.get(site_id) after_item = after_save_site_map.get(site_id) if not before_item or not after_item: continue assert_or_exit( after_item.get("enrollment_target") == before_item.get("enrollment_target") and after_item.get("enrollment_plan_start_date") == before_item.get("enrollment_plan_start_date") and after_item.get("enrollment_plan_end_date") == before_item.get("enrollment_plan_end_date") and after_item.get("enrollment_plan_note") == before_item.get("enrollment_plan_note"), f"保存草稿不应写入中心业务表字段 site={site_id}", ) print("[4/12] save_no_projection_ok") status, stale_result = request_json( f"/api/v1/studies/{study_id}/setup-config", token=token, method="PUT", payload={"expected_version": version, "data": updated["data"]}, ) stale_code = (stale_result.get("detail") or {}).get("code") if isinstance(stale_result.get("detail"), dict) else "" assert_or_exit(status == 409 and stale_code == "SETUP_CONFIG_VERSION_CONFLICT", f"冲突校验失败 status={status} body={stale_result}") print("[5/12] conflict_409_ok") invalid_payload = json.loads(json.dumps(updated)) invalid_payload["expected_version"] = latest_version invalid_payload["data"]["enrollmentPlan"]["startDate"] = "2026-12-31" invalid_payload["data"]["enrollmentPlan"]["endDate"] = "2026-01-01" status, invalid_result = request_json( f"/api/v1/studies/{study_id}/setup-config", token=token, method="PUT", payload={"expected_version": invalid_payload["expected_version"], "data": invalid_payload["data"]}, ) detail = invalid_result.get("detail") if isinstance(invalid_result, dict) else None err_code = detail.get("code") if isinstance(detail, dict) else "" err_list = detail.get("errors") if isinstance(detail, dict) else [] has_enrollment_date_error = any((item.get("field") == "enrollmentPlan.endDate") for item in (err_list or [])) assert_or_exit(status == 422 and err_code == "VALIDATION_ERROR" and has_enrollment_date_error, f"422校验失败 status={status} body={invalid_result}") print("[6/12] validation_422_ok") status, published = request_json( f"/api/v1/studies/{study_id}/setup-config/publish", token=token, method="POST", payload={"expected_version": latest_version}, ) assert_or_exit(status == 200 and published.get("publish_status") == "PUBLISHED", f"发布失败 status={status} body={published}") projection_status = str(published.get("projection_status") or "").lower() assert_or_exit( projection_status in {"success", "partial_success"}, f"发布联动状态异常 projection_status={published.get('projection_status')} body={published}", ) first_published_version = published["version"] print(f"[7/12] first_publish_ok version={first_published_version}") status, study_detail = request_json(f"/api/v1/studies/{study_id}", token=token) assert_or_exit(status == 200, f"获取项目详情失败 status={status} body={study_detail}") assert_or_exit( study_detail.get("planned_enrollment_count") == payload["data"]["enrollmentPlan"]["totalTarget"], f"项目计划入组联动失败 expected={payload['data']['enrollmentPlan']['totalTarget']} actual={study_detail.get('planned_enrollment_count')}", ) assert_or_exit( study_detail.get("plan_start_date") == payload["data"]["enrollmentPlan"]["startDate"] and study_detail.get("plan_end_date") == payload["data"]["enrollmentPlan"]["endDate"], f"项目计划日期联动失败 expected=({payload['data']['enrollmentPlan']['startDate']},{payload['data']['enrollmentPlan']['endDate']}) actual=({study_detail.get('plan_start_date')},{study_detail.get('plan_end_date')})", ) assert_or_exit( study_detail.get("enrollment_monthly_goal_note") == payload["data"]["enrollmentPlan"]["monthlyGoalNote"] and study_detail.get("enrollment_stage_breakdown") == payload["data"]["enrollmentPlan"]["stageBreakdown"], "项目入组补充说明联动失败", ) status, sites_after_resp = request_json(f"/api/v1/studies/{study_id}/sites/?include_inactive=true&skip=0&limit=200", token=token) assert_or_exit(status == 200, f"发布后获取中心失败 status={status} body={sites_after_resp}") sites_after = extract_items(sites_after_resp if isinstance(sites_after_resp, dict) else {"items": sites_after_resp}) site_target_map = {str(item.get("id")): item.get("enrollment_target") for item in sites_after} site_plan_start_map = {str(item.get("id")): item.get("enrollment_plan_start_date") for item in sites_after} site_plan_end_map = {str(item.get("id")): item.get("enrollment_plan_end_date") for item in sites_after} site_plan_note_map = {str(item.get("id")): item.get("enrollment_plan_note") for item in sites_after} for plan in payload["data"]["siteEnrollmentPlans"]: actual_target = site_target_map.get(str(plan["siteId"])) assert_or_exit( actual_target == plan["target"], f"中心目标联动失败 site={plan['siteId']} expected={plan['target']} actual={actual_target}", ) assert_or_exit( site_plan_start_map.get(str(plan["siteId"])) == plan["startDate"] and site_plan_end_map.get(str(plan["siteId"])) == plan["endDate"] and site_plan_note_map.get(str(plan["siteId"])) == plan["note"], f"中心入组计划附加字段联动失败 site={plan['siteId']}", ) monitoring_rows = db_fetch( """ SELECT strategy_type, detail, frequency, enabled FROM study_monitoring_strategies WHERE study_id = $1::uuid """, study_id, ) assert_or_exit(len(monitoring_rows) == 1, f"监查策略联动失败 rows={monitoring_rows}") assert_or_exit( monitoring_rows[0]["strategy_type"] == "风险监查" and monitoring_rows[0]["detail"] == "基于关键风险触发" and monitoring_rows[0]["frequency"] == "按触发" and monitoring_rows[0]["enabled"] is True, f"监查策略字段联动失败 rows={monitoring_rows}", ) center_confirm_rows = db_fetch( """ SELECT site_id::text AS site_id, confirmer, confirm_status, confirm_date::text AS confirm_date, note FROM study_center_confirms WHERE study_id = $1::uuid """, study_id, ) assert_or_exit(len(center_confirm_rows) == 1, f"中心确认联动失败 rows={center_confirm_rows}") assert_or_exit( center_confirm_rows[0]["site_id"] == active_sites[0]["id"] and center_confirm_rows[0]["confirm_status"] == "已确认" and center_confirm_rows[0]["confirm_date"] == "2026-03-15" and center_confirm_rows[0]["note"] == "smoke-center-confirm", f"中心确认字段联动失败 rows={center_confirm_rows}", ) milestone_rows = db_fetch( """ SELECT type, name, owner_id::text AS owner_id, owner_name, notes FROM milestones WHERE study_id = $1::uuid AND type IN ('SETUP_PROJECT_MILESTONE', 'SETUP_SITE_MILESTONE') ORDER BY type, name """, study_id, ) milestone_map = {(row["type"], row["name"]): row for row in milestone_rows} project_row = milestone_map.get(("SETUP_PROJECT_MILESTONE", "立项启动")) site_row = milestone_map.get(("SETUP_SITE_MILESTONE", "中心启动")) assert_or_exit(project_row is not None and site_row is not None, f"里程碑联动失败 rows={milestone_rows}") assert_or_exit( project_row.get("owner_name") == ADMIN_EMAIL and bool(project_row.get("owner_id")) and project_row.get("notes") == "project-ms-remark", f"项目里程碑owner结构化失败 row={project_row}", ) assert_or_exit( site_row.get("owner_name") == ADMIN_EMAIL and bool(site_row.get("owner_id")) and site_row.get("notes") == "site-ms-remark", f"中心里程碑owner结构化失败 row={site_row}", ) print("[8/12] publish_projection_db_assert_ok") payload_v2 = {"expected_version": published["version"], "data": published["data"]} payload_v2["data"]["enrollmentPlan"]["totalTarget"] = payload["data"]["enrollmentPlan"]["totalTarget"] + 3 payload_v2["data"]["enrollmentPlan"]["monthlyGoalNote"] = payload["data"]["enrollmentPlan"]["monthlyGoalNote"] + "-v2" if payload_v2["data"]["siteEnrollmentPlans"]: payload_v2["data"]["siteEnrollmentPlans"][0]["target"] = payload_v2["data"]["siteEnrollmentPlans"][0]["target"] + 2 status, updated_v2 = request_json( f"/api/v1/studies/{study_id}/setup-config", token=token, method="PUT", payload=payload_v2, ) assert_or_exit(status == 200, f"二次保存草稿失败 status={status} body={updated_v2}") status, published_v2 = request_json( f"/api/v1/studies/{study_id}/setup-config/publish", token=token, method="POST", payload={"expected_version": updated_v2["version"]}, ) assert_or_exit(status == 200 and published_v2.get("publish_status") == "PUBLISHED", f"二次发布失败 status={status} body={published_v2}") status, rolled = request_json( f"/api/v1/studies/{study_id}/setup-config/rollback", token=token, method="POST", payload={"expected_version": published_v2["version"], "target_version": first_published_version}, ) assert_or_exit(status == 200, f"回滚失败 status={status} body={rolled}") assert_or_exit( rolled.get("publish_status") == "DRAFT" and (rolled.get("data") or {}).get("enrollmentPlan", {}).get("totalTarget") == payload["data"]["enrollmentPlan"]["totalTarget"], f"回滚结果异常 body={rolled}", ) print("[9/12] rollback_ok") status, _ = request_json(f"/api/v1/studies/{study_id}/lock", token=token, method="PATCH", payload={}) assert_or_exit(status == 200, f"锁定失败 status={status}") try: status, locked_result = request_json( f"/api/v1/studies/{study_id}/setup-config", token=token, method="PUT", payload={"expected_version": rolled["version"], "data": rolled["data"]}, ) assert_or_exit(status == 403, f"锁定后写入应403，实际 status={status} body={locked_result}") print("[10/10] lock_403_check_ok") finally: unlock_status, unlock_body = request_json(f"/api/v1/studies/{study_id}/unlock", token=token, method="PATCH", payload={}) assert_or_exit(unlock_status == 200, f"解锁失败 status={unlock_status} body={unlock_body}") print("setup-config smoke test passed") return 0 if __name__ == "__main__": sys.exit(main())