diff --git a/docs/API_REVIEW.md b/docs/API_REVIEW.md index 167fcf4..10cfdd9 100644 --- a/docs/API_REVIEW.md +++ b/docs/API_REVIEW.md @@ -50,6 +50,6 @@ ## 文档与脚本清理 -104 份阶段报告、重复 JSON、旧目录审查和独立菜单 SQL，以及 26 个旧脚本已逐文件复制到仓库外并校验 SHA-256 后移除。保留统一迁移 SQL、业务口径、综合验收索引和 14 个当前验证工具，补充维护入口与脚本说明。 +首次清理将 104 份阶段报告、重复 JSON、旧目录审查和独立菜单 SQL，以及 26 个旧脚本移出项目。用户追加要求后，剩余 14 个验证工具、脚本说明和 Python 缓存也全部移出，`scripts` 目录已删除。所有移除文件均在仓库外归档并校验 SHA-256；统一迁移 SQL、业务口径与综合验收索引保留。 真实本地配置 `configs/config.yaml`、私有运行数据，以及其他会话的 V2 估算和 outputs 成果不纳入本次提交。未推送远端。 diff --git a/docs/README.md b/docs/README.md index fd3dc1f..4d6b49a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -4,6 +4,6 @@ - [当前接口审查](API_REVIEW.md)：此次修复、验证范围及保留的技术风险。 - [后台说明](ADMIN_BACKEND.md)、[支付接入](PAYMENT.md)、[短信接入](SMS_INTEGRATION.md)。 - [OA 迁移计划](OA_MIGRATION_PLAN.md)：独立 OA 工作的设计材料，不是当前商务迁移的完成证明。 -- [验证脚本](../scripts/README.md)：保留工具的用途和私有输入约定。 +- [汇总任务说明](sw6-to-kra/report-build-source-notes.md)：任务读写表及接口依赖；临时验证脚本已移出项目。 历史阶段记录已归档到仓库外；后续将最新结果更新到对应维护文档，不再每个接口生成一份日期报告。其他会话生成的商务 V2 估算及表格保持原状，本次清理和提交不包含这些文件。 diff --git a/docs/sw6-to-kra/README.md b/docs/sw6-to-kra/README.md index 380161c..0bc8d9e 100644 --- a/docs/sw6-to-kra/README.md +++ b/docs/sw6-to-kra/README.md @@ -8,7 +8,7 @@ 2. 在 [统一迁移 SQL](sw_to_kra_migrate.sql) 开头修改 `USE`，当前为 `USE sw9`。用支持 `DELIMITER` 的客户端在同一连接执行，遇错停止。 3. 启动当前 Go 版本，执行版本迁移，补齐业务锁、密码重置表、附件引用和查询索引。大表首次建索引需要等待完成。 4. 配置主库和 `lx`、`hw`、`yq`、`oa` 数据源。配置和凭据保存在仓库外；数据库连接信息不写入本目录。 -5. 对照 [验收记录](validation.md) 与 [验证脚本说明](../../scripts/README.md) 复核接口和任务。迁入计划默认停用，启用前应确认业务所需运行时段。 +5. 对照 [验收记录](validation.md) 与 [汇总任务说明](report-build-source-notes.md) 复核接口和任务。迁入计划默认停用，启用前应确认业务所需运行时段。 SQL 自动区分首次迁移、已完成迁移的菜单调整和异常中间状态；不会对完成迁移的库重新导入账号。重复执行及旧表已归档/清理的路径均有演练记录。 @@ -25,4 +25,4 @@ SQL 自动区分首次迁移、已完成迁移的菜单调整和异常中间状 SQL 生成器位于旧仓库 `script/sql/sw6_to_kra/build_sw7_migration.py`。修改生成器后同步本目录统一 SQL，并重新核对散列及迁移结果；不要恢复分阶段 SQL 补丁。 -阶段报告、重复 JSON、旧 sw7 冒烟脚本已归档到本机仓库外 `D:/Users/Yvan/backups/sw9/20261104_rehearsal/cleanup-archive/`。该目录以及相邻的运行配置、原始响应、分区备份不纳入 Git。历史索引中 `reports` 项的文件名可在归档内查找。 +阶段报告、重复 JSON、全部临时验证脚本已归档到本机仓库外 `D:/Users/Yvan/backups/sw9/20261104_rehearsal/cleanup-archive/`，最后保留的工具位于其中的 `scripts-final/`。项目不再保留 `scripts` 目录，Go 任务运行不依赖这些脚本。归档及相邻的运行配置、原始响应、分区备份不纳入 Git。历史索引中 `reports` 项的文件名可在归档内查找。 diff --git a/docs/sw6-to-kra/report-build-source-notes.md b/docs/sw6-to-kra/report-build-source-notes.md index dfb2a27..a9cee09 100644 --- a/docs/sw6-to-kra/report-build-source-notes.md +++ b/docs/sw6-to-kra/report-build-source-notes.md @@ -25,3 +25,38 @@ 领域与存储测试覆盖身份/账号规则、重复构建、自然日筛选和身份明细写失败后的原子回滚。真实 TCP HTTP 使用隔离数据库和 Kra 原生任务创建、手动触发、执行日志，核对两种报表实际落库且任务仍为禁用。 当前验证状态统一见 [验收与性能记录](validation.md)，不再维护阶段性运行报告。 + +## 任务、表与接口的对应关系 + +这些任务是 Go 内部注册的方法，与已删除的临时验证脚本无关。它们从当前主库读取已同步的原始数据，再生成统计结果；不负责将 SDK 平台数据同步进主库。当前运行服务主库为 sw9。 + +| 任务方法 | 日期范围 | 写入的统计表 | +|---|---|---| +| swReportDayTodayTask | 上海时区今天 | tab_sw_report_day、tab_sw_report_day_key | +| swReportDayYesterdayTask | 上海时区昨天 | 同上 | +| swReportDayBackfillTask | startDate 至 endDate，含两端 | 同上 | +| swReportServerDayTodayTask | 上海时区今天 | tab_sw_report_server_day、tab_sw_report_server_day_key | +| swReportServerDayYesterdayTask | 上海时区昨天 | 同上 | +| swReportServerDayBackfillTask | startDate 至 endDate，含两端 | 同上 | + +商务日报读取 `tab_user`（主账号注册及身份资料）、`tab_user_day_login`（登录）、`tab_spend`（成功充值）、`tab_game`（游戏名）。区服日报读取 `tab_user_play_info`（创角 create_time、活跃 play_time）、`tab_spend`、`tab_user`、`tab_game`。 + +`tab_sw_report_day` 按日期、平台、渠道、游戏、去重模式保存注册数、活跃人数、付费人数、新增付费人数和充值金额。`tab_sw_report_server_day` 在相同维度上增加区服，注册数代表创角玩家数，不包含新增付费人数。 + +两个 `_key` 表保存参与统计的身份键，供跨日期、渠道或游戏聚合时重新 `COUNT(DISTINCT identity_key)`。例如同一玩家连续两天活跃，查询这两天的独立活跃人数应为 1，不能直接将两个日活人数相加。`dedup_mode=0` 使用账号 ID，`dedup_mode=1` 使用规范化身份证；空身份证不计入模式 1。`metric` 为注册 1、活跃 2、付费 3，商务日报另有新增付费 4；金额查询只读模式 0，避免两种模式重复累加。 + +每次重建会在一个事务内锁定“报表类型＋日期”，读取原始数据，删除统计表及对应 key 表该 `stat_date` 的旧行，批量写入新结果。读取或写入失败、取消时回滚；每个日期单独提交，回跑中途失败不会撤销此前已完成日期。除统计表外，会使用 `sys_business_report_build_locks` 保存互斥锁键，任务执行框架将结果记入 `sys_timed_task_logs`。原始用户、充值和登录记录不会被这些任务改写。 + +| 使用方 | 接口 | 依赖 | +|---|---|---| +| 渠道数据统计 | GET /sw/channelData/list | 商务日报两表 | +| 游戏数据 | GET /sw/gameData/list | 商务日报两表 | +| 三类排行榜 | GET /sw/rank/channel、/business、/game | 商务日报两表 | +| 仪表盘/工作台 | GET /sw/workbench/summary、/registerTop、/payRank | 商务日报两表；汇总中的新增渠道另查渠道表 | +| 商务画像 | GET /sw/business/:userId/portrait、/sw/business/bindPlatform/portrait | 商务日报两表 | +| 渠道画像及游戏分布 | GET /sw/promote/portrait、/sw/promote/game-distribution | 商务日报两表；付费分层还会读取原始充值 | +| 区服分析 | GET /sw/serverAnalysis/list | 区服日报两表 | + +普通渠道列表、注册明细、充值明细、渠道激活、工作台未登录榜，以及游戏/区服下拉选项不直接使用这四张汇总表。预警检测 `alertTriggerTask` 另行读取原始注册/充值数据，不由这些汇总任务生成预警。 + +今天任务用于当天持续刷新，昨天任务用于补算昨日完整数据，backfill 用于补历史或纠正迟到数据。2026-11-04 核实时六条计划全部停用：不自动刷新汇总，但已有历史结果仍可查询；若原始数据继续同步，依赖汇总的指标会与明细逐渐产生时间差。 diff --git a/docs/sw6-to-kra/validation.md b/docs/sw6-to-kra/validation.md index 5d88b4a..59ba8c9 100644 --- a/docs/sw6-to-kra/validation.md +++ b/docs/sw6-to-kra/validation.md @@ -65,4 +65,4 @@ ## 页面 -渠道激活页不再在发起分页请求时清零 total；排行榜与仪表盘四榜并排、付费率用百分比、头部两项偏好直接勾选；三级商务保留商务列表/绑定列表切换。完整页面检查脚本仍保留，运行截图和带业务数据的响应仅存仓库外。 +渠道激活页不再在发起分页请求时清零 total；排行榜与仪表盘四榜并排、付费率用百分比、头部两项偏好直接勾选；三级商务保留商务列表/绑定列表切换。页面检查脚本、运行截图和带业务数据的响应均已保存在仓库外。 diff --git a/scripts/README.md b/scripts/README.md deleted file mode 100644 index c29fa35..0000000 --- a/scripts/README.md +++ /dev/null @@ -1,49 +0,0 @@ -# 商务验证工具 - -仅保留仍用于当前 sw8/sw9 验证的工具。旧 sw7、固定端口 18000、重复页面夹具脚本已归档；接口边界回归优先运行相应 Go 测试。 - -## 环境 - -- Python 3.10+；数据库工具需要 PyMySQL、PyYAML，Excel 对比需要 openpyxl。 -- 页面检查使用 Node 和 Playwright；通过 `SW_PLAYWRIGHT_MODULE` 指定已有模块路径。 -- `--runtime` 必须指向仓库外的私有目录。配置、令牌、实际响应、导出文件和 SQL 日志不得提交。 -- 旧 Java 连接 sw8，新 Go 连接 sw9。运行命令时明确传入端口，避免误连日常服务。任务保持停用。 - -通用输入：`legacy-.token`、`new-.token`、`legacy-client.json`、`id-maps.json`；写入测试还需要专门创建的三级测试身份及对应凭据。不要把业务正常账号当作可随意清理的测试身份。 - -## 工具用途 - -| 工具 | 用途与写入范围 | -|---|---| -| start-sw-rehearsal.ps1 | 从私有配置启动本机 Go 服务，临时注入 SDK 兼容参数，日志留在 runtime | -| sw_database_rehearsal.py | 数据库盘点、备份、迁移及验证；同时提供其他脚本依赖的 connect/query | -| sw_api_compare.py | 新旧读取接口逐字段对比；输入 case JSON 明确路径、参数、允许的协议转换 | -| sw_write_verify.py | 规则与预警状态流程测试；提供其他写入工具的 Rehearsal 支持 | -| sw_account_create_verify.py | 创建临时下级、验证层级/角色并清理；不删除预先存在的同名账号 | -| sw_account_write_verify.py | 测试账号资料/状态并恢复原值 | -| sw_preference_compare.py | 四种包含下级/身份证去重组合，结束恢复偏好 | -| sw_export_compare.py | 新旧注册和充值 Excel 逐单元格对比 | -| sw_sdk_write_verify.py | 独立平台夹具上的绑定、重提、解绑与渠道创建；受控 OA 审核 | -| sw_report_rehearsal.py | 指定日报分区备份、比较和恢复 | -| sw_run_report_tasks.py | 要求已有日期匹配的备份；创建停用的临时计划、手工执行并清理计划 | -| sw_endpoint_benchmark.py | 接口耗时、完整响应摘要及 SQL 采集，只读 | -| sw_endpoint_parity.py | 优化前后接口及偏好矩阵对比；即使 HTTP 恢复失败也尝试还原数据库偏好 | -| sw_dashboard_layout_smoke.cjs | 激活分页、偏好和榜单布局检查，截图留在指定私有目录 | - -先通过各 Python 工具的 `--help` 查看必需参数。日报、预警及写入工具会改动测试数据；使用其备份/恢复流程，并核对最终表计数与任务状态。 - -## 读取对比示例 - -```powershell -python scripts/sw_api_compare.py --runtime D:/private/sw-review --cases D:/private/sw-review/cases.json --old-url http://127.0.0.1:18080 --new-url http://127.0.0.1:19004 --output D:/private/sw-review/result.json -``` - -`--recompare` 只比较已保存响应；`--refresh-target` 使用已保存的同参数旧响应、重新请求新服务。报告会标记 `source_capture_reused`，不能把快照对比描述为双方重新实测。 - -本地 MySQL 回归只接受 `SW_TEST_MYSQL_DSN` 指向 `127.0.0.1` 的隔离实例： - -```powershell -go test ./internal/data/business -run MySQL -count=1 -``` - -这些测试创建并清理自己的随机数据库，不在 sw8/sw9 上运行测试建表和删除。 diff --git a/scripts/start-sw-rehearsal.ps1 b/scripts/start-sw-rehearsal.ps1 deleted file mode 100644 index 5356e59..0000000 --- a/scripts/start-sw-rehearsal.ps1 +++ /dev/null @@ -1,37 +0,0 @@ -param( - [Parameter(Mandatory = $true)][string]$RuntimePath, - [string]$ConfigName = 'config-next.yaml', - [string]$BinaryPath = '.cache/sw89/sw-new-complete.exe' -) - -$ErrorActionPreference = 'Stop' -$workspaceRoot = (Resolve-Path (Join-Path $PSScriptRoot '..')).Path -$runtimeRoot = (Resolve-Path -LiteralPath $RuntimePath).Path -$configPath = (Resolve-Path -LiteralPath (Join-Path $runtimeRoot $ConfigName)).Path -if (-not [System.IO.Path]::IsPathRooted($BinaryPath)) { - $BinaryPath = Join-Path $workspaceRoot $BinaryPath -} -$binary = (Resolve-Path -LiteralPath $BinaryPath).Path -$compat = Get-Content -LiteralPath (Join-Path $runtimeRoot 'sdk-compatibility.json') -Raw | ConvertFrom-Json -$previous = @{} -$settings = @{ - SW_SDK_AUTH_CODE = $compat.AUTH_CODE - SW_SDK_MIGRATE_AUTH_CODE = $compat.MIGRATE_AUTH_CODE - SW_SDK_TABLE_PREFIX = $compat.TABLE_PREFIX -} -try { - foreach ($name in $settings.Keys) { - $previous[$name] = [Environment]::GetEnvironmentVariable($name, 'Process') - [Environment]::SetEnvironmentVariable($name, $settings[$name], 'Process') - } - $process = Start-Process -FilePath $binary -ArgumentList @('-conf', ('"' + $configPath + '"')) ` - -WorkingDirectory $workspaceRoot -WindowStyle Hidden ` - -RedirectStandardOutput (Join-Path $runtimeRoot 'rehearsal.stdout.log') ` - -RedirectStandardError (Join-Path $runtimeRoot 'rehearsal.stderr.log') -PassThru - $process.Id | Set-Content -LiteralPath (Join-Path $runtimeRoot 'rehearsal.pid') - Write-Output "Rehearsal process: $($process.Id)" -} finally { - foreach ($name in $previous.Keys) { - [Environment]::SetEnvironmentVariable($name, $previous[$name], 'Process') - } -} diff --git a/scripts/sw_account_create_verify.py b/scripts/sw_account_create_verify.py deleted file mode 100644 index 74f5655..0000000 --- a/scripts/sw_account_create_verify.py +++ /dev/null @@ -1,73 +0,0 @@ -"""在 sw8/sw9 用专用二级账号创建临时三级账号，核对层级和角色后清理。""" - -import argparse -from pathlib import Path - -from sw_write_verify import Rehearsal - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--config", type=Path, required=True) - parser.add_argument("--old-url", default="http://127.0.0.1:18080") - parser.add_argument("--new-url", default="http://127.0.0.1:19002") - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]): - raise ValueError("测试凭据必须位于仓库外") - run = Rehearsal(args) - username = "mig_recheck_" + run.suffix - password = (args.runtime / "synthetic-test-password.txt").read_text(encoding="utf-8").strip() - source_roles = run.query("SELECT role_id FROM sw8.sys_user_role WHERE user_id=%s", (run.owner("legacy", 3),)) - expected_roles = run.query("SELECT sys_authority_authority_id id FROM sw9.sys_user_authority WHERE sys_user_id=%s", (run.owner("new", 3),)) - created_sides = set() - try: - run.check("测试角色存在", len(source_roles) == 1 and len(expected_roles) == 1) - for side, database, table, column in (("legacy", "sw8", "sys_user", "user_name"), ("new", "sw9", "sys_users", "username")): - run.check("临时账号不存在 " + side, not run.query(f"SELECT 1 FROM {database}.{table} WHERE {column}=%s", (username,))) - payload = {"userName": username, "password": password, "nickName": "迁移复核临时账号", "realName": "迁移复核临时账号"} - run.call("legacy", "testlevel2", "POST", "/system/user", {**payload, "roleIds": [str(source_roles[0]["role_id"])]}) - created_sides.add("sw8") - run.call("new", "testlevel2", "POST", "/sw/business", {**payload, "parentId": 0, "level": 1, "authorityId": 888}) - created_sides.add("sw9") - old = run.query("SELECT user_id,level,parent_id,id_path,status FROM sw8.sys_user WHERE user_name=%s", (username,))[0] - new = run.query("SELECT id,level,parent_id,id_path,business_status,authority_id,enable FROM sw9.sys_users WHERE username=%s", (username,))[0] - run.check("新旧创建三级且挂靠当前二级", old["level"] == new["level"] == 3 and str(old["parent_id"]) == str(run.owner("legacy", 2)) and new["parent_id"] == run.owner("new", 2)) - path = ",".join(run.mapping.get(part, part) for part in str(old["id_path"]).split(",")) - run.check("祖先路径按用户映射一致", path == new["id_path"]) - run.check("默认状态正常且角色不能由请求提权", str(old["status"]) == str(new["business_status"]) == "0" and new["enable"] == 1 and new["authority_id"] == expected_roles[0]["id"] and new["authority_id"] != 888) - old_roles = run.query("SELECT role_id FROM sw8.sys_user_role WHERE user_id=%s", (old["user_id"],)) - new_roles = run.query("SELECT sys_authority_authority_id id FROM sw9.sys_user_authority WHERE sys_user_id=%s", (new["id"],)) - run.check("关联角色正确", old_roles == source_roles and new_roles == expected_roles) - values = [] - for database, user_id in (("sw8", old["user_id"]), ("sw9", new["id"])): - values.append(run.query(f"SELECT include_sub_promote,idcard_dedup FROM {database}.sys_user_preference WHERE user_id=%s", (user_id,))) - run.check("默认统计偏好一致", values[0] == values[1] == [{"include_sub_promote": 1, "idcard_dedup": 1}]) - finally: - # 只清理本次唯一名称创建的临时账号及其角色/偏好，不触碰原有测试链。 - for database, table, id_col, name_col in (("sw8", "sys_user", "user_id", "user_name"), ("sw9", "sys_users", "id", "username")): - if database not in created_sides: - continue - rows = run.query(f"SELECT {id_col} id FROM {database}.{table} WHERE {name_col}=%s", (username,)) - for row in rows: - user_id = row["id"] - run.db.begin() - try: - run.query(f"DELETE FROM {database}.sys_user_preference WHERE user_id=%s", (user_id,)) - if database == "sw8": - for relation in ("sys_user_role", "sys_user_post"): - run.query(f"DELETE FROM sw8.{relation} WHERE user_id=%s", (user_id,)) - else: - run.query("DELETE FROM sw9.sys_user_authority WHERE sys_user_id=%s", (user_id,)) - run.query(f"DELETE FROM {database}.{table} WHERE {id_col}=%s AND {name_col}=%s", (user_id, username)) - run.db.commit() - except Exception: - run.db.rollback() - raise - run.check("临时账号清理完成 " + database, not run.query(f"SELECT 1 FROM {database}.{table} WHERE {name_col}=%s", (username,))) - run.db.close() - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_account_write_verify.py b/scripts/sw_account_write_verify.py deleted file mode 100644 index 2223507..0000000 --- a/scripts/sw_account_write_verify.py +++ /dev/null @@ -1,75 +0,0 @@ -"""核对专用二级测试账号的资料更新、四种商务状态及恢复，不修改真实商务账号。""" - -import argparse -import json -import time -from pathlib import Path - -from sw_write_verify import Rehearsal - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--config", type=Path, default=Path("configs/config.yaml")) - parser.add_argument("--old-url", default="http://127.0.0.1:18080") - parser.add_argument("--new-url", default="http://127.0.0.1:19001") - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - run = Rehearsal(args) - account = run.users[1] - if not account["username"].startswith("mig_verify_"): - raise ValueError("仅允许修改本轮创建的专用测试账号") - originals = { - "legacy": run.query("SELECT real_name,status,reject_reason,update_time FROM sw8.sys_user WHERE user_id=%s AND user_name=%s", (account["source_id"], account["username"]))[0], - "new": run.query("SELECT real_name,business_status,enable,reject_reason,updated_at FROM sw9.sys_users WHERE id=%s AND username=%s", (account["target_id"], account["username"]))[0], - } - (args.runtime / "test-account-before-write.json").write_text(json.dumps(originals, default=str, indent=2), encoding="utf-8") - - def update(side, status, name, reason): - payload = {"userId": account["source_id" if side == "legacy" else "target_id"], "userName": account["username"], "realName": name, - "status": str(status), "rejectReason": reason or "", "roleIds": ["2087435436573081601"]} - return run.call(side, "admin", "PUT", "/system/user" if side == "legacy" else "/sw/business", payload) - - try: - started = time.monotonic() - for index, status in enumerate((0, 3, 2, 1, 0)): - if index == 4: - # 旧控制器对相同请求体有五秒重复提交限制。 - time.sleep(max(0, 5.2 - (time.monotonic() - started))) - name = "迁移核对二级资料更新" - reason = "迁移核对驳回" if status == 2 else "" - for side in ("legacy", "new"): - update(side, status, name, reason) - path = "/system/user/" + str(account["source_id"]) if side == "legacy" else "/sw/business/" + str(account["target_id"]) - detail = run.call(side, "admin", "GET", path)["data"] - if side == "legacy": - detail = detail["user"] - run.check(f"资料及状态 {status} {side}", detail["realName"] == name and str(detail["status"]) == str(status) and detail["level"] == 2) - enabled = run.query("SELECT enable FROM sw9.sys_users WHERE id=%s", (account["target_id"],))[0]["enable"] - run.check(f"商务状态 {status} 与启停一致", enabled == (1 if status == 0 else 0)) - replies = {side: run.call(side, "testlevel2", "GET", run.prefix(side) + "/user/preference", success=False) for side in ("legacy", "new")} - if status != 0: - run.check(f"状态 {status} 拒绝旧会话", replies["new"]["code"] != 0, - legacy_session_allowed=replies["legacy"]["code"] == 200, accepted_change="新系统每次请求核对商务状态，阻止停用/待审/驳回账号沿用旧会话") - else: - run.check("正常账号会话可用", replies["new"]["code"] == 0 and replies["legacy"]["code"] == 200) - finally: - for side in ("legacy", "new"): - original = originals[side] - update(side, original["status" if side == "legacy" else "business_status"], original["real_name"], original["reject_reason"]) - old, new = originals["legacy"], originals["new"] - run.query("UPDATE sw8.sys_user SET real_name=%s,status=%s,reject_reason=%s,update_time=%s WHERE user_id=%s AND user_name=%s", - (old["real_name"], old["status"], old["reject_reason"], old["update_time"], account["source_id"], account["username"])) - run.query("UPDATE sw9.sys_users SET real_name=%s,business_status=%s,enable=%s,reject_reason=%s,updated_at=%s WHERE id=%s AND username=%s", - (new["real_name"], new["business_status"], new["enable"], new["reject_reason"], new["updated_at"], account["target_id"], account["username"])) - actual = { - "legacy": run.query("SELECT real_name,status,reject_reason,update_time FROM sw8.sys_user WHERE user_id=%s", (account["source_id"],))[0], - "new": run.query("SELECT real_name,business_status,enable,reject_reason,updated_at FROM sw9.sys_users WHERE id=%s", (account["target_id"],))[0], - } - run.check("测试账号原始资料和状态已恢复", actual == originals) - run.db.close() - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_api_compare.py b/scripts/sw_api_compare.py deleted file mode 100644 index a851503..0000000 --- a/scripts/sw_api_compare.py +++ /dev/null @@ -1,201 +0,0 @@ -"""请求隔离的新旧商务实例；原始响应和令牌留在仓库外，报告只记录校验结果。""" - -import argparse -import concurrent.futures -import decimal -import hashlib -import json -import re -import time -import urllib.error -import urllib.parse -import urllib.request -from pathlib import Path - - -def fetch(base, token, path, params, legacy, client_id="", timeout=120): - url = base.rstrip("/") + path - if params: - url += "?" + urllib.parse.urlencode(params, doseq=True) - headers = {"Authorization": "Bearer " + token} if legacy else {"x-token": token} - if legacy: - headers["clientid"] = client_id - request = urllib.request.Request(url, headers=headers) - started = time.monotonic() - try: - with urllib.request.urlopen(request, timeout=timeout) as response: - body = json.load(response) - return {"http": response.status, "elapsed_ms": round((time.monotonic() - started) * 1000), "body": body} - except (urllib.error.URLError, TimeoutError, ValueError) as error: - return {"http": None, "elapsed_ms": round((time.monotonic() - started) * 1000), "error": type(error).__name__} - - -def data_of(reply): - body = reply.get("body", {}) - if "rows" in body: - return {key: value for key, value in body.items() if key not in ("code", "msg")} - return body.get("data") - - -def shape(value): - if isinstance(value, dict): - return {key: (len(item) if isinstance(item, list) else type(item).__name__) for key, item in value.items()} - return {"type": type(value).__name__, "length": len(value) if isinstance(value, list) else None} - - -def expected_relative_pay_rate(row): - """用户确认的新口径：由旧的本期付费率及百分点差推导相对环比。""" - if row.get("momValue") is None: - return row - current = decimal.Decimal(str(row["metricValue"])) - difference = decimal.Decimal(str(row["momValue"])) - previous = current - difference - result = dict(row) - if previous <= 0: - result.update(momValue=None, momStatus=None) - else: - change = (difference / previous * 100).quantize(decimal.Decimal("0.01"), rounding=decimal.ROUND_HALF_UP) - result.update(momValue=format(abs(change) if change == 0 else change, ".2f"), momStatus=0 if change == 0 else 1 if change > 0 else -1) - return result - - -def compare_data(source, target, case, users): - differences = [] - omitted = {"source": case.get("source_omit_fields", []), "target": case.get("target_omit_fields", [])} - # 两边通过 HTTP 分别创建的专用测试账号，其创建时刻天然不同。 - # 只忽略显式列出的夹具账号时间，真实迁移账号的所有时间字段仍须相等。 - generated = {str(value) for value in case.get("generated_user_ids", [])} - generated_ids = {"source": generated, "target": {users.get(value, value) for value in generated}} - user_fields = set(case.get("user_fields", ["userId", "parentId", "creatorId", "businessId", "operatorId", "parentUserId"])) - - def normalize(value, side, field="", path=""): - if isinstance(value, dict): - if side == "source" and field == "payRateRankList" and case.get("relative_pay_rate"): - value = expected_relative_pay_rate(value) - if side == "target" and value.get("available") is False: - differences.append({"path": path + ".available", "kind": "unavailable"}) - if side == "source" and case.get("normalize_root_path") and value.get("level") in (0, 1, "0", "1") and value.get("idPath") is None: - value = {**value, "idPath": "0"} - excluded = set(omitted[side]) - if str(value.get("userId")) in generated_ids[side]: - excluded.add("createTime") - return {key: normalize(item, side, key, path + "." + key) for key, item in value.items() if key not in excluded} - if isinstance(value, list): - rows = [normalize(item, side, field, path) for item in value] - keys = case.get("row_key") - if keys and rows and all(isinstance(row, dict) and all(key in row for key in keys) for row in rows): - rows.sort(key=lambda row: tuple(str(row[key]) for key in keys)) - return rows - if value is None and field in case.get("null_empty_fields", []): - return "" - if side == "source" and case.get("map_users", True) and value is not None: - if field in user_fields: - value = users.get(str(value), str(value)) - elif field == "idPath": - value = ",".join(users.get(item, item) for item in str(value).split(",")) - return value - - def walk(left, right, path="data"): - # 只兼容数值与 JSON 数字字符串；账号、手机号、区服编号等字符串保留前导零。 - numeric_field = path.rsplit(".", 1)[-1] in case.get("numeric_fields", []) - if not isinstance(left, bool) and not isinstance(right, bool) and (numeric_field or isinstance(left, (int, float)) or isinstance(right, (int, float))): - try: - if decimal.Decimal(str(left)) == decimal.Decimal(str(right)): - return - except decimal.InvalidOperation: - pass - if type(left) is not type(right): - differences.append({"path": path, "kind": "type", "source": type(left).__name__, "target": type(right).__name__}) - elif isinstance(left, dict): - for key in sorted(left.keys() | right.keys()): - if key not in left or key not in right: - differences.append({"path": path + "." + key, "kind": "missing_field", "side": "source" if key not in left else "target"}) - else: - walk(left[key], right[key], path + "." + key) - elif isinstance(left, list): - if len(left) != len(right): - differences.append({"path": path, "kind": "length", "source": len(left), "target": len(right)}) - for i, (a, b) in enumerate(zip(left, right)): - walk(a, b, path + f"[{i}]") - elif left != right: - # 数据值只留在仓库外；公开报告保留字段路径和摘要供复核。 - differences.append({"path": path, "kind": "value", "source_sha256": hashlib.sha256(str(left).encode()).hexdigest(), "target_sha256": hashlib.sha256(str(right).encode()).hexdigest()}) - - walk(normalize(source, "source"), normalize(target, "target")) - return {"semantic_equal": not differences, "difference_count": len(differences), "differences": differences[:30], - "normalization": {"user_ids_mapped": case.get("map_users", True), "numeric_representation": True, - "user_fields": sorted(user_fields), "numeric_string_fields": case.get("numeric_fields", []), - "normalize_root_path": case.get("normalize_root_path", False), - "expected_relative_pay_rate": case.get("relative_pay_rate", False), - "generated_user_creation_times": len(generated), - "row_key": case.get("row_key"), "omitted_fields": omitted, "null_empty_fields": case.get("null_empty_fields", [])}} - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--cases", type=Path, required=True) - parser.add_argument("--identity", default="admin") - parser.add_argument("--old-url", default="http://127.0.0.1:18080") - parser.add_argument("--new-url", default="http://127.0.0.1:19000") - parser.add_argument("--output", type=Path, required=True) - mode = parser.add_mutually_exclusive_group() - mode.add_argument("--recompare", action="store_true", help="使用已保存的同参数 HTTP 响应重新核对，不重复访问数据库") - mode.add_argument("--refresh-target", action="store_true", help="保留已采集的同参数旧响应，重新请求优化后的新实现；新增用例仍请求两边") - args = parser.parse_args() - root = Path(__file__).resolve().parents[1] - if args.runtime.resolve().is_relative_to(root): - raise ValueError("原始业务响应必须保存在仓库外") - tokens = [(args.runtime / (prefix + "-" + args.identity + ".token")).read_text(encoding="utf-8").strip() for prefix in ("legacy", "new")] - client_id = json.loads((args.runtime / "legacy-client.json").read_text(encoding="utf-8"))["client_id"] - users = {str(row["old_id"]): str(row["new_id"]) for row in json.loads((args.runtime / "id-maps.json").read_text(encoding="utf-8")) if row["entity"] == "user"} - cases = json.loads(args.cases.read_text(encoding="utf-8")) - reports = [] - folder = args.runtime / ("responses-" + args.identity) - folder.mkdir(exist_ok=True) - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: - for case in cases: - capture = folder / (case["name"] + ".json") - reused_source = args.refresh_target and capture.exists() - if args.recompare or reused_source: - saved = json.loads(capture.read_text(encoding="utf-8")) - if any(saved["case"].get(key) != case.get(key) for key in ("source", "target", "params", "source_params", "target_params")): - raise ValueError("已有 HTTP 响应的请求条件不符：" + case["name"]) - replies = [saved["source"], saved["target"]] - if args.refresh_target: - replies[1] = fetch(args.new_url, tokens[1], case["target"], case.get("target_params", case.get("params", {})), False, timeout=case.get("timeout_seconds", 120)) - capture.write_text(json.dumps({"case": case, "source": replies[0], "target": replies[1]}, ensure_ascii=False, indent=2), encoding="utf-8") - else: - pairs = [pool.submit(fetch, base, tokens[i], case[side], case.get(side + "_params", case.get("params", {})), i == 0, client_id, case.get("timeout_seconds", 120)) - for i, (base, side) in enumerate(((args.old_url, "source"), (args.new_url, "target")))] - replies = [job.result() for job in pairs] - capture.write_text(json.dumps({"case": case, "source": replies[0], "target": replies[1]}, ensure_ascii=False, indent=2), encoding="utf-8") - values = [data_of(reply) for reply in replies] - for i, side in enumerate(("source", "target")): - for key in case.get(side + "_data_path", []): - values[i] = values[i][key] - codes = [reply.get("body", {}).get("code") for reply in replies] - report = {"case": case["name"], "source": case["source"], "target": case["target"], "codes": codes, - "source_capture_reused": reused_source, - "http_passed": codes == [200, 0], "elapsed_ms": [reply["elapsed_ms"] for reply in replies], - "shapes": [shape(value) for value in values], - "exact_data_equal": values[0] == values[1]} - if all(isinstance(value, dict) and "total" in value for value in values): - report["totals"] = [value["total"] for value in values] - report["totals_equal"] = str(values[0]["total"]) == str(values[1]["total"]) - reports.append(report) - report.update(compare_data(values[0], values[1], case, users)) - report["passed"] = report["http_passed"] and report["semantic_equal"] - if case.get("expected_error_codes"): - messages = [reply.get("body", {}).get("msg") for reply in replies] - report["rejection_equal"] = codes == case["expected_error_codes"] and messages[0] == messages[1] - report["expected_error_codes"] = case["expected_error_codes"] - report["passed"] = report["rejection_equal"] - print(json.dumps(report, ensure_ascii=False), flush=True) - args.output.write_text(json.dumps({"identity": args.identity, "checks": reports}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") - if not all(report["passed"] for report in reports): - raise SystemExit(1) - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_dashboard_layout_smoke.cjs b/scripts/sw_dashboard_layout_smoke.cjs deleted file mode 100644 index 287d050..0000000 --- a/scripts/sw_dashboard_layout_smoke.cjs +++ /dev/null @@ -1,218 +0,0 @@ -// 真实开发服务：检查 OA 数据、统计口径持久化与响应式布局。 -// 临时修改当前测试账号的统计口径，结束时恢复；截图只写入指定的私有目录。 -const fs = require('node:fs/promises') -const path = require('node:path') -const assert = require('node:assert/strict') -const { chromium } = require(process.env.SW_PLAYWRIGHT_MODULE || 'playwright') - -;(async () => { - const base = process.env.SW_UI_BASE_URL || 'http://127.0.0.1:8080' - const output = process.env.SW_UI_OUTPUT_DIR - assert.ok(output && process.env.SW_TEST_TOKEN_FILE, 'Set SW_UI_OUTPUT_DIR and SW_TEST_TOKEN_FILE') - const token = (await fs.readFile(process.env.SW_TEST_TOKEN_FILE, 'utf8')).trim() - const browser = await chromium.launch({ channel: 'chrome', headless: true }) - const context = await browser.newContext({ viewport: { width: 1600, height: 1000 } }) - const headers = { 'x-token': token } - const api = async (method, endpoint, data) => { - const response = await context.request.fetch(`${base}/api${endpoint}`, { method, headers, data }) - assert.ok(response.ok()) - const body = await response.json() - assert.equal(body.code, 0) - return body.data - } - let original - const report = { checks: [], pageErrors: [] } - try { - await fs.mkdir(output, { recursive: true }) - original = await api('GET', '/sw/user/preference') - await fs.writeFile(path.join(output, 'original-preference.json'), JSON.stringify(original)) - await context.addInitScript(value => localStorage.setItem('token', value), token) - const page = await context.newPage() - page.on('pageerror', e => report.pageErrors.push(e.message)) - const activationPath = '/sw/promoteActivePool/bindList' - await page.goto(`${base}/#/layout/sw-BusMgmt/ChannelActivationList`) - await page.locator('.el-table__body-wrapper tr').first().waitFor() - const activation = await api('GET', `${activationPath}?pageNum=1&pageSize=10&platformNames=lx,hw,yq`) - assert.equal(activation.available, true) - assert.ok(activation.total > 0) - assert.equal(await page.locator('.el-table__body-wrapper tr').count(), activation.rows.length) - assert.equal(await page.locator('.el-alert--warning').count(), 0) - report.checks.push({ name: 'actual OA activation rows', total: activation.total }) - await page.screenshot({ path: path.join(output, 'activation.png'), animations: 'disabled' }) - - const pagination = page.locator('.kra-pagination') - const pageRequests = [] - const trackPage = request => { - if (request.url().includes(activationPath)) pageRequests.push(Number(new URL(request.url()).searchParams.get('pageNum'))) - } - page.on('request', trackPage) - const movePage = async (target, action) => { - pageRequests.length = 0 - const [response] = await Promise.all([ - page.waitForResponse(r => r.url().includes(activationPath) && Number(new URL(r.url()).searchParams.get('pageNum')) === target), - action() - ]) - const expected = (await response.json()).data - await page.waitForFunction(() => !document.querySelector('.el-loading-mask')) - assert.equal((await pagination.locator('.el-pager .is-active').innerText()).trim(), String(target)) - assert.equal(await page.locator('.el-table__body-wrapper tr').count(), expected.rows.length) - const names = await page.locator('.el-table__body-wrapper tr td:nth-child(2) .cell').allTextContents() - assert.deepEqual(names.map(name => name.trim()), expected.rows.map(row => row.promoteAccount)) - assert.deepEqual(pageRequests, [target], 'pagination must not trigger a second request back to page 1') - } - assert.ok(activation.total > 10, 'Pagination check needs more than one page') - await movePage(2, () => pagination.locator('.btn-next').click()) - const lastPage = Math.ceil(activation.total / 10) - if (lastPage > 2) await movePage(lastPage, () => pagination.locator('.el-pager .number').filter({ hasText: new RegExp(`^${lastPage}$`) }).click()) - await page.screenshot({ path: path.join(output, 'activation-last-page.png'), animations: 'disabled' }) - await movePage(lastPage - 1, () => pagination.locator('.btn-prev').click()) - if (lastPage - 1 !== 1) await movePage(1, () => pagination.locator('.el-pager .number').filter({ hasText: /^\s*1\s*$/ }).click()) - page.off('request', trackPage) - report.checks.push({ name: 'activation pagination: next, last, previous, first; response rows match and no page reset', total: activation.total, lastPage }) - - const preference = page.getByRole('group', { name: '统计口径', exact: true }) - for (const [key, label] of [['includeSubPromote', '包含下级渠道'], ['idcardDedup', '身份证去重']]) { - const checkbox = preference.getByRole('checkbox', { name: label, exact: true }) - await page.waitForFunction(label => [...document.querySelectorAll('.business-preferences input')].some(el => el.closest('label')?.textContent.includes(label) && !el.disabled), label) - assert.equal(await checkbox.isChecked(), original[key]) - const [saved] = await Promise.all([ - page.waitForResponse(r => r.url().includes('/sw/user/preference') && r.request().method() === 'PUT'), - page.waitForResponse(r => r.url().includes(activationPath)), - preference.getByText(label, { exact: true }).click() - ]) - assert.equal((await saved.json()).code, 0) - assert.equal((await api('GET', '/sw/user/preference'))[key], !original[key]) - await page.reload() - await preference.waitFor() - await page.waitForFunction(() => [...document.querySelectorAll('.business-preferences input')].every(el => !el.disabled)) - assert.equal(await preference.getByRole('checkbox', { name: label, exact: true }).isChecked(), !original[key]) - report.checks.push({ name: `${key}: visible, persisted, page refreshed` }) - } - await api('PUT', '/sw/user/preference', original) - - // 保存失败不能让勾选状态与服务端失去同步。 - await page.reload() - await page.waitForFunction(() => document.querySelectorAll('.business-preferences input').length === 2 && [...document.querySelectorAll('.business-preferences input')].every(el => !el.disabled)) - const rejectSave = route => route.request().method() === 'PUT' - ? route.fulfill({ json: { code: 7, msg: '测试保存失败', data: {} } }) : route.continue() - await page.route('**/api/sw/user/preference', rejectSave) - await Promise.all([ - page.waitForResponse(r => r.url().includes('/sw/user/preference') && r.request().method() === 'PUT'), - preference.getByText('包含下级渠道', { exact: true }).click() - ]) - await page.waitForFunction(() => [...document.querySelectorAll('.business-preferences input')].every(el => !el.disabled)) - assert.equal(await preference.getByRole('checkbox', { name: '包含下级渠道', exact: true }).isChecked(), original.includeSubPromote) - await page.unroute('**/api/sw/user/preference', rejectSave) - report.checks.push({ name: 'failed preference save preserves previous value' }) - - const search = () => page.locator('.kra-search-box').getByRole('button', { name: '查询', exact: true }).click() - const activationPattern = '**/api/sw/promoteActivePool/bindList?*' - await page.route(activationPattern, route => route.fulfill({ json: { code: 7, msg: '测试查询失败', data: {} } })) - await search() - await page.getByText('查询失败，请重试', { exact: true }).waitFor() - assert.equal(await page.locator('.el-table__body-wrapper tr').count(), 0) - await page.unroute(activationPattern) - await page.route(activationPattern, route => route.fulfill({ json: { code: 0, data: { available: false, rows: [], total: 0 } } })) - await search() - await page.getByText('OA 数据源暂不可用', { exact: true }).waitFor() - await page.unroute(activationPattern) - await search() - await page.locator('.el-table__body-wrapper tr').first().waitFor() - report.checks.push({ name: 'activation error and unavailable states differ from empty results; retry recovers' }) - - for (const [name, route, grid] of [ - ['ranking', 'sw-DataAnalysis/RankingAnalysis', '.ranking-panels'], - ['dashboard', 'dashboard', '.business-ranks'] - ]) { - await page.goto(`${base}/#/layout/${route}`) - await page.locator(grid).waitFor() - await page.waitForFunction(() => !document.querySelector('.el-loading-mask')) - await page.locator('.el-message').evaluateAll(nodes => nodes.forEach(node => node.querySelector('.el-message__closeBtn')?.click())) - if (name === 'ranking') { - for (const panel of await page.locator('.ranking-panel').all()) { - assert.match(await panel.innerText(), /上期环比/) - assert.doesNotMatch(await panel.innerText(), /百分点|\bpp\b/) - } - } else { - assert.equal(await page.getByText('快捷功能', { exact: true }).count(), 0) - assert.equal(await page.getByText('公告', { exact: true }).count(), 0) - assert.equal(await page.getByText('商务数据', { exact: true }).count(), 0) - assert.ok(await page.getByText('数据平台', { exact: true }).isVisible()) - for (const index of [0, 1]) { - const panel = page.locator('.business-ranks > .el-card').nth(index) - await panel.locator('.business-rank-period').click() - const optionsId = await panel.getByRole('combobox').getAttribute('aria-controls') - await Promise.all([ - page.waitForResponse(r => /\/sw\/workbench\/(registerTop|payRank)/.test(r.url()) && r.url().includes('period=last7')), - page.locator(`[id="${optionsId}"]`).getByRole('option', { name: '近7天', exact: true }).click() - ]) - await panel.locator('.ranking-table tbody tr').first().waitFor() - assert.equal(await panel.locator('.ranking-empty').count(), 0) - } - report.checks.push({ name: 'dashboard period filters refresh real data in compact tables' }) - } - for (const [width, columns] of [[1920, 4], [1600, 4], [1366, 4], [1024, 2], [390, 1]]) { - await page.setViewportSize({ width, height: 1000 }) - if (width === 390) await page.getByRole('button', { name: '打开菜单', exact: true }).waitFor() - await page.waitForFunction(({grid, columns}) => getComputedStyle(document.querySelector(grid)).gridTemplateColumns.split(' ').length === columns, { grid, columns }) - const boxes = await page.locator(`${grid} > *`).evaluateAll(nodes => nodes.map(n => { const r = n.getBoundingClientRect(); return { x: r.x, y: r.y, right: r.right } })) - if (columns === 4) assert.ok(boxes.every(box => Math.abs(box.y - boxes[0].y) < 1)) - assert.ok(boxes.every(box => box.x >= 0 && box.right <= width + 1), `panel overflow: ${name}, ${width}`) - const overflow = await page.locator(`${grid} .el-table__body-wrapper, ${grid} .ranking-table-wrap`).evaluateAll(nodes => nodes.some(n => n.querySelector('table')?.clientWidth > n.clientWidth + 1)) - assert.equal(overflow, false, `table horizontal overflow: ${name}, ${width}`) - if (name === 'ranking') { - const tallRows = await page.locator('.ranking-table tr').evaluateAll(nodes => nodes.some(n => n.getBoundingClientRect().height > 40)) - assert.equal(tallRows, false, `ranking rows should stay compact: ${width}`) - } - const clippedNumbers = await page.locator(`${grid} tbody .ranking-numeric, ${grid} .ranking-number`).evaluateAll(nodes => nodes.some(n => n.scrollWidth > n.clientWidth + 1)) - assert.equal(clippedNumbers, false, `numeric values must be fully visible: ${name}, ${width}`) - const prefBox = await preference.boundingBox() - assert.ok(prefBox && prefBox.x >= 0 && prefBox.x + prefBox.width <= width, `preference overflow at ${width}`) - const toolbarOverflow = await page.getByRole('button', { name: '打开菜单', exact: true }).count() ? await page.locator('.business-preferences').evaluate(el => { - const header = el.closest('.fixed') - return header.scrollWidth > header.clientWidth + 1 - }) : false - assert.equal(toolbarOverflow, false, `mobile header overflow: ${width}`) - await page.screenshot({ path: path.join(output, `${name}-${width}.png`), animations: 'disabled' }) - report.checks.push({ name: `${name} layout`, width, columns, noOverflow: true }) - } - await page.setViewportSize({ width: 1600, height: 1000 }) - } - await page.goto(`${base}/#/layout/sw-BusMgmt/BusinessList`) - await page.getByRole('button', { name: '添加子商务', exact: true }).waitFor() - assert.equal(await page.locator('.business-view-switch').count(), 0) - assert.equal(await page.getByRole('tab', { name: '商务列表', exact: true }).count(), 0) - await page.screenshot({ path: path.join(output, 'business-list-admin.png'), animations: 'disabled' }) - // 只替换展示身份核对三级专属切换器；读取接口仍用授权测试身份，不做业务写入。 - await page.route('**/api/user/getUserInfo', async route => { - const response = await route.fetch() - const body = await response.json() - Object.assign(body.data.userInfo, { level: 3, authorityId: 903 }) - await route.fulfill({ response, json: body }) - }) - await page.reload() - await page.getByRole('radiogroup', { name: '商务管理视图', exact: true }).waitFor() - assert.equal(await page.getByRole('button', { name: '添加子商务', exact: true }).count(), 0) - const viewSwitch = page.getByRole('radiogroup', { name: '商务管理视图', exact: true }) - await Promise.all([ - page.waitForResponse(r => r.url().includes('/sw/business/bindPlatform')), - viewSwitch.getByText('绑定列表', { exact: true }).click() - ]) - assert.equal(await page.getByPlaceholder('请输入账号', { exact: true }).isVisible(), false) - await viewSwitch.getByText('商务列表', { exact: true }).click() - assert.ok(await page.getByPlaceholder('请输入账号', { exact: true }).isVisible()) - await page.screenshot({ path: path.join(output, 'business-list-level3.png'), animations: 'disabled' }) - report.checks.push({ name: 'business list: no redundant single tab; controlled level-3 switch works' }) - assert.deepEqual(report.pageErrors, []) - report.passed = true - } catch (e) { - const page = context.pages()[0] - if (page) await page.screenshot({ path: path.join(output, 'failure.png'), animations: 'disabled' }).catch(() => {}) - throw e - } finally { - if (original) await api('PUT', '/sw/user/preference', original) - await fs.writeFile(path.join(output, 'ui-checks.json'), JSON.stringify(report, null, 2) + '\n') - await browser.close() - } - console.log(JSON.stringify(report)) -})().catch(e => { console.error(e.stack); process.exitCode = 1 }) diff --git a/scripts/sw_database_rehearsal.py b/scripts/sw_database_rehearsal.py deleted file mode 100644 index a1f1526..0000000 --- a/scripts/sw_database_rehearsal.py +++ /dev/null @@ -1,188 +0,0 @@ -"""在同一测试服务器核对 SW 源副本、备份目标库并执行迁移；凭据只读取本地配置。""" - -import argparse -import datetime as dt -import gzip -import hashlib -import json -import re -import shutil -import subprocess -import tempfile -from pathlib import Path - -import pymysql -import yaml - - -def identifier(value): - if not re.fullmatch(r"[A-Za-z0-9_]+", value): - raise ValueError("数据库和表名只能包含字母、数字、下划线") - return "`" + value + "`" - - -def write_json(path, data): - path.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8") - - -def connect(config): - return pymysql.connect(host=config["host"], port=int(config["port"]), user=config["user"], - password=config["password"], charset="utf8mb4", connect_timeout=10, - read_timeout=180, autocommit=True, cursorclass=pymysql.cursors.DictCursor) - - -def query(conn, sql, args=None): - with conn.cursor() as cursor: - cursor.execute(sql, args) - return cursor.fetchall() - - -def source_tables(sql): - tables = re.findall(r"CREATE TABLE `bak_sw2kra_20261103_([^`]+)` LIKE", sql) - if len(tables) != 26 or len(set(tables)) != 26: - raise ValueError("迁移源表清单发生变化，请先复核备份范围") - return tables - - -def fingerprint(conn, database, table): - columns = query(conn, "SELECT column_name,data_type FROM information_schema.columns WHERE table_schema=%s AND table_name=%s ORDER BY ordinal_position", (database, table)) - keys = query(conn, "SELECT column_name FROM information_schema.statistics WHERE table_schema=%s AND table_name=%s AND index_name='PRIMARY' ORDER BY seq_in_index", (database, table)) - if not keys: - raise ValueError("备份核对要求稳定主键：" + table) - order = ",".join(identifier(item["column_name"]) for item in keys) - digest = hashlib.sha256() - rows = query(conn, f"SELECT * FROM {identifier(database)}.{identifier(table)} ORDER BY {order}") - json_columns = [item["column_name"] for item in columns if item["data_type"] == "json"] - for row in rows: - # JSON 的内部二进制布局在导入后可能不同，比较实际业务值而非 CHECKSUM 的存储字节。 - for name in json_columns: - if row[name] is not None: - row[name] = json.loads(row[name]) - digest.update(json.dumps(row, ensure_ascii=False, sort_keys=True, default=str, separators=(",", ":")).encode()) - digest.update(b"\n") - return {"rows": len(rows), "sha256": digest.hexdigest()} - - -def client_options(config, folder): - # 使用独立选项文件，避免密码出现在进程命令行或输出中。 - def escape(value): - return str(value).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n").replace("\r", "\\r") - handle = tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".cnf", dir=folder, delete=False) - with handle: - handle.write("[client]\n") - for key in ("host", "port", "user", "password"): - handle.write(f'{key}="{escape(config[key])}"\n') - handle.write("default-character-set=utf8mb4\n") - return Path(handle.name) - - -def prepare(args, config, sql): - if (args.state / (args.target + "-migration-source.sql.gz")).exists(): - raise ValueError("目标备份已存在；请使用新的状态目录，保留原始备份") - tables = source_tables(sql) - manifest = {"source": args.source, "target": args.target, "created_at": dt.datetime.now(dt.timezone.utc), - "sql_sha256": hashlib.sha256(sql.encode()).hexdigest(), "checks": []} - with connect(config) as conn: - manifest["server"] = query(conn, "SELECT @@hostname hostname,@@version version")[0] - if query(conn, "SELECT 1 FROM information_schema.tables WHERE table_schema=%s AND table_name='sys_users'", (args.target,)): - raise ValueError("目标已迁移，不能覆盖原始基线") - for table in tables: - checksums = query(conn, f"CHECKSUM TABLE {identifier(args.source)}.{identifier(table)}, {identifier(args.target)}.{identifier(table)} EXTENDED") - values = [row["Checksum"] for row in checksums] - fingerprints = [fingerprint(conn, db, table) for db in (args.source, args.target)] - item = {"table": table, "rows": [item["rows"] for item in fingerprints], "checksums": values, - "fingerprints": fingerprints, "passed": fingerprints[0] == fingerprints[1]} - manifest["checks"].append(item) - print(json.dumps(item, ensure_ascii=False), flush=True) - if not all(item["passed"] for item in manifest["checks"]): - write_json(args.state / "baseline.json", manifest) - raise RuntimeError("源库与目标库框架数据不一致，未执行迁移") - manifest["tables_before"] = query(conn, "SELECT table_name,engine FROM information_schema.tables WHERE table_schema=%s ORDER BY table_name", (args.target,)) - schemas = {} - for row in manifest["tables_before"]: - table = row["table_name"] - schemas[table] = query(conn, f"SHOW CREATE TABLE {identifier(args.target)}.{identifier(table)}")[0]["Create Table"] - write_json(args.state / "schema-before.json", schemas) - dump = Path(shutil.which("mysqldump") or Path(shutil.which("mysql")).with_name("mysqldump.exe")) - options = client_options(config, args.state) - backup = args.state / (args.target + "-migration-source.sql.gz") - try: - command = [str(dump), "--defaults-extra-file=" + str(options), "--single-transaction", "--quick", "--hex-blob", - "--set-gtid-purged=OFF", "--skip-lock-tables", args.target, *tables] - with (args.state / "backup.stderr.log").open("wb") as errors: - process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=errors) - with gzip.open(backup, "wb") as output: - shutil.copyfileobj(process.stdout, output) - if process.wait() != 0: - raise RuntimeError("备份失败，详情仅保存在本地日志") - # 完整解压检查，避免把未写完的压缩文件当作成功备份。 - with gzip.open(backup, "rb") as stream: - while stream.read(1024 * 1024): - pass - manifest["backup"] = {"path": str(backup.resolve()), "bytes": backup.stat().st_size, - "sha256": hashlib.sha256(backup.read_bytes()).hexdigest()} - write_json(args.state / "baseline.json", manifest) - print(json.dumps({"backup": manifest["backup"], "passed": True}), flush=True) - finally: - options.unlink(missing_ok=True) - - -def migrate(args, config, sql): - manifest = json.loads((args.state / "baseline.json").read_text(encoding="utf-8")) - if (manifest["source"], manifest["target"]) != (args.source, args.target): - raise ValueError("备份对应的数据库与本次选择不符") - backup = Path(manifest["backup"]["path"]) - if hashlib.sha256(backup.read_bytes()).hexdigest() != manifest["backup"]["sha256"]: - raise ValueError("备份文件校验失败") - statements = re.findall(r"(?m)^USE `[^`]+`;", sql) - if len(statements) != 1: - raise ValueError("SQL 必须只有一个显式 USE 入口") - prepared = sql.replace(statements[0], "USE " + identifier(args.target) + ";") - (args.state / "executed-migration.sql").write_text(prepared, encoding="utf-8") - with connect(config) as conn: - actual_server = query(conn, "SELECT @@hostname hostname,@@version version")[0] - if actual_server != manifest["server"]: - raise ValueError("连接服务器与备份时不同") - # 仅首次执行需要核对原始基线；重复执行由迁移 SQL 的完成账本保护。 - migrated = query(conn, "SELECT 1 FROM information_schema.tables WHERE table_schema=%s AND table_name='sys_users'", (args.target,)) - if not migrated: - for item in manifest["checks"]: - if fingerprint(conn, args.target, item["table"]) != item["fingerprints"][1]: - raise ValueError("目标数据在备份后变化：" + item["table"]) - options = client_options(config, args.state) - try: - with (args.state / "migration.stdout.log").open("wb") as output, (args.state / "migration.stderr.log").open("wb") as errors: - result = subprocess.run([shutil.which("mysql"), "--defaults-extra-file=" + str(options), "--batch", "--binary-mode"], - input=prepared.encode("utf-8"), stdout=output, stderr=errors) - print(json.dumps({"target": args.target, "exit_code": result.returncode, - "sql_sha256": hashlib.sha256(prepared.encode()).hexdigest()}), flush=True) - if result.returncode: - raise RuntimeError("迁移 SQL 失败，请检查本地日志；不会自动覆盖部分迁移状态") - finally: - options.unlink(missing_ok=True) - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("action", choices=("prepare", "migrate")) - parser.add_argument("--config", type=Path, default=Path("configs/config.yaml")) - parser.add_argument("--sql", type=Path, default=Path("docs/sw6-to-kra/sw_to_kra_migrate.sql")) - parser.add_argument("--source", default="sw8") - parser.add_argument("--target", default="sw9") - parser.add_argument("--state", type=Path, required=True) - args = parser.parse_args() - identifier(args.source) - identifier(args.target) - if args.source == args.target or args.target in ("sw", "sw6"): - raise ValueError("演练目标必须为独立测试副本") - args.state = args.state.resolve() - if args.state.is_relative_to(Path(__file__).resolve().parents[1]): - raise ValueError("含账号数据的备份必须保存在仓库外") - args.state.mkdir(parents=True, exist_ok=True) - config = yaml.safe_load(args.config.read_text(encoding="utf-8"))["data"]["database"] - sql = args.sql.read_text(encoding="utf-8") - {"prepare": prepare, "migrate": migrate}[args.action](args, config, sql) - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_endpoint_benchmark.py b/scripts/sw_endpoint_benchmark.py deleted file mode 100644 index a0f7f60..0000000 --- a/scripts/sw_endpoint_benchmark.py +++ /dev/null @@ -1,76 +0,0 @@ -"""只读请求商务接口，记录耗时、完整数据摘要及 SQL；敏感内容必须存放仓库外。""" - -import argparse -import hashlib -import json -import re -import statistics -import time -import urllib.parse -import urllib.request -from pathlib import Path - - -def log_offsets(directory): - return {p: p.stat().st_size for p in directory.rglob("*.log")} - - -def read_queries(directory, offsets): - queries = [] - files = list(directory.rglob("*.log")) - files = [file for file in files if file.parent.name == "sql"] or files - for file in files: - with file.open("rb") as stream: - stream.seek(offsets.get(file, 0)) - for line in stream: - try: - row = json.loads(line) - except (ValueError, UnicodeDecodeError): - continue - if row.get("gorm_logger") and "sql" in row: - queries.append(row) - return queries - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--base-url", default="http://127.0.0.1:19004") - parser.add_argument("--phase", required=True) - parser.add_argument("--repeat", type=int, default=2) - parser.add_argument("--cases", type=Path) - args = parser.parse_args() - if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]): - raise ValueError("完整接口响应和 SQL 必须留在仓库外") - output = args.runtime / args.phase - output.mkdir(exist_ok=True) - cases = json.loads((args.cases or args.runtime / "cases.json").read_text(encoding="utf-8")) - token = (args.runtime / "new-admin.token").read_text(encoding="utf-8").strip() - report = [] - for case in cases: - elapsed = [] - for round_number in range(args.repeat): - offsets = log_offsets(args.runtime / "sql-logs") - url = args.base_url + case["path"] + "?" + urllib.parse.urlencode(case["params"], doseq=True) - request = urllib.request.Request(url, headers={"x-token": token, "X-Request-Id": f"perf-{args.phase}-{case['name']}-{round_number}"}) - started = time.monotonic() - with urllib.request.urlopen(request, timeout=300) as response: - body = json.load(response) - elapsed.append(round((time.monotonic() - started) * 1000, 2)) - if body.get("code") != 0: - raise RuntimeError("请求失败：" + case["name"]) - queries = read_queries(args.runtime / "sql-logs", offsets) - data = body["data"] - capture = {"case": case, "data": data, "elapsedMs": elapsed[-1], "queries": queries} - (output / f"{case['name']}-{round_number}.json").write_text(json.dumps(capture, ensure_ascii=False, indent=2), encoding="utf-8") - digest = hashlib.sha256(json.dumps(data, sort_keys=True, ensure_ascii=False).encode()).hexdigest() - slow = sorted(queries, key=lambda row: row.get("elapsed_ms", 0), reverse=True)[:5] - item = {"case": case["name"], "round": round_number, "elapsedMs": elapsed[-1], "dataSha256": digest, "queryCount": len(queries), "slowQueries": [{"elapsedMs": q.get("elapsed_ms"), "rows": q.get("rows"), "tables": list(dict.fromkeys(re.findall(r"(?:FROM|JOIN)\s+`?([\w.]+)", q["sql"], re.I)))} for q in slow]} - report.append(item) - (output / "report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") - print(json.dumps(item), flush=True) - print(json.dumps({"case": case["name"], "medianMs": statistics.median(elapsed)}), flush=True) - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_endpoint_parity.py b/scripts/sw_endpoint_parity.py deleted file mode 100644 index ed8abe7..0000000 --- a/scripts/sw_endpoint_parity.py +++ /dev/null @@ -1,79 +0,0 @@ -"""在同一个 sw9 快照上核对优化前后接口，涵盖分页/排序及统计偏好。""" - -import argparse -import json -import urllib.parse -import urllib.request -from pathlib import Path - -import yaml - -from sw_database_rehearsal import connect, query - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--before-url", default="http://127.0.0.1:19002") - parser.add_argument("--after-url", default="http://127.0.0.1:19004") - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]): - raise ValueError("原始响应必须保存在仓库外") - config = yaml.safe_load((args.runtime / "config.yaml").read_text(encoding="utf-8"))["data"]["database"] - if config["name"] != "sw9": - raise ValueError("统计偏好测试仅允许 sw9") - folder = args.runtime / "parity" - folder.mkdir(exist_ok=True) - reports = [] - - def api(base, token, path, params=None, payload=None): - url = base + path + ("?" + urllib.parse.urlencode(params, doseq=True) if params else "") - request = urllib.request.Request(url, headers={"x-token": token, "Content-Type": "application/json"}, data=json.dumps(payload).encode() if payload is not None else None, method="PUT" if payload is not None else "GET") - with urllib.request.urlopen(request, timeout=300) as response: - data = json.load(response) - if data.get("code") != 0: - raise RuntimeError("接口请求失败：" + path) - return data["data"] - - def compare(case, token, group): - values = [api(base, token, case["path"], case["params"]) for base in (args.before_url, args.after_url)] - name = group + "-" + case["name"] - (folder / (name + ".json")).write_text(json.dumps({"case": case, "before": values[0], "after": values[1]}, ensure_ascii=False, indent=2), encoding="utf-8") - row = {"case": name, "passed": values[0] == values[1]} - reports.append(row) - args.output.write_text(json.dumps({"checks": reports, "passed": all(row["passed"] for row in reports)}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") - print(json.dumps(row), flush=True) - if not row["passed"]: - raise AssertionError("接口数据不一致：" + name) - - admin = (args.runtime / "new-admin.token").read_text(encoding="utf-8").strip() - for case in json.loads((args.runtime / "parity-cases.json").read_text(encoding="utf-8")): - compare(case, admin, "admin") - - token = (args.runtime / "new-level3.token").read_text(encoding="utf-8").strip() - user = api(args.before_url, token, "/user/getUserInfo")["userInfo"]["ID"] - original_preference = api(args.before_url, token, "/sw/user/preference") - cases = json.loads((args.runtime / "cases.json").read_text(encoding="utf-8")) - with connect(config) as db: - original = query(db, "SELECT * FROM sw9.sys_user_preference WHERE user_id=%s", (user,)) - try: - for include, dedup in ((True, True), (True, False), (False, True), (False, False)): - api(args.before_url, token, "/sw/user/preference", payload={"includeSubPromote": include, "idcardDedup": dedup}) - for case in cases: - compare(case, token, f"level3-sub{int(include)}-dedup{int(dedup)}") - finally: - try: - api(args.before_url, token, "/sw/user/preference", payload=original_preference) - finally: - # Restore the row even if the service stopped during verification. - if original: - row = original[0] - query(db, "UPDATE sw9.sys_user_preference SET include_sub_promote=%s,idcard_dedup=%s,update_time=%s WHERE user_id=%s", (row["include_sub_promote"], row["idcard_dedup"], row["update_time"], user)) - else: - query(db, "DELETE FROM sw9.sys_user_preference WHERE user_id=%s", (user,)) - assert query(db, "SELECT * FROM sw9.sys_user_preference WHERE user_id=%s", (user,)) == original - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_export_compare.py b/scripts/sw_export_compare.py deleted file mode 100644 index 2ed5eee..0000000 --- a/scripts/sw_export_compare.py +++ /dev/null @@ -1,76 +0,0 @@ -"""以同一主账号的实际数据请求新旧明细导出，并逐单元格核对，文件保留在仓库外。""" - -import argparse -import decimal -import io -import json -import time -import urllib.parse -import urllib.request -from pathlib import Path - -import openpyxl - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--old-url", default="http://127.0.0.1:18080") - parser.add_argument("--new-url", default="http://127.0.0.1:19001") - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]): - raise ValueError("导出文件必须保存在仓库外") - client = json.loads((args.runtime / "legacy-client.json").read_text(encoding="utf-8"))["client_id"] - headers = [{"Authorization": "Bearer " + (args.runtime / "legacy-admin.token").read_text().strip(), "clientid": client}, - {"x-token": (args.runtime / "new-admin.token").read_text().strip()}] - reports = [] - for kind, time_key in (("register", "Register"), ("spend", "Pay")): - params = {"sourceDb": "lx", "pageSize": 10, "begin" + time_key + "Time": "2026-11-03", "end" + time_key + "Time": "2026-11-03"} - request = urllib.request.Request(args.new_url + "/sw/" + kind + "/list?" + urllib.parse.urlencode(params), headers=headers[1]) - with urllib.request.urlopen(request, timeout=60) as response: - body = json.load(response) - if body.get("code") != 0: - raise RuntimeError("读取导出样本失败：" + kind) - sample = next(row for row in body["data"]["rows"] if row.get("userId") is not None) - params["userId"] = sample["userId"] - sheets, durations = [], [] - for i, base in enumerate((args.old_url, args.new_url)): - payload = urllib.parse.urlencode(params).encode() - request = urllib.request.Request(base + ("/system/" if i == 0 else "/sw/") + kind + "/export", data=payload, - headers={**headers[i], "Content-Type": "application/x-www-form-urlencoded"}, method="POST") - started = time.monotonic() - with urllib.request.urlopen(request, timeout=120) as response: - raw = response.read() - durations.append(round((time.monotonic() - started) * 1000)) - if not raw.startswith(b"PK"): - raise RuntimeError("导出未返回工作簿：" + kind + "/" + str(i)) - (args.runtime / (("legacy-" if i == 0 else "new-") + kind + "-export.xlsx")).write_bytes(raw) - workbook = openpyxl.load_workbook(io.BytesIO(raw), read_only=True, data_only=True) - sheets.append(list(workbook.worksheets[0].values)) - workbook.close() - differences = [] - for row_number, (old, new) in enumerate(zip(*sheets), 1): - if len(old) != len(new): - differences.append({"row": row_number, "kind": "columns"}) - for col, (a, b) in enumerate(zip(old, new), 1): - if a == b or (a in (None, "") and b in (None, "")): - continue - if isinstance(a, (int, float)) or isinstance(b, (int, float)): - try: - if decimal.Decimal(str(a)) == decimal.Decimal(str(b)): - continue - except decimal.InvalidOperation: - pass - differences.append({"row": row_number, "column": col, "kind": "value"}) - rows = [len(sheet) - 1 for sheet in sheets] - reports.append({"kind": kind, "rows": rows, "columns": [len(sheet[0]) for sheet in sheets], "elapsed_ms": durations, - "differences": differences, "passed": rows[0] > 0 and rows[0] == rows[1] and not differences}) - print(json.dumps(reports[-1], ensure_ascii=False), flush=True) - args.output.write_text(json.dumps({"checks": reports, "passed": all(row["passed"] for row in reports)}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") - if not all(row["passed"] for row in reports): - raise SystemExit(1) - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_preference_compare.py b/scripts/sw_preference_compare.py deleted file mode 100644 index 4993e72..0000000 --- a/scripts/sw_preference_compare.py +++ /dev/null @@ -1,61 +0,0 @@ -"""用已有真实绑定的三级账号逐项切换统计偏好，对比报表后恢复原始偏好行。""" - -import argparse -import json -import subprocess -import sys -from pathlib import Path - -from sw_write_verify import Rehearsal - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--config", type=Path, default=Path("configs/config.yaml")) - parser.add_argument("--old-url", default="http://127.0.0.1:18080") - parser.add_argument("--new-url", default="http://127.0.0.1:19001") - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - run = Rehearsal(args) - identity = json.loads((args.runtime / "test-business-identities.json").read_text())["level3"] - ids = {"legacy": identity["user_id"], "new": run.mapping[str(identity["user_id"])]} - originals, preferences = {}, {} - for side, db in (("legacy", "sw8"), ("new", "sw9")): - originals[side] = run.query(f"SELECT * FROM {db}.sys_user_preference WHERE user_id=%s", (ids[side],)) - preferences[side] = run.call(side, "level3", "GET", run.prefix(side) + "/user/preference")["data"] - (args.runtime / "preferences-before-matrix.json").write_text(json.dumps(originals, ensure_ascii=False, indent=2, default=str), encoding="utf-8") - templates = json.loads((args.runtime / "level3-business-cases.json").read_text()) - templates = [row for row in templates if row["name"] in ("channelData-list", "gameData-list", "rank-game", "workbench-summary")] - try: - for include, dedup in ((True, True), (True, False), (False, True), (False, False)): - mode = f"sub{int(include)}-dedup{int(dedup)}" - value = {"includeSubPromote": include, "idcardDedup": dedup} - for side in ("legacy", "new"): - run.call(side, "level3", "PUT", run.prefix(side) + "/user/preference", value) - actual = run.call(side, "level3", "GET", run.prefix(side) + "/user/preference")["data"] - run.check("偏好写入 " + mode + " " + side, actual == value) - cases = [{**row, "name": mode + "-" + row["name"]} for row in templates] - cases_file = args.runtime / ("preference-cases-" + mode + ".json") - cases_file.write_text(json.dumps(cases), encoding="utf-8") - output = args.output.with_name(args.output.stem + "-" + mode + args.output.suffix) - with (args.runtime / ("preference-" + mode + ".log")).open("wb") as log: - completed = subprocess.run([sys.executable, str(Path(__file__).with_name("sw_api_compare.py")), "--runtime", str(args.runtime), "--cases", str(cases_file), - "--identity", "level3", "--old-url", args.old_url, "--new-url", args.new_url, "--output", str(output)], stdout=log, stderr=subprocess.STDOUT) - run.check("偏好模式报表一致 " + mode, completed.returncode == 0, report=output.name) - finally: - for side, db in (("legacy", "sw8"), ("new", "sw9")): - run.call(side, "level3", "PUT", run.prefix(side) + "/user/preference", preferences[side]) - if not originals[side]: - run.query(f"DELETE FROM {db}.sys_user_preference WHERE user_id=%s", (ids[side],)) - else: - row = originals[side][0] - run.query(f"UPDATE {db}.sys_user_preference SET include_sub_promote=%s,idcard_dedup=%s,update_time=%s WHERE user_id=%s", - (row["include_sub_promote"], row["idcard_dedup"], row["update_time"], ids[side])) - actual = run.query(f"SELECT * FROM {db}.sys_user_preference WHERE user_id=%s", (ids[side],)) - run.check("偏好原始行已恢复 " + side, actual == originals[side]) - run.db.close() - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_report_rehearsal.py b/scripts/sw_report_rehearsal.py deleted file mode 100644 index 9562811..0000000 --- a/scripts/sw_report_rehearsal.py +++ /dev/null @@ -1,119 +0,0 @@ -"""备份/比较/恢复 sw8 与 sw9 指定自然日的四张报表，不备份或改写原始业务流水。""" - -import argparse -import datetime as dt -import gzip -import hashlib -import json -import shutil -import subprocess -from pathlib import Path - -import yaml - -from sw_database_rehearsal import client_options, connect, query, write_json - -TABLES = ("tab_sw_report_day", "tab_sw_report_day_key", "tab_sw_report_server_day", "tab_sw_report_server_day_key") - - -def snapshot(conn, database, day, logical=False): - result = {} - for table in TABLES: - rows = query(conn, f"SELECT * FROM `{database}`.`{table}` WHERE stat_date=%s ORDER BY id", (day,)) - if logical: - rows = [{key: value for key, value in row.items() if key not in ("id", "created_at", "updated_at")} for row in rows] - serialized = [json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str) for row in rows] - if logical: - serialized.sort() - result[table] = {"rows": len(rows), "sha256": hashlib.sha256("\n".join(serialized).encode()).hexdigest()} - return result - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("action", choices=("backup", "verify", "restore")) - parser.add_argument("--config", type=Path, default=Path("configs/config.yaml")) - parser.add_argument("--date", default="2026-11-03") - parser.add_argument("--state", type=Path, required=True) - parser.add_argument("--output", type=Path) - args = parser.parse_args() - dt.date.fromisoformat(args.date) - args.state = args.state.resolve() - if args.state.is_relative_to(Path(__file__).resolve().parents[1]): - raise ValueError("报表备份和身份键必须保存在仓库外") - args.state.mkdir(parents=True, exist_ok=True) - config = yaml.safe_load(args.config.read_text(encoding="utf-8"))["data"]["database"] - manifest_path = args.state / "report-baseline.json" - if args.action == "backup": - if manifest_path.exists(): - raise ValueError("已有基线备份，禁止覆盖") - manifest = {"date": args.date, "databases": {}} - with connect(config) as conn: - manifest["server"] = query(conn, "SELECT @@hostname hostname,@@version version")[0] - for database in ("sw8", "sw9"): - baseline = snapshot(conn, database, args.date) - options = client_options(config, args.state) - file = args.state / (database + "-report-partition.sql.gz") - try: - with (args.state / (database + "-dump.stderr.log")).open("wb") as log: - process = subprocess.Popen([shutil.which("mysqldump"), "--defaults-extra-file=" + str(options), "--single-transaction", "--quick", "--hex-blob", - "--no-create-info", "--skip-triggers", "--skip-add-locks", "--skip-disable-keys", "--complete-insert", "--set-gtid-purged=OFF", - "--where=stat_date='" + args.date + "'", database, *TABLES], stdout=subprocess.PIPE, stderr=log) - with gzip.open(file, "wb") as target: - shutil.copyfileobj(process.stdout, target) - if process.wait() != 0: - raise RuntimeError("报表备份失败") - with gzip.open(file, "rb") as stream: - while stream.read(1024 * 1024): - pass - manifest["databases"][database] = {"tables": baseline, "backup": str(file), "sha256": hashlib.sha256(file.read_bytes()).hexdigest()} - finally: - options.unlink(missing_ok=True) - write_json(manifest_path, manifest) - print(json.dumps({"date": args.date, "tables": {db: value["tables"] for db, value in manifest["databases"].items()}}, ensure_ascii=False), flush=True) - return - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - if manifest["date"] != args.date: - raise ValueError("日期与备份不符") - with connect(config) as conn: - if query(conn, "SELECT @@hostname hostname,@@version version")[0] != manifest["server"]: - raise ValueError("数据库服务器与备份不符") - if args.action == "restore": - for database, value in manifest["databases"].items(): - if database not in ("sw8", "sw9"): - raise ValueError("未知数据库") - file = Path(value["backup"]) - if hashlib.sha256(file.read_bytes()).hexdigest() != value["sha256"]: - raise ValueError("备份校验失败") - content = gzip.decompress(file.read_bytes()) - if b"LOCK TABLES" in content or b"ALTER TABLE" in content or b"CREATE TABLE" in content: - raise ValueError("分区恢复文件不允许包含 DDL 或隐式提交") - deletes = "\n".join(f"DELETE FROM `{table}` WHERE stat_date='{args.date}';" for table in TABLES) - script = ("START TRANSACTION;\n" + deletes + "\n").encode() + content + b"\nCOMMIT;\n" - options = client_options(config, args.state) - try: - completed = subprocess.run([shutil.which("mysql"), "--defaults-extra-file=" + str(options), "--batch", database], input=script, capture_output=True) - (args.state / (database + "-restore.stderr.log")).write_bytes(completed.stderr) - if completed.returncode: - raise RuntimeError("分区恢复失败") - finally: - options.unlink(missing_ok=True) - with connect(config) as conn: - checks = [{"database": db, "passed": snapshot(conn, db, args.date) == value["tables"]} for db, value in manifest["databases"].items()] - else: - with connect(config) as conn: - source, target = [snapshot(conn, db, args.date, logical=True) for db in ("sw8", "sw9")] - checks = [{"table": table, "source": source[table], "target": target[table], "passed": source[table] == target[table]} for table in TABLES] - for db in ("sw8", "sw9"): - for table in TABLES: - rows = query(conn, f"SELECT * FROM `{db}`.`{table}` WHERE stat_date=%s ORDER BY id", (args.date,)) - write_json(args.state / (db + "-" + table + "-rebuilt.json"), rows) - result = {"action": args.action, "date": args.date, "checks": checks, "passed": all(item["passed"] for item in checks)} - write_json(args.output or args.state / (args.action + ".json"), result) - print(json.dumps(result, ensure_ascii=False), flush=True) - if not result["passed"]: - raise SystemExit(1) - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_run_report_tasks.py b/scripts/sw_run_report_tasks.py deleted file mode 100644 index 44ba3f2..0000000 --- a/scripts/sw_run_report_tasks.py +++ /dev/null @@ -1,58 +0,0 @@ -"""通过新商务原生任务 HTTP 重跑指定日期，要求已有分区备份并保持任务禁用。""" - -import argparse -import json -import time -from pathlib import Path - -from sw_write_verify import Rehearsal - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--config", type=Path, default=Path("configs/config.yaml")) - parser.add_argument("--old-url", default="http://127.0.0.1:18080") - parser.add_argument("--new-url", default="http://127.0.0.1:19001") - parser.add_argument("--output", type=Path, required=True) - parser.add_argument("--date", default="2026-11-03") - parser.add_argument("--backup", type=Path, required=True) - parser.add_argument("--methods", nargs="+", choices=("swReportDayBackfillTask", "swReportServerDayBackfillTask", "alertTriggerTask"), - default=["swReportDayBackfillTask", "swReportServerDayBackfillTask"]) - args = parser.parse_args() - backup = json.loads(args.backup.read_text(encoding="utf-8")) - if backup["date"] != args.date: - raise ValueError("已有备份的日期与任务不符") - run = Rehearsal(args) - ids = [] - try: - for method in args.methods: - name = "迁移核对-" + method + "-" + str(int(time.time())) - run.call("new", "admin", "POST", "/timedTask/createTimedTask", {"name": name, "spec": "@daily", "executorType": "method", "methodName": method, - "params": {"startDate": args.date, "endDate": args.date, "_execution": {"timeoutSeconds": 1800}}, "enabled": False}) - task = run.query("SELECT id,enabled FROM sw9.sys_timed_tasks WHERE name=%s AND deleted_at IS NULL", (name,))[0] - ids.append(task["id"]) - (args.runtime / "report-rehearsal-task-ids.json").write_text(json.dumps(ids), encoding="utf-8") - run.check(method + " 创建且保持禁用", task["enabled"] == 0) - started = time.monotonic() - run.call("new", "admin", "POST", "/timedTask/triggerTimedTask", {"id": task["id"]}) - while time.monotonic() - started < 1830: - value = run.call("new", "admin", "GET", f"/timedTask/getTimedTaskLogList?taskId={task['id']}&page=1&pageSize=1")["data"] - logs = value.get("list", []) - if logs and logs[0].get("status") not in ("running", "pending"): - (args.runtime / (method + "-execution.json")).write_text(json.dumps(logs[0], ensure_ascii=False, indent=2), encoding="utf-8") - run.check(method + " 真实执行完成", logs[0]["status"] == "success", status=logs[0]["status"], elapsed_seconds=round(time.monotonic() - started, 2)) - break - time.sleep(2) - else: - run.check(method + " 等待执行完成", False, reason="timeout") - except Exception as error: - if all(check["passed"] for check in run.checks): - run.check("报表重跑执行异常", False, error_type=type(error).__name__) - raise - finally: - run.db.close() - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_sdk_write_verify.py b/scripts/sw_sdk_write_verify.py deleted file mode 100644 index b9070f0..0000000 --- a/scripts/sw_sdk_write_verify.py +++ /dev/null @@ -1,118 +0,0 @@ -"""主库使用 sw8/sw9，平台写入使用独立本地 SDK 夹具；核对验密、重提、解绑与新增渠道。""" - -import argparse -import json -import time -from pathlib import Path - -import pymysql - -from sw_api_compare import compare_data -from sw_write_verify import Rehearsal - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--config", type=Path, default=Path("configs/config.yaml")) - parser.add_argument("--old-url", default="http://127.0.0.1:18081") - parser.add_argument("--new-url", default="http://127.0.0.1:19002") - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - run = Rehearsal(args) - fixture = json.loads((args.runtime / "sdk-fixture/fixture.json").read_text()) - owner = fixture["owner_id"] - local = pymysql.connect(host="127.0.0.1", port=13309, user="root", autocommit=True, cursorclass=pymysql.cursors.DictCursor) - channel = "migchannel" + str(int(time.time())) - - def query(sql, params=()): - with local.cursor() as cursor: - cursor.execute(sql, params) - return cursor.fetchall() - - def api_path(side, path): - return ("/system/user" if side == "legacy" else "/sw/business") + path - - def bind(side, password=None, success=True, identity="testlevel3"): - return run.call(side, identity, "POST", api_path(side, "/bindPlatform"), - {"sourceDb": "lx", "account": fixture["account"], "password": password or fixture["password"]}, success=success) - - def snapshot(label, status): - values = [] - for side, database in (("legacy", "sw8"), ("new", "sw9")): - rows = run.query(f"SELECT user_id AS userId,source_db AS sourceDb,platform_busier_id AS platformBusierId,platform_busier_account AS platformBusierAccount,status,reject_reason AS rejectReason,review_account AS reviewAccount FROM {database}.sys_user_bind_platform WHERE source_db='lx' AND platform_busier_id=%s", (owner,)) - run.check(label + " 行数和状态 " + side, len(rows) == 1 and rows[0]["status"] == status) - values.append(rows) - compared = compare_data(*values, {}, run.mapping) - run.check(label + " 字段一致", compared["semantic_equal"], differences=compared["differences"]) - - try: - for database in ("sw_test_sdk_old", "sw_test_sdk_new"): - for table in ("tab_promote", "tab_promote_settlement_time"): - run.check("本地平台写入夹具为空 " + database + "." + table, query(f"SELECT COUNT(*) n FROM {database}.{table}")[0]["n"] == 0) - query(f"ALTER TABLE {database}.{table} AUTO_INCREMENT=1") - for side, database in (("legacy", "sw8"), ("new", "sw9")): - run.check("夹具绑定键未占用 " + side, not run.query(f"SELECT user_id FROM {database}.sys_user_bind_platform WHERE source_db='lx' AND platform_busier_id=%s", (owner,))) - denied = bind(side, password="invalid-password", success=False) - run.check("错误密码不允许绑定 " + side, denied["code"] != (200 if side == "legacy" else 0)) - denied = bind(side, success=False, identity="testlevel1") - run.check("仅三级允许绑定 " + side, denied["code"] != (200 if side == "legacy" else 0)) - started = time.monotonic() - for side in ("legacy", "new"): - bind(side) - snapshot("首次绑定待审核", "3") - for side in ("legacy", "new"): - denied = bind(side, success=False) - run.check("待审核重复提交被拒绝 " + side, denied["code"] != (200 if side == "legacy" else 0)) - # 仅在主库夹具模拟 OA 的审核结果，不调用外部 OA 写接口。 - for database in ("sw8", "sw9"): - run.query(f"UPDATE {database}.sys_user_bind_platform SET status='2',reject_reason='测试驳回',review_account='migration-test',review_time=NOW() WHERE source_db='lx' AND platform_busier_id=%s", (owner,)) - time.sleep(max(0, 5.2 - (time.monotonic() - started))) - for side in ("legacy", "new"): - bind(side) - snapshot("驳回后重提并清空旧审核信息", "3") - for database in ("sw8", "sw9"): - run.query(f"UPDATE {database}.sys_user_bind_platform SET status='0',review_account='migration-test',review_time=NOW() WHERE source_db='lx' AND platform_busier_id=%s", (owner,)) - snapshot("审核通过绑定", "0") - payload = {"sourceDb": "lx", "platformBusierId": owner, "account": channel, "password": "Channel168", "registerType": 0} - for side in ("legacy", "new"): - denied = run.call(side, "testlevel1", "POST", run.prefix(side) + "/promote", payload, success=False) - run.check("保留原一级角色的新增渠道权限限制 " + side, denied["code"] != (200 if side == "legacy" else 0)) - run.call(side, "admin", "POST", run.prefix(side) + "/promote", payload) - saved, settlements = [], [] - for side, database in (("legacy", "sw_test_sdk_old"), ("new", "sw_test_sdk_new")): - rows = query(f"SELECT id,account,password,status,busier_id,register_type,promote_level,parent_id,top_promote_id,game_ids2 FROM {database}.tab_promote WHERE account=%s", (channel,)) - run.check("源平台渠道只创建一条 " + side, len(rows) == 1) - saved.append(rows[0]) - settlement = query(f"SELECT * FROM {database}.tab_promote_settlement_time WHERE promote_id=%s", (rows[0]["id"],)) - run.check("七日结算及时间关系 " + side, len(settlement) == 1 and settlement[0]["day_period"] == 7 and settlement[0]["time_type"] == 0 and settlement[0]["next_count_time"] - settlement[0]["create_time"] == 7 * 86400) - relation = query(f"SELECT promote_ids FROM {database}.tab_promote_business WHERE id=%s", (owner,))[0]["promote_ids"] - run.check("平台商务渠道关系已追加 " + side, str(rows[0]["id"]) in (relation or "").split(",")) - run.check("新渠道属性及兼容密码完全一致", saved[0] == saved[1] and saved[0]["status"] == -3 and saved[0]["promote_level"] == 1) - for side, database in (("legacy", "sw8"), ("new", "sw9")): - denied = run.call(side, "admin", "POST", run.prefix(side) + "/promote", payload, success=False) - run.check("重复渠道账号被拒绝 " + side, denied["code"] != (200 if side == "legacy" else 0)) - run.call(side, "testlevel3", "DELETE", api_path(side, "/bindPlatform"), {"sourceDb": "lx", "platformBusierId": owner}) - run.check("解绑只删除当前绑定 " + side, not run.query(f"SELECT user_id FROM {database}.sys_user_bind_platform WHERE user_id=%s AND source_db='lx' AND platform_busier_id=%s", (run.owner(side, 3), owner))) - except Exception as error: - if all(check["passed"] for check in run.checks): - run.check("SDK 验证执行异常", False, error_type=type(error).__name__) - raise - finally: - for side, database in (("legacy", "sw8"), ("new", "sw9")): - run.query(f"DELETE FROM {database}.sys_user_bind_platform WHERE user_id=%s AND source_db='lx' AND platform_busier_id=%s AND platform_busier_account=%s", (run.owner(side, 3), owner, fixture["account"])) - run.query("DELETE FROM sw9.sys_business_binding_locks WHERE source_db='lx' AND platform_busier_id=%s", (owner,)) - for database in ("sw_test_sdk_old", "sw_test_sdk_new"): - rows = query(f"SELECT id FROM {database}.tab_promote WHERE account=%s", (channel,)) - for row in rows: - query(f"DELETE FROM {database}.tab_promote_settlement_time WHERE promote_id=%s", (row["id"],)) - query(f"DELETE FROM {database}.tab_promote WHERE account=%s", (channel,)) - query(f"UPDATE {database}.tab_promote_business SET promote_ids='' WHERE id=%s AND account=%s", (owner, fixture["account"])) - remaining = sum(run.query(f"SELECT COUNT(*) n FROM {db}.sys_user_bind_platform WHERE source_db='lx' AND platform_busier_id=%s", (owner,))[0]["n"] for db in ("sw8", "sw9")) - run.check("主库 SDK 绑定夹具已清理", remaining == 0) - local.close() - run.db.close() - - -if __name__ == "__main__": - main() diff --git a/scripts/sw_write_verify.py b/scripts/sw_write_verify.py deleted file mode 100644 index 57395e3..0000000 --- a/scripts/sw_write_verify.py +++ /dev/null @@ -1,207 +0,0 @@ -"""在 sw8/sw9 专用测试账号下核对写流程；临时渠道、绑定、预警及规则在 finally 中清理。""" - -import argparse -import json -import time -import urllib.error -import urllib.parse -import urllib.request -from pathlib import Path - -import pymysql -import yaml - -from sw_api_compare import compare_data - - -class Rehearsal: - def __init__(self, args): - self.args = args - self.checks = [] - args.output.write_text(json.dumps({"checks": [], "passed": False, "status": "running"}) + "\n", encoding="utf-8") - self.root = args.runtime - self.users = json.loads((self.root / "synthetic-test-users.json").read_text(encoding="utf-8")) - self.tokens = {(side, identity): (self.root / f"{side}-{identity}.token").read_text().strip() - for side in ("legacy", "new") for identity in ("admin", "level3", "testlevel1", "testlevel2", "testlevel3")} - self.client = json.loads((self.root / "legacy-client.json").read_text())["client_id"] - maps = json.loads((self.root / "id-maps.json").read_text()) - self.mapping = {str(row["old_id"]): str(row["new_id"]) for row in maps if row["entity"] == "user"} - config = yaml.safe_load(args.config.read_text(encoding="utf-8"))["data"]["database"] - self.db = pymysql.connect(host=config["host"], port=int(config["port"]), user=config["user"], password=config["password"], - autocommit=True, charset="utf8mb4", read_timeout=60, cursorclass=pymysql.cursors.DictCursor) - self.rule_ids = {} - self.suffix = str(int(time.time())) - self.rule_type = "mig_verify_" + self.suffix - self.account = "mig_verify_" + self.suffix - self.promote_id = 2000000000 + int(self.suffix[-7:]) - self.alert_id = 9007199254740993 + int(self.suffix[-7:]) - self.fixtures = [] - - def query(self, sql, params=()): - with self.db.cursor() as cursor: - cursor.execute(sql, params) - return cursor.fetchall() - - def check(self, name, passed, **details): - self.checks.append({"check": name, "passed": bool(passed), **details}) - self.args.output.write_text(json.dumps({"checks": self.checks, "passed": all(item["passed"] for item in self.checks)}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") - print(json.dumps(self.checks[-1], ensure_ascii=False), flush=True) - if not passed: - raise AssertionError(name) - - def call(self, side, identity, method, path, payload=None, success=True): - old = side == "legacy" - headers = {"Content-Type": "application/json"} - headers.update({"Authorization": "Bearer " + self.tokens[side, identity], "clientid": self.client} if old else {"x-token": self.tokens[side, identity]}) - request = urllib.request.Request((self.args.old_url if old else self.args.new_url) + path, - data=json.dumps(payload).encode() if payload is not None else None, method=method, headers=headers) - try: - with urllib.request.urlopen(request, timeout=60) as response: - raw = response.read() - status = response.status - except urllib.error.HTTPError as response: - raw = response.read() - status = response.code - try: - reply = json.loads(raw) - except (ValueError, UnicodeDecodeError): - self.check(path + " " + side, False, http=status, response_bytes=len(raw), error="响应不是 JSON") - if success and reply.get("code") != (200 if old else 0): - self.check(path + " " + side, False, code=reply.get("code"), message=reply.get("msg")) - return reply - - def owner(self, side, level): - return self.users[level - 1]["source_id" if side == "legacy" else "target_id"] - - def prefix(self, side): - return "/system" if side == "legacy" else "/sw" - - def rule_snapshot(self, name): - values = [self.call(side, "testlevel1", "GET", self.prefix(side) + "/alertRule/" + str(self.rule_ids[side]))["data"] for side in ("legacy", "new")] - options = {"source_omit_fields": ["id", "createdAt", "updatedAt"], "target_omit_fields": ["id", "createdAt", "updatedAt"], - "null_empty_fields": ["momDirectionLabel"]} - result = compare_data(*values, options, self.mapping) - self.check(name, result["semantic_equal"], differences=result["differences"]) - return values - - def rules(self): - self.rule = {"ruleName": "迁移核对临时规则", "ruleType": self.rule_type, "monitorMetric": 3, "unloginDays": 2, "monitorScope": 1, "warningColor": "#f56c6c"} - for side, database in (("legacy", "sw8"), ("new", "sw9")): - self.call(side, "testlevel1", "POST", self.prefix(side) + "/alertRule/add", self.rule) - rows = self.query(f"SELECT id FROM {database}.sys_alert_rule WHERE creator_id=%s AND rule_type=%s", (self.owner(side, 1), self.rule_type)) - self.check("新增规则 " + side, len(rows) == 1) - self.rule_ids[side] = rows[0]["id"] - self.rule_snapshot("新增规则字段一致") - for side in ("legacy", "new"): - self.call(side, "testlevel1", "PUT", self.prefix(side) + f"/alertRule/toggle/{self.rule_ids[side]}?enabled=0") - values = self.rule_snapshot("规则禁用一致") - self.check("禁用状态生效", all(row["enabled"] == 0 for row in values)) - self.rule.update(monitorMetric=1, statPeriodDays=3, momDirection=2, momThreshold=12.5) - self.rule.pop("unloginDays") - for side in ("legacy", "new"): - self.call(side, "testlevel1", "PUT", self.prefix(side) + "/alertRule/edit", {**self.rule, "id": self.rule_ids[side]}) - values = self.rule_snapshot("编辑保留启停和可选历史字段") - self.check("编辑未重启规则且保留未登录天数", all(row["enabled"] == 0 and row["unloginDays"] == 2 for row in values)) - for side in ("legacy", "new"): - rejected = self.call(side, "testlevel2", "PUT", self.prefix(side) + f"/alertRule/toggle/{self.rule_ids[side]}?enabled=1", success=False) - self.check("二级不能修改一级规则 " + side, rejected.get("code") != (200 if side == "legacy" else 0)) - self.call(side, "testlevel1", "PUT", self.prefix(side) + f"/alertRule/toggle/{self.rule_ids[side]}?enabled=1") - self.rule_snapshot("重新启用规则一致") - - def prepare_alert(self): - for side, database in (("legacy", "sw8"), ("new", "sw9")): - existing = self.query(f"SELECT id FROM {database}.tab_promote WHERE id=%s AND source_db='lx'", (self.promote_id,)) - self.check("临时渠道 ID 未占用 " + side, not existing) - existing = self.query(f"SELECT id FROM {database}.sys_alert_record WHERE id=%s", (self.alert_id,)) - self.check("临时预警 ID 未占用 " + side, not existing) - self.fixtures.append({"side": side, "database": database, "promote_id": self.promote_id, "alert_id": self.alert_id, "account": self.account, - "owner": self.owner(side, 1), "user": self.owner(side, 3)}) - (self.root / "write-fixtures.json").write_text(json.dumps(self.fixtures, indent=2), encoding="utf-8") - self.query(f"INSERT INTO {database}.tab_promote(id,source_db,account,busier_id,promote_level,status) VALUES(%s,'lx',%s,%s,1,1)", (self.promote_id, self.account, self.promote_id)) - self.query(f"INSERT INTO {database}.sys_user_bind_platform(user_id,source_db,platform_busier_id,platform_busier_account,status,create_time) VALUES(%s,'lx',%s,%s,'0',NOW())", (self.owner(side, 3), self.promote_id, self.account)) - self.query(f"INSERT INTO {database}.sys_alert_record(id,platform_code,platform_name,rule_id,rule_name,rule_type,monitor_metric,target_promote_id,target_promote_name,business_id,business_name,alert_content,status,alert_date,created_at,updated_at) VALUES(%s,'lx','龙翔',%s,%s,%s,1,%s,%s,%s,%s,'迁移核对临时预警',1,CURDATE(),NOW(),NOW())", - (self.alert_id, self.rule_ids[side], self.rule["ruleName"], self.rule_type, self.promote_id, self.account, self.owner(side, 1), self.users[0]["username"])) - - def alert_snapshot(self, label, expected_status): - values = [self.call(side, "testlevel1", "GET", self.prefix(side) + f"/alertRecord/detail/{self.alert_id}")["data"] for side in ("legacy", "new")] - for row in values: - row["alertRecord"].pop("ruleId", None) - options = {"source_omit_fields": ["id", "createdAt", "updatedAt"], "target_omit_fields": ["id", "createdAt", "updatedAt"], - "null_empty_fields": ["actionDetail", "noteContent"]} - # 日志主键独立生成；按原操作序列比较业务字段及操作人映射。 - result = compare_data(*values, options, self.mapping) - self.check(label, result["semantic_equal"] and all(row["alertRecord"]["status"] == expected_status for row in values), - status=expected_status, log_counts=[len(row["operationLogs"]) for row in values], differences=result["differences"]) - - def workflow(self): - self.prepare_alert() - self.alert_snapshot("临时预警初始状态一致", 1) - for side in ("legacy", "new"): - rejected = self.call(side, "level3", "GET", self.prefix(side) + f"/alertRecord/detail/{self.alert_id}", success=False) - self.check("其他商务不可查看临时预警 " + side, rejected.get("code") != (200 if side == "legacy" else 0)) - steps = [("startProcess", "testlevel3", 2), ("submitAudit", "testlevel3", 3), ("reject", "testlevel1", 4), - ("reProcess", "testlevel3", 2), ("submitAudit", "testlevel3", 3), ("approve", "testlevel1", 5)] - for index, (action, identity, status) in enumerate(steps): - for side in ("legacy", "new"): - self.call(side, identity, "POST", self.prefix(side) + "/alertRecord/" + action, {"alertId": str(self.alert_id), "note": "迁移核对步骤" + str(index)}) - self.alert_snapshot(action + " 状态及日志一致", status) - if index == 1: - for side in ("legacy", "new"): - rejected = self.call(side, "testlevel3", "POST", self.prefix(side) + "/alertRecord/approve", {"alertId": str(self.alert_id)}, success=False) - self.check("三级商务不能审核 " + side, rejected.get("code") != (200 if side == "legacy" else 0)) - for side in ("legacy", "new"): - self.call(side, "testlevel3", "POST", self.prefix(side) + "/alertRecord/addNote", {"alertId": str(self.alert_id), "noteContent": "迁移核对独立备注", "attachmentIds": []}) - self.alert_snapshot("已处理记录追加备注不改变状态", 5) - - def remove_rule(self): - for side, database in (("legacy", "sw8"), ("new", "sw9")): - self.call(side, "testlevel1", "DELETE", self.prefix(side) + "/alertRule/" + str(self.rule_ids[side])) - row = self.query(f"SELECT is_deleted FROM {database}.sys_alert_rule WHERE id=%s", (self.rule_ids[side],))[0] - self.check("删除规则保留历史行 " + side, row["is_deleted"] == 1) - - def cleanup(self): - for fixture in reversed(self.fixtures): - database = fixture["database"] - self.query(f"DELETE FROM {database}.sys_alert_operation_log WHERE alert_id=%s", (fixture["alert_id"],)) - self.query(f"DELETE FROM {database}.sys_alert_record WHERE id=%s AND target_promote_name=%s", (fixture["alert_id"], fixture["account"])) - self.query(f"DELETE FROM {database}.sys_user_bind_platform WHERE user_id=%s AND source_db='lx' AND platform_busier_id=%s AND platform_busier_account=%s", (fixture["user"], fixture["promote_id"], fixture["account"])) - self.query(f"DELETE FROM {database}.tab_promote WHERE id=%s AND source_db='lx' AND account=%s", (fixture["promote_id"], fixture["account"])) - for side, database in (("legacy", "sw8"), ("new", "sw9")): - self.query(f"DELETE FROM {database}.sys_alert_rule WHERE creator_id=%s AND rule_type=%s", (self.owner(side, 1), self.rule_type)) - remaining = 0 - for fixture in self.fixtures: - database = fixture["database"] - for table, condition, params in ( - ("sys_alert_record", "id=%s", (fixture["alert_id"],)), - ("sys_alert_operation_log", "alert_id=%s", (fixture["alert_id"],)), - ("tab_promote", "id=%s AND source_db='lx'", (fixture["promote_id"],)), - ("sys_user_bind_platform", "user_id=%s AND source_db='lx' AND platform_busier_id=%s", (fixture["user"], fixture["promote_id"])), - ): - remaining += self.query(f"SELECT COUNT(*) n FROM {database}.{table} WHERE {condition}", params)[0]["n"] - for side, database in (("legacy", "sw8"), ("new", "sw9")): - remaining += self.query(f"SELECT COUNT(*) n FROM {database}.sys_alert_rule WHERE creator_id=%s AND rule_type=%s", (self.owner(side, 1), self.rule_type))[0]["n"] - self.check("临时预警渠道绑定和规则已清理", remaining == 0, remaining=remaining) - self.db.close() - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--config", type=Path, default=Path("configs/config.yaml")) - parser.add_argument("--old-url", default="http://127.0.0.1:18080") - parser.add_argument("--new-url", default="http://127.0.0.1:19001") - parser.add_argument("--output", type=Path, required=True) - args = parser.parse_args() - if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]): - raise ValueError("测试状态必须保存在仓库外") - run = Rehearsal(args) - try: - run.rules() - run.workflow() - run.remove_rule() - finally: - run.cleanup() - - -if __name__ == "__main__": - main()