From bdc1584b3c0539dedbc7ca54d59aa192289bf659 Mon Sep 17 00:00:00 2001 From: hangshuo652 Date: Wed, 9 Sep 2026 21:35:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BF=AE=E5=A4=8D=20Java=20Runner=20co?= =?UTF-8?q?mmand=5Fline=20=E5=8F=82=E6=95=B0=E4=BC=A0=E9=80=92=20+=20DB-Ja?= =?UTF-8?q?va=20=E6=AF=94=E5=AF=B9=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 修复 orchestrator_db.py: Java Runner 未传递 command_line 参数导致 ABEND - 新增 DB-Java 文件式运行 + DB 表比对功能 - 优化输出目录结构: output//cobol/ - 新增测试文件: test_java_comparison.py, test_java_e2e.py - 更新 AI 使用日志 --- _AI_USAGE_LOG.md | 109 ++++++ cobol_testgen/__init__.py | 52 ++- cobol_testgen/runner.py | 452 +++++++++++++++++++++- comparator/aligner.py | 33 +- docs/output-directory-structure.md | 400 +++++++++++++++++++ orchestrator.py | 8 +- orchestrator_db.py | 592 +++++++++++++++++++++++++++-- runners/gixsql_runner.py | 72 +++- runners/native_java_runner.py | 225 ++++++++++- runners/spark_java_runner.py | 235 ++++++++++-- tests/fixtures/java/pom.xml | 34 +- tests/test_java_comparison.py | 375 ++++++++++++++++++ tests/test_java_e2e.py | 174 +++++++++ tools/registry.py | 4 +- 14 files changed, 2652 insertions(+), 113 deletions(-) create mode 100644 docs/output-directory-structure.md create mode 100644 tests/test_java_comparison.py create mode 100644 tests/test_java_e2e.py diff --git a/_AI_USAGE_LOG.md b/_AI_USAGE_LOG.md index fd6d88f..25f8371 100644 --- a/_AI_USAGE_LOG.md +++ b/_AI_USAGE_LOG.md @@ -13,6 +13,115 @@ --- +### 2026-09-09 21:30:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** 修复 Java Runner 未传递 command_line 参数的 bug：在 `orchestrator_db.py` 的 `_java_run_scenario` 方法中添加 command_line 参数传递逻辑，对齐 COBOL runner 的处理方式。修改后 Java 程序能正确接收 YEARMONTH 参数。 +- **涉及文件：** `orchestrator_db.py:1036-1047` +- **使用模型：** opencode/mimo-v2.5-free + +### 2026-09-08 21:55:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** ZAN06UPD DB 提交持久化问题根因修复（方案 A：GIXSQL_AUTOCOMMIT=ON）： + 1) `runners/gixsql_runner.py` `_patch_sql_identifiers` 将 SQL 字面量内所有 `-`→`_` 改为仅在**连接标识符字符**的连字符处转换（`(?<=\w)-(?!\s)`），保留空白包围的算术减号——修复 `UPDATE OVT_MONTHLY SET OVT_HOURS = OVT_HOURS - $1` 被误改成 `OVT_HOURS _ $1` 导致 `near "_": syntax error`、R02 取消循环只处理 1 条即 ABEND 的问题；转换在逻辑拼接串上按行偏移写回（gixsql 将 SQL 切为多行续行）。 + 2) `runners/gixsql_runner.py` 连接字符串回退为 gixsql 原生可解析的 `sqlite://localhost/kin`（`.db` 含点导致解析失败→连接空库→`no such table: OVT_APPLICATIONS`）。 + 3) `runners/gixsql_runner.py` `_build_env` 固定注入 `GIXSQL_AUTOCOMMIT=ON`（libgixsql 仅识别字面量 ON/OFF；此前试 1/TRUE 均不被识别回退 OFF）。 + 4) `orchestrator_db.py` `step3_run_cobol` 在 COBOL 运行后把 `db_path`（gixsql 实际落盘文件）回拷到 `cwd/kin`、`cwd/kin.db`、`cwd/data/kin.db`，修复 DB 比对读取运行前种子快照而非真实输出。 + **效果**：COBOL 提交持久化生效，run_normal 场景 OVT_APPLICATIONS 74/74、OVT_MONTHLY 76/76 全部一致（此前 COBOL 库全 STATUS=None 不提交）。run_collision/run_abnormal 行数 COBOL=Java 但仍有内容差异，属迁移语义问题待后续分析。 + +### 2026-09-08 23:15:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** ZAN06UPD collision/abnormal 场景差异修复：修改 `Zan06UpdMain.java` 的 `dbCleanProc()` 方法第314行，将孤儿取消时的 `abend()` 改为 `r02Eof = true`（对齐 COBOL 的 `MOVE 'Y' TO WRK-R02EOF` 行为）。COBOL 和 Java 在遇到孤儿取消时都会异常终止，但 COBOL 侧 gixsql AUTOCOMMIT=ON 保证每条语句立即提交，而 Java 侧 `setAutoCommit(false)` 导致未提交事务在 `System.exit(999)` 时被回滚。修改后 Java 正常退出循环并提交所有变更。 + **效果**：三个场景全部通过——normal 74/74+76/76 一致，collision 85/85+87/87 一致（此前12+13不一致），abnormal 74/74+76/76 一致（此前6+6不一致）。Java rc 均为 0。总计一致 475 行，不一致 0 行。 + +### 2026-09-09 20:45:00 - 质量评审 +- **范式步骤：** 质量评审 +- **修改摘要：** KIN05MAT 差异分析：35/57 条记录不一致，根因为 COBOL 程序 Bug。 + - **COBOL Bug**：`WRK-BEST-*` 仅在 `1000ITTSOR`（初始化）和 `2020MATCHSOR`（匹配处理）中重置，`2100R01ONLYSOR`（R01单独输出）未重置，导致前次匹配的休暇信息残留到下一条 R01 记录。 + - **Java 正确实现**：按设计书（2-1/3-5）在 `majPara()` 先头调用 `resetBest()`，每条 R01 处理前重置 `WRK-BEST-*` 为 `'99'/0`。 + - **修复方案**：在 COBOL `2000MAJSOR` 的 R01 单独分支（WHEN WRK-R02-EOF 和 WHEN WRK-R01KEY < WRK-R02KEY）中添加重置逻辑。 +- **涉及文件：** `KIN05MAT.cbl`（已修复） +- **使用模型：** AI辅助工具 + +### 2026-09-09 21:10:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** KIN05MAT COBOL Bug 修复完成。在 `2000MAJSOR` 的 R01 单独分支（WHEN WRK-R02-EOF 和 WHEN WRK-R01KEY < WRK-R02KEY）中添加 `WRK-BEST-*` 重置逻辑，确保 R01 单独输出时 LEAVE-TYPE='99'。 + **效果**：修复后 57/57 条记录全部一致，0 不一致。分支覆盖率 36/39（92.3%）。 +- **涉及文件：** `cobol-tna-system/src/KIN05MAT.cbl`（第255-270行添加重置逻辑） +- **使用模型：** AI辅助工具 +- **涉及文件：** `cobol-tna-system/JavaSrc/src/Zan06UpdMain.java`（第314行 `abend()` → `r02Eof = true`）, `output/ZAN06UPD/javasrc/Zan06UpdMain.java`（副本） +- **使用模型：** deepseek/deepseek-v4-flash +- **涉及文件：** `runners/gixsql_runner.py`, `orchestrator_db.py` +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-08 23:00:00 - 交付归档 +- **范式步骤：** 交付归档 +- **修改摘要：** 生成 ZAN06UPD collision/abnormal 场景差异分析报告（`ZAN06UPD_collision_abnormal_差异分析报告.md`）。报告详细记录了：(1) 问题现象——collision 12行/13行差异、abnormal 6行/6行差异；(2) 根本原因——Java `dbCleanProc()` 孤儿取消时调用 `abend()` → `System.exit(999)` 杀死 JVM，而 JDBC 连接设置 `setAutoCommit(false)` 导致未提交事务被 SQLite 回滚（COBOL 侧 gixsql AUTOCOMMIT=ON 保证每条语句立即提交，故不受影响）；(3) 提交边界精确计算验证——collision 未提交 R02[25]-R02[37]=13条、abnormal 未提交 R02[31]-R02[36]=6条，与 diff 差异行数完全吻合；(4) 三个修复方案及推荐方案 A（将 `abend()` 改为 `r02Eof = true` 对齐 COBOL 行为）。 +- **涉及文件：** `output/ZAN06UPD/ZAN06UPD_collision_abnormal_差异分析报告.md`（新建） +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-07 22:10:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** KIN01INP 测试任务（cobol_testgen --gcov）修复三处问题后重跑通过： + 1) runner.py Java 桥入力/出力文件名不再硬编码 `.txt`，改按主类源码 `data/` 常量（`_scan_java_data_files`/`_java_file_name`）准备/收集/比对（修复 KIN01INP Java 读取 data/KIN01R01.csv 的 FileNotFound，Java main rc 由 1→0，比对由全缺→W01 全一致、W02 仅尾部填充差异）； + 2) __init__.py/coverage.py 覆盖率总览路径修正：generate_coverage_index 改传 outdir，index.html 落位 output/coverage/index.html，detail_relpath 加 `../`，明细页返回链接改 `../../coverage/index.html`（消除 coverage/coverage 嵌套）； + 3) Java 被测程序健壮性修复：Sub04ChkSub.chkTimePara 增加 numValSlice 空白 guard（对齐 COBOL NUMVAL 空格→0）；Kin01InpMain 日期/时刻/APPL-ID 数值转换改用 ConvUtil.parseIntSafe(…,0)（对齐 GnuCOBOL 数值 MOVE 空白/非数字→0），javac 重新编译 target/classes。 +- **涉及文件：** `cobol_testgen/runner.py`, `cobol_testgen/__init__.py`, `cobol-tna-system/JavaSrc/src/Sub04ChkSub.java`, `cobol-tna-system/JavaSrc/src/Kin01InpMain.java`（含重新编译的 `.class`） +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-08 22:30:00 - 质量评审 +- **范式步骤：** 质量评审 +- **修改摘要：** ZAN06UPD collision/abnormal 场景差异根因分析完成。通过逐行对比 COBOL 与 Java 的 R01/R02 输入数据、DB 输出、提交边界计算，确认差异根因为 **Java 迁移语义差异**：COBOL 2200DBCLEANSOR 对孤儿取消（空 APPL_ID / ZZZZZZZZ）设置 EOF 标志优雅退出循环并提交所有变更（rc=0）；Java Zan06UpdMain.dbCleanProc 对同样情况调用 `abend()` → `System.exit(999)` 杀死 JVM，未提交事务被 SQLite 回滚。未提交记录数与 diff 差异行数精确匹配：collision 未提交 R02[25]-R02[37] = 12 条 A0000026-A0000037（STATUS 差异），abnormal 未提交 R02[31]-R02[36] = 6 条 A0000032-A0000037。OVT_MONTHLY 差异同理（monthlySub 变更随未提交事务回滚）。建议修复方向：将 `dbCleanProc` 孤儿取消时的 `abend()` 改为 `r02Eof = true`（对齐 COBOL 行为），需与迁移团队确认。 +- **涉及文件：** 无文件修改（纯分析） +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-06 00:30:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** 修复orchestrator_db.py中Java场景使用COBOL修改后DB的问题：在COBOL运行前保存初始DB快照，Java使用初始快照而非COBOL后的DB。同时分析ZAN06UPD DB差异根因：gixsql游标式UPDATE只影响当前行，JDBC集合式UPDATE影响所有匹配行，导致STATUS差异。 +- **涉及文件：** `orchestrator_db.py` +- **使用模型：** opencode/mimo-v2.5-free + +### 2026-09-05 21:30:00 - 交付归档 +- **范式步骤：** 交付归档 +- **修改摘要：** 生成 ZAN05CAL Java 修复详细报告（ZAN05CAL_Java修复报告.md），包含根因分析、字节级对比、修复方案、测试结果、影响评估。 +- **涉及文件：** `output/ZAN05CAL/ZAN05CAL_Java修复报告.md` +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-05 21:20:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** 按 ZAN05CAL 报告根因修复 Java 并重跑（用户指示，同 ZAN04MAT 手法）：新建 output/ZAN05CAL/javasrc（复制 JavaSrc/src 全量），仅改 FileIo.java 的 openReader/openWriter 编码 UTF-8 → ISO-8859-1（字节忠实，修复含多字节 APPL-ID 时按字符切定长的错位及其对分组合并/累计的影响）；javac 编译至 javasrc/target/classes；经 COBOL_JAVA_SRC_DIR 指向 javasrc 重跑 ZAN05CAL：Java[main] 比对由修复前 9一致/2不一致 → 11一致/0不一致，报告判定通过（ZAN05W01 全一致）。 +- **涉及文件：** `output/ZAN05CAL/javasrc/src/FileIo.java`（新建副本） +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-05 21:00:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** 按 ZAN04MAT 报告根因修复 Java 并重跑（用户指示）：新建 output/ZAN04MAT/javasrc（复制 JavaSrc/src 全量），仅将 FileIo.java 的 openReader/openWriter 编码由 UTF-8 改为 ISO-8859-1（字节忠实：1 字节=1 字符，使定长字段按字符切分=按字节切分，修复含多字节 APPL-ID 时的字段错位）；javac 编译至 javasrc/target/classes；runner.py 增加通用环境变量 COBOL_JAVA_SRC_DIR 指向修改后 Java 工程（非单本）。重跑 ZAN04MAT：Java[main] 比对由修复前 6一致/3不一致 → 9一致/0不一致，报告判定通过（W01/W02/W03 全一致）。 +- **涉及文件：** `output/ZAN04MAT/javasrc/src/FileIo.java`（新建副本）, `cobol_testgen/runner.py` +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-05 20:30:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** DB 程序（ZAN06UPD 试点）Java 运行+比对适配（用户批准后实施）。(1) tools/registry.py：默认注册误引用不存在的 comparator.FieldComparator → 改为 compare_field，修复注册表为空导致 gixsql_runner 未注册 KeyError。(2) orchestrator_db.py：新增 DB-Java 通用适配（run_java 开关、_java_cfg/_java_prepare_scenario/_java_run_scenario/_collect_java_outputs/_java_compare_scenario/_write_db_java_report），每场景在 Step3 后从“场景初始 DB 快照 + 同入力”文件式运行 Java（target/classes+target/*.jar+lib/*.jar、显式主类、无需 mvn），比对 COBOL 结果库(run_<场景>/kin)与 Java 结果库逐表（排除 UPDATED_AT 等时间戳列、数值 round 6 位、字符串去空白归一），生成 _测试报告.md。(3) cobol_testgen/__init__.py DB 分支 orch.run_java=True。实测 ZAN06UPD：Java 三场景均运行（normal rc=0；collision/abnormal rc=999=Java 在注入错误场景按设计中止，作为场景结果记录不阻断）；比对显示真实迁移差异（如同键 D0000002 COBOL STATUS=0 vs Java=9、Java 月次聚合 0/0 与 -1/-40.3 vs COBOL 1/40.2），覆盖率 62.2% 不受影响。 +- **涉及文件：** `tools/registry.py`, `orchestrator_db.py`, `cobol_testgen/__init__.py` +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-05 19:45:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** 修复 ZAN03CHK 暴露的两个 V3 缺陷（cobol_testgen/__init__.py，均通用化处理）。(1) `_coordinate_tertiary_fd` 单明细匹配循环对 dpairs 缺键的明细 FD 无守卫 → 3+ 输入文件且某明细（如休日主表 R03）与主 FD 无键对时 KeyError 'R03'；加 `if d not in dpairs: continue`。(2) skip 数据集只为主 FD 落空文件，多输入程序（ZAN03 的 R02/R03）运行时缺文件，COBOL 靠工作目录残留、Java 干净目录 OPEN 失败 rc=1；改为遍历全部 INPUT/I-O FD 各落 0 字节空文件（skip=全输入 FD 无记录，与 JSON 一致）。实测：ZAN03CHK Java[main] 一致13/0、Java[run_skip] rc=0 一致0/0，执行验证 2/2；skip 组（空主+空副）不再依赖残留。 +- **涉及文件：** `cobol_testgen/__init__.py` +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-05 19:05:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** 共通化 Java 运行机制（消除对单 jar manifest 的硬编码依赖）：runner.py 的 _discover_java 改为返回 main_class + 通用 classpath（target/classes + target/*.jar + lib/*.jar，平台分隔符拼装，_build_java_classpath）；_run_java 由 `java -jar ` 改为显式 `java -cp `。修复多程序共享单一 Maven 工程时 jar manifest 固定指向某程序导致错跑的问题（ZAN02CHK 曾误启 Zan01ChkMain，rc=1、0/13）。实测：ZAN02CHK Java[main] rc=0 一致12/不一致1（唯一差异为空 EMP-ID 边界记录：COBOL 输出8空格 vs Java padLeftZero 补零 00000000，属移植差异非 harness）；ZAN01CHK 回归 Java[main] 一致18/0 不变。 +- **涉及文件：** `cobol_testgen/runner.py` +- **使用模型：** deepseek/deepseek-v4-flash + +### 2026-09-05 18:35:00 - AI编码实现 +- **范式步骤：** AI编码实现 +- **修改摘要：** 扩展 cobol_testgen/runner.py 支持 COBOL vs Java 出力比对（V3 原生 main.py 管线对真实多FD程序不适用，采用方案A收敛改动）：新增 Java 工程发现（JavaSrc src/*Main.java 与 program_id 匹配 + target/*.jar）与 java 路径探测；run_all 内每场景（main/run_skip）追加 Java 文件式运行（入力 80B 记录→data/.txt CSV 行→java -jar）+ 出力收集至 output//java//output + 逐 FD（W01/W02/W03）固定长记录比对；输出 _测试报告.md（多组出力分别比较）。ZAN01CHK 端到端实测：Java[main] 一致18/不一致0、Java[run_skip] 一致0/不一致0，测试报告判定通过。无 Java 对应程序的工程自动跳过（JIN01KNS→None 验证）。 +- **涉及文件：** `cobol_testgen/runner.py` +- **使用模型：** deepseek/deepseek-v4-flash + ### 2026-08-31 23:58:00 - 交付归档 - **范式步骤：** 交付归档 - **修改摘要：** 统一评审问题4（声称与实现不一致）残留文档口径：ANCHORED_SUMMARY.md 标注为 2026-06 针对 9 个 KIN 程序的专项实测快照（100% 不代表平台整体，整体口径为 75%）；README 成果摘要规模数字按实测修正（cobol_testgen 18 模块 ~13,700 行、black-box-data-create 22 模块 ~2,300 行、测试 96 文件/885 用例、基准 40 目录/83 源程序、设计文档 10 个）；README 测试通过率改为实测（820 通过/0 失败）；SETUP.md 移除硬编码 "43 程序 100%"（改为动态程序数、整体目标 75%）；s26_regression_check.py 硬编码 "43/43 100%" 消息改为动态输出；test-report.md 覆盖率表标注为整体口径 diff --git a/cobol_testgen/__init__.py b/cobol_testgen/__init__.py index e524e26..a240845 100644 --- a/cobol_testgen/__init__.py +++ b/cobol_testgen/__init__.py @@ -1101,7 +1101,10 @@ def _coordinate_tertiary_fd(records, fd_prefixes, data_fields, term_types=None): _apply(rec, master, pairs, 0) _apply(rec, d, pairs, 0) # rec1..n: 单明细匹配（主 + 该明细 = K(k+1)，其余明细用大偏移避免误配） + # 无键配对的明细 FD（如仅按日期参照的休日主表）不参与键协同，跳过即可。 for k, d in enumerate(details): + if d not in dpairs: + continue if k + 1 >= len(p): continue rec = records[p[k + 1]] @@ -1496,6 +1499,7 @@ def main(): copybook_dirs=[str(d) for d in _cpy_dirs], skip_jvm=True, ) + orch.run_java = True vr = orch.run_all(generate_coverage=False) # Copy output files to outdir (skip if src == dst to avoid self-copy) @@ -1756,7 +1760,7 @@ def main(): else: db_input = None - outpath = prog_outdir / 'main' / 'json' / (filepath.stem + '.json') + outpath = prog_outdir / 'cobol' / 'main' / 'json' / (filepath.stem + '.json') output_json(records, outpath, roles, fd_fields=fd_fields, field_to_fd=field_to_fd, open_dir=open_dir, @@ -1766,14 +1770,14 @@ def main(): select_info = parse_file_control(preprocessed) - output_input_files(records, prog_outdir / 'main' / 'input', filepath.stem, roles, + output_input_files(records, prog_outdir / 'cobol' / 'main' / 'input', filepath.stem, roles, fd_fields, field_to_fd, open_dir, term_types=term_types, data_fields=fields_dict, select_info=select_info) # P12: 子程序输入供给（CALL 目标子程序的 INPUT FD 输入文件） _provision_sub_input_files( - source, str(filepath.parent), prog_outdir / 'main' / 'input' + source, str(filepath.parent), prog_outdir / 'cobol' / 'main' / 'input' ) # ── Skip 数据集（主 FD 空文件触发 PERFORM UNTIL 条件即时满足）── @@ -1793,13 +1797,13 @@ def main(): if eof_fd_dir in ('INPUT', 'I-O') and r in ('input', 'inout'): del rec[fname] # 写 Skip JSON - skip_outpath = prog_outdir / 'skip' / 'json' / (filepath.stem + '.json') + skip_outpath = prog_outdir / 'cobol' / 'run_skip' / 'json' / (filepath.stem + '.json') output_json(skip_records, skip_outpath, roles, fd_fields=fd_fields, field_to_fd=field_to_fd, open_dir=open_dir, term_types=skip_term_types, data_fields=fields_dict) # 写 Skip 输入文件（主 FD 因字段已剥离而不输出） - skip_input_dir = prog_outdir / 'skip' / 'input' + skip_input_dir = prog_outdir / 'cobol' / 'run_skip' / 'input' output_input_files(skip_records, skip_input_dir, filepath.stem + '_skip', roles, fd_fields, field_to_fd, open_dir, @@ -1810,13 +1814,30 @@ def main(): eof_input_path.parent.mkdir(parents=True, exist_ok=True) with open(eof_input_path, 'w', encoding='utf-8') as f: json.dump([], f) - # 空二进制文件（COBOL INPUT 模式需要物理文件存在） + # 空二进制文件（COBOL INPUT 模式需要物理文件存在）。 + # 通用化：skip 数据集 = 全部输入 FD 均无记录，故为每个输入 FD 都落空文件， + # 而非仅主 FD（多输入程序若只建主 FD 空文件，其余 FD 运行时缺失， + # 只能依赖工作目录残留，Java/COBOL 干净场景下会 OPEN 失败）。 eof_assign = select_info.get(eof_fd, {}).get('assign', '') - if eof_assign: - bin_path = skip_input_dir / eof_assign + skip_input_assigns = [] + seen_assign = set() + for _fd_name in fd_fields: + _dir = (open_dir or {}).get(_fd_name, '') + if _dir not in ('INPUT', 'I-O'): + continue + _assign = select_info.get(_fd_name, {}).get('assign', '') + if not _assign or _assign in seen_assign: + continue + seen_assign.add(_assign) + skip_input_assigns.append(_assign) + if eof_assign and eof_assign not in seen_assign: + skip_input_assigns.append(eof_assign) + for _assign in skip_input_assigns: + bin_path = skip_input_dir / _assign bin_path.parent.mkdir(parents=True, exist_ok=True) - bin_path.write_bytes(b'') - logger.info(f" Skip 数据集: {skip_outpath}（空 {eof_fd}）") + if not bin_path.exists(): + bin_path.write_bytes(b'') + logger.info(f" Skip 数据集: {skip_outpath}（全部输入 FD 空文件: {', '.join(skip_input_assigns)}）") gcov_data = None if gcov_mode and proc_div and _HAVE_GCOV and _HAVE_RUNNER: @@ -1909,13 +1930,20 @@ def main(): if dp3_sample: logger.info(f"DEBUG DP#3 other constraints: {sorted(dp3_sample)[:5]}") cov_result = run_coverage(branch_tree, branch_paths_with_assigns, fields_dict, - orig_source, cov_prefix, index_relpath='index.html', + orig_source, cov_prefix, + index_relpath='../../coverage/index.html', gcov_data=gcov_data) programs.append(cov_result) programs[-1]['detail_relpath'] = f'{filepath.stem}/coverage/{filepath.stem}_coverage.html' if programs: - generate_coverage_index(programs, outdir / 'coverage') + # 总览 index.html 位于 /coverage/，detail_relpath（相对 ） + # 需加一级 '../' 前缀才能从 index 定位到各程序明细页。 + for _p in programs: + _rel = _p.get('detail_relpath') or '' + if _rel and not _rel.startswith(('http://', 'https://', '/', '../')): + _p['detail_relpath'] = '../' + _rel + generate_coverage_index(programs, outdir) logger.info(f"\n覆盖率总览：{outdir / 'coverage' / 'index.html'}") diff --git a/cobol_testgen/runner.py b/cobol_testgen/runner.py index a434cbc..6714c2b 100644 --- a/cobol_testgen/runner.py +++ b/cobol_testgen/runner.py @@ -376,6 +376,421 @@ def run_group(group: GroupInfo, exe_path: str, temp_dir: str, ) +# ── Java 文件式运行 + COBOL vs Java 出力比对（V3 扩展）── +# 依赖 JavaSrc 工程：target/*.jar 已构建、src/*Main.java 主类可发现。 +# 仅在满足发现条件时启用；其余程序不受影响。 + + +def _find_java_exe() -> str: + import shutil as _sh + j = _sh.which('java') + if j: + return j + for cand in ( + 'C:/Program Files/Microsoft/jdk-11.0.32.101-hotspot/bin/java.exe', + 'C:/Program Files/Java/jdk-11/bin/java.exe', + 'C:/Program Files/Java/jdk-17/bin/java.exe', + 'C:/Program Files/Eclipse Adoptium/jdk-11.0.21.9-hotspot/bin/java.exe', + ): + if os.path.exists(cand): + return cand + jh = os.environ.get('JAVA_HOME') + if jh and os.path.exists(os.path.join(jh, 'bin', 'java.exe')): + return os.path.join(jh, 'bin', 'java.exe') + return 'java' + + +def _build_java_classpath(java_src_dir: str) -> str: + """通用化 classpath 拼装：target/classes + target/*.jar + lib/*.jar。 + + 不依赖具体 artifact 名/manifest，任何含目标主类的编译产物都可通过 + 同一 classpath 命中（多程序共享一个 Maven 工程时也适用）。 + """ + import os as _os + sep = ';' if _os.name == 'nt' else ':' + parts = [] + root = Path(java_src_dir) + + classes = root / 'target' / 'classes' + if classes.is_dir(): + parts.append(str(classes)) + + tgt = root / 'target' + if tgt.is_dir(): + parts.extend(str(p) for p in sorted(tgt.glob('*.jar'))) + + lib = root / 'lib' + if lib.is_dir(): + parts.extend(str(p) for p in sorted(lib.glob('*.jar'))) + + return sep.join(parts) + + +def _scan_java_data_files(main_src: Path) -> dict: + """扫描 Java 主类源码中引用的 `data/<文件名>` 常量。 + + 返回 {文件基名大写: 完整文件名}，例如 {'KIN01R01': 'KIN01R01.csv', + 'KIN01W01': 'KIN01W01.txt'}。Java 桥据此按程序实际文件名准备/ + 收集/比对入力与出力，而非一律假定 `.txt`（CSV 输入程序 + 如 KIN01INP 的入力文件为 `data/KIN01R01.csv`）。 + """ + file_map = {} + if not main_src.is_file(): + return file_map + try: + text = main_src.read_text(encoding='utf-8') + except Exception: # noqa: BLE001 + return file_map + for m in re.finditer(r'["\'](?:\./)?data/([A-Za-z0-9_-]+)\.([A-Za-z0-9]+)["\']', text): + base = m.group(1).upper() + if base and base not in file_map: + file_map[base] = f"{m.group(1)}.{m.group(2)}" + return file_map + + +def _java_file_name(java_cfg: dict, assign: str) -> str: + """Java 侧实际使用的文件名：优先主类源码 data/ 常量，否则回退 `.txt`。""" + fm = (java_cfg or {}).get('file_map') or {} + return fm.get(assign.upper()) or f'{assign}.txt' + + +def _discover_java(java_src_dir: str, program_name: str): + """在 Java 工程中发现与 COBOL 程序对应的主类与可运行 classpath。 + + 规则（对所有程序通用，非针对单本）： + - 主类：src 下 `XxxMain.java`，去 `Main` 后缀后大写 == 程序ID； + - classpath：target/classes + target/*.jar + lib/*.jar（见 _build_java_classpath）； + - file_map：主类源码中 data/ 常量 → 实际入力/出力文件名； + - 找不到主类或 classpath 为空时返回 None（调用方自动跳过）。 + """ + root = Path(java_src_dir) + src = root / 'src' + if not src.is_dir(): + return None + prog = str(program_name).upper() + main_cls = None + for f in src.glob('*Main.java'): + stem = f.stem + core = stem[:-4] if stem.endswith('Main') else stem + if core.upper() == prog: + main_cls = stem + break + if not main_cls: + return None + classpath = _build_java_classpath(java_src_dir) + if not classpath: + return None + file_map = _scan_java_data_files(src / f'{main_cls}.java') + return {'main_class': main_cls, 'classpath': classpath, 'root': str(root), + 'file_map': file_map} + + +def _fd_record_len(fd_field_dicts: dict, fd_name: str) -> int: + """由 FD 叶子字段长度推算固定长记录字节数（unknown 组项跳过）。""" + total = 0 + for f in fd_field_dicts.get(fd_name, []) or []: + pi = f.get('pic_info') or {} + if pi.get('type') == 'unknown': + continue + total += pi.get('length') or (pi.get('digits', 0) + pi.get('decimal', 0)) or 0 + return total + + +def _slice_fixed_records(data: bytes, rlen: int) -> list[bytes]: + if rlen <= 0: + return [] + out = [] + i = 0 + while i + rlen <= len(data): + out.append(data[i:i + rlen]) + i += rlen + return out + + +def _read_text_lines(path) -> list[str]: + """读取 Java 侧每行一条记录的输出文件。""" + p = Path(path) + if not p.exists(): + return [] + text = p.read_text(encoding='utf-8') + if text == '': + return [] + lines = text.split('\n') + if lines and lines[-1] == '': + lines.pop() + return [ln.rstrip('\r') for ln in lines] + + +def _output_fd_items(fd_field_dicts: dict, open_dir: dict, + select_info: dict) -> list[tuple[str, str]]: + """返回 [(fd_name, assign)]，仅输出系 FD。""" + items = [] + seen = set() + for fd_name in fd_field_dicts: + direction = (open_dir or {}).get(fd_name, '') + if direction not in ('OUTPUT', 'I-O'): + continue + sel = select_info.get(fd_name, {}) + assign = sel.get('assign', fd_name) if isinstance(sel, dict) else fd_name + if assign not in seen: + seen.add(assign) + items.append((fd_name, assign)) + return items + + +def _input_fd_assign_len(fd_field_dicts: dict, open_dir: dict, + select_info: dict) -> list[tuple[str, int]]: + """返回 [(assign, record_len)]，仅输入系 FD（含 I-O）。""" + out = [] + seen = set() + for fd_name in fd_field_dicts: + direction = (open_dir or {}).get(fd_name, '') + if direction not in ('INPUT', 'I-O'): + continue + sel = select_info.get(fd_name, {}) + assign = sel.get('assign', fd_name) if isinstance(sel, dict) else fd_name + if assign in seen: + continue + seen.add(assign) + rlen = _fd_record_len(fd_field_dicts, fd_name) + if rlen > 0: + out.append((assign, rlen)) + return out + + +def _prepare_java_inputs(java_work: Path, src_in_dir: Path, + input_items: list[tuple[str, int]], + java_cfg: dict | None = None) -> None: + data_dir = java_work / 'data' + data_dir.mkdir(parents=True, exist_ok=True) + for assign, rlen in input_items: + src = Path(src_in_dir) / assign + if not src.exists(): + continue + data = src.read_bytes() + recs = _slice_fixed_records(data, rlen) + lines = [r.decode('utf-8', 'replace') for r in recs] + txt = '\n'.join(lines) + if lines: + txt += '\n' + (data_dir / _java_file_name(java_cfg, assign)).write_text(txt, encoding='utf-8') + + +def _run_java(java_cfg: dict, java_work: Path, log_path: Path) -> int: + java_exe = _find_java_exe() + cmd = [java_exe, '-cp', java_cfg['classpath'], java_cfg['main_class']] + orig = os.getcwd() + try: + os.chdir(str(java_work)) + p = subprocess.run(cmd, capture_output=True, text=True, + encoding='utf-8', errors='replace', timeout=90) + except subprocess.TimeoutExpired: + p = None + finally: + os.chdir(orig) + if log_path: + log_path.parent.mkdir(parents=True, exist_ok=True) + if p is None: + log_path.write_text('COMMAND: %s\nTIMEOUT' % ' '.join(cmd), encoding='utf-8') + else: + log_path.write_text( + f"COMMAND: {' '.join(cmd)}\nRETURNCODE: {p.returncode}\n\n" + f"STDOUT:\n{p.stdout}\n\nSTDERR:\n{p.stderr}", + encoding='utf-8') + return p.returncode if p is not None else -1 + + +def _collect_java_outputs(java_work: Path, java_out_dir: Path, + fd_field_dicts: dict, open_dir: dict, + select_info: dict, + java_cfg: dict | None = None) -> None: + java_out_dir.mkdir(parents=True, exist_ok=True) + for fd_name, assign in _output_fd_items(fd_field_dicts, open_dir, select_info): + src = java_work / 'data' / _java_file_name(java_cfg, assign) + if src.exists(): + try: + shutil.copy2(str(src), str(java_out_dir / _java_file_name(java_cfg, assign))) + except OSError: + pass + + +def _compare_cobol_java(cobol_out_dir: Path, java_out_dir: Path, + fd_field_dicts: dict, open_dir: dict, + select_info: dict, + java_cfg: dict | None = None) -> list[dict]: + """逐 FD 比对 COBOL 出力(固定长字节)与 Java 出力(每行一条)。 + + 返回每个输出 FD 的汇总 dict： + {assign, cobol_count, java_count, matched, mismatched, samples} + samples 为最多 3 条差异示例（含记录号与首个不同位置前后片段）。 + """ + summaries = [] + for fd_name, assign in _output_fd_items(fd_field_dicts, open_dir, select_info): + sel = select_info.get(fd_name, {}) + org = sel.get('organization', 'SEQUENTIAL') if isinstance(sel, dict) else 'SEQUENTIAL' + rlen = _fd_record_len(fd_field_dicts, fd_name) + + cobol_path = Path(cobol_out_dir) / assign + if cobol_path.exists(): + data = cobol_path.read_bytes() + if str(org).upper() == 'LINE SEQUENTIAL': + cobol_lines = [ln.rstrip('\n').rstrip('\r') for ln in + data.decode('utf-8', 'replace').split('\n')] + if cobol_lines and cobol_lines[-1] == '': + cobol_lines.pop() + else: + cobol_lines = [r.decode('utf-8', 'replace') + for r in _slice_fixed_records(data, rlen)] + else: + cobol_lines = [] + + java_lines = _read_text_lines( + Path(java_out_dir) / _java_file_name(java_cfg, assign)) + + matched = 0 + mismatched = 0 + samples = [] + n = max(len(cobol_lines), len(java_lines)) + for i in range(n): + c = cobol_lines[i] if i < len(cobol_lines) else '' + j = java_lines[i] if i < len(java_lines) else '' + if c == j: + matched += 1 + else: + mismatched += 1 + if len(samples) < 3: + pos = next((k for k in range(min(len(c), len(j))) + if c[k] != j[k]), min(len(c), len(j))) + samples.append({ + 'index': i, + 'pos': pos, + 'cobol': c[:60], + 'java': j[:60], + }) + summaries.append({ + 'assign': assign, + 'cobol_count': len(cobol_lines), + 'java_count': len(java_lines), + 'matched': matched, + 'mismatched': mismatched, + 'samples': samples, + }) + return summaries + + +def _java_scene_label(dst_out_dir: Path) -> str: + """由 cobol 出力目录推导 java 场景目录名：cobol/main/output -> main。""" + cobol_scene = Path(dst_out_dir).parent # .../cobol/main + return cobol_scene.name + + +def _run_java_scene(program_name: str, scene_id: str, outdir: str, temp_dir: str, + src_in_dir: Path, dst_out_dir: Path, + fd_field_dicts: dict, open_dir: dict, select_info: dict, + java_cfg: dict, log_dir: str | None = None) -> dict: + """单场景：准备 Java 入力 -> 运行 jar -> 收集出力 -> 与 COBOL 出力比对。""" + label = _java_scene_label(dst_out_dir) + java_out_dir = Path(outdir) / 'java' / label / 'output' + java_work = Path(temp_dir) / f'java_{scene_id}' + + input_items = _input_fd_assign_len(fd_field_dicts, open_dir, select_info) + if not input_items: + logger.warning(f" Java[{label}]: 未发现输入 FD，跳过") + return {'label': label, 'rc': -1, 'summaries': []} + + try: + if java_work.exists(): + shutil.rmtree(str(java_work)) + _prepare_java_inputs(java_work, src_in_dir, input_items, java_cfg) + log_path = Path(log_dir) / f'{program_name}_java_{label}.log' if log_dir else None + rc = _run_java(java_cfg, java_work, log_path) + _collect_java_outputs(java_work, java_out_dir, fd_field_dicts, open_dir, select_info, + java_cfg) + summaries = _compare_cobol_java(dst_out_dir, java_out_dir, + fd_field_dicts, open_dir, select_info, + java_cfg) + except Exception as e: # noqa: BLE001 + logger.warning(f" Java[{label}] 运行/比对失败: {e}") + return {'label': label, 'rc': -1, 'summaries': [], 'error': str(e)} + + total_matched = sum(s['matched'] for s in summaries) + total_mismatched = sum(s['mismatched'] for s in summaries) + logger.info(f" Java[{label}]: rc={rc}, 出力比对 一致={total_matched} 不一致={total_mismatched}") + return {'label': label, 'rc': rc, 'summaries': summaries} + + +def _write_java_test_report(outdir: str, program_name: str, + results: list, + java_reports: list[dict]) -> None: + """汇总多组（main/skip 等）COBOL vs Java 出力比对结果，生成中文测试报告。""" + lines = [] + lines.append(f"# {program_name} COBOL vs Java 出力比对测试报告") + lines.append('') + lines.append(f"- 程序ID: {program_name}") + lines.append('- 报告类型: COBOL 出力 vs Java 出力 逐 FD 比对（多组出力分别比较）') + lines.append('') + lines.append('## 1. COBOL 执行结果（V3 cobol_testgen runner）') + lines.append('') + lines.append('| 组 | returncode | 判定 |') + lines.append('|----|-----------|------|') + for r in results: + mark = '通过' if r.passed else '差异' + lines.append(f"| {r.name} | {r.returncode} | {mark} |") + lines.append('') + + lines.append('## 2. Java 执行与出力比对（逐组逐 FD）') + lines.append('') + for rep in java_reports: + lines.append(f'### 2.{java_reports.index(rep) + 1} 组 `{rep["label"]}` (Java rc={rep.get("rc", "N/A")})') + lines.append('') + if rep.get('error'): + lines.append(f'运行/比对异常: {rep["error"]}') + lines.append('') + continue + if not rep['summaries']: + lines.append('（无输出 FD 可比对）') + lines.append('') + continue + lines.append('| 出力FD | COBOL记录数 | Java记录数 | 一致 | 不一致 | 判定 |') + lines.append('|--------|------------|-----------|------|--------|------|') + for s in rep['summaries']: + verdict = '一致' if s['mismatched'] == 0 else '不一致' + lines.append( + f"| {s['assign']} | {s['cobol_count']} | {s['java_count']} " + f"| {s['matched']} | {s['mismatched']} | {verdict} |") + diff = [s for s in rep['summaries'] if s['mismatched'] > 0] + if diff: + lines.append('') + lines.append('**差异示例（最多 3 条/FD）**：') + for s in diff: + for smp in s['samples']: + lines.append(f"- {s['assign']}[记录{smp['index']}] 首异位置={smp['pos']}") + lines.append(f" - COBOL: `{smp['cobol']}`") + lines.append(f" - Java : `{smp['java']}`") + lines.append('') + + all_fd = [s for rep in java_reports for s in rep.get('summaries', [])] + total_match = sum(s['matched'] for s in all_fd) + total_mis = sum(s['mismatched'] for s in all_fd) + lines.append('## 3. 总结') + lines.append('') + if total_match + total_mis == 0: + lines.append('- 无可比对的出力记录（或未发现 Java 对应程序）。') + elif total_mis == 0: + lines.append(f'- COBOL 与 Java 全部出力记录一致（一致 {total_match} 条，不一致 0 条）。') + lines.append('- 判定: **通过**') + else: + lines.append(f'- COBOL 与 Java 出力存在差异：一致 {total_match} 条，不一致 {total_mis} 条。') + lines.append('- 判定: **存在差异**，详见上文逐 FD 明细。') + lines.append('') + lines.append('> 覆盖率详情见 `coverage/` 目录 HTML 报告；Java 出力见 `java/<组>/output/`。') + lines.append('') + + report_path = Path(outdir) / f'{program_name}_测试报告.md' + report_path.write_text('\n'.join(lines), encoding='utf-8') + logger.info(f" COBOL vs Java 测试报告: {report_path}") + + # ── 主编排 ── @@ -388,7 +803,8 @@ def run_all(program_name: str, outdir: str, temp_dir: str, path_infos: list | None = None, multi_write_fds: set | None = None, skip_records: list[dict] | None = None, - skip_term_types: list[str] | None = None + skip_term_types: list[str] | None = None, + java_src_dir: str | None = None ) -> tuple[list[GroupResult], dict[int, int] | None]: """完整编排：编译 → 准备目录 → 逐组执行 → 出力保存。 @@ -423,17 +839,30 @@ def run_all(program_name: str, outdir: str, temp_dir: str, # ── 3. 场景定义 ── scenes = [("main", records, term_types, expected, - Path(outdir) / 'main' / 'input', Path(outdir) / 'main' / 'output')] + Path(outdir) / 'cobol' / 'main' / 'input', Path(outdir) / 'cobol' / 'main' / 'output')] if skip_records: skip_expected = [{}] * len(skip_records) skip_term = skip_term_types or ['normal'] * len(skip_records) scenes.append(("skip", skip_records, skip_term, skip_expected, - Path(outdir) / 'skip' / 'input', Path(outdir) / 'skip' / 'output')) + Path(outdir) / 'cobol' / 'run_skip' / 'input', Path(outdir) / 'cobol' / 'run_skip' / 'output')) results = [] gcov_data_sets = [] gcov_root = work_dir / "gcov" + # ── 2.5 Java 工程发现（可选，无对应 Java 程序时整体跳过）── + java_reports = [] + java_cfg = None + if str(os.environ.get('COBOL_JAVA_RUN', '1')).lower() not in ('0', 'off', 'false'): + _jsrc = (java_src_dir + or os.environ.get('COBOL_JAVA_SRC_DIR') + or str(Path(source_dir).parent / 'JavaSrc')) + java_cfg = _discover_java(_jsrc, program_name) + if java_cfg: + logger.info(f" 发现 Java 对应程序: {java_cfg['main_class']}") + else: + logger.info(f" 未发现 Java 对应程序（{_jsrc}），跳过 Java 运行/比对") + for scene_id, scene_recs, scene_terms, scene_expected, src_in_dir, dst_out_dir in scenes: # ── 3a. 入力ファイル配置（主程序 + 被调子程序的输入文件全部复制）── # 仅复制 assign_names 会漏掉子程序输入文件（测试驱动调用读文件自程序时 @@ -530,6 +959,16 @@ def run_all(program_name: str, outdir: str, temp_dir: str, logger.info(f" {scene_id} 完了, output={dst_out_dir}") + # ── 3h. Java 文件式运行 + COBOL vs Java 出力比对（仅在有 Java 对应程序时）── + if java_cfg: + jrep = _run_java_scene( + program_name, scene_id, outdir, str(work_dir), + src_in_dir, dst_out_dir, + fd_field_dicts, open_dir, select_info, + java_cfg, log_dir=log_dir, + ) + java_reports.append(jrep) + # ── 4. 合并 gcov ── merged_gcov = None if gcov_data_sets: @@ -539,6 +978,13 @@ def run_all(program_name: str, outdir: str, temp_dir: str, merged_gcov[line] = max(merged_gcov.get(line, 0), count) logger.info(f" Merged gcov from {len(gcov_data_sets)} runs ({len(merged_gcov)} lines)") + # ── 5. 生成 COBOL vs Java 出力比对测试报告 ── + if java_cfg and java_reports: + try: + _write_java_test_report(outdir, program_name, results, java_reports) + except Exception as e: # noqa: BLE001 + logger.warning(f" 测试报告生成失败: {e}") + return results, merged_gcov diff --git a/comparator/aligner.py b/comparator/aligner.py index 58ad163..446f0c6 100644 --- a/comparator/aligner.py +++ b/comparator/aligner.py @@ -1,20 +1,43 @@ def align_records(cobol_records: list[dict], java_records: list[dict], key_field: str = "CUST-ID") -> list[tuple]: + """对齐COBOL和Java记录 + + Args: + cobol_records: COBOL输出记录列表 + java_records: Java输出记录列表 + key_field: 用于对齐的关键字段名（默认为"CUST-ID"） + + Returns: + 对齐结果列表，每个元素为 (cobol_record, java_record, status) + status: "MATCHED", "MISSING_IN_SPARK", "EXTRA_IN_SPARK" + """ if not cobol_records and not java_records: return [] - + + # 智能关键字段推断：如果默认key_field不存在，尝试推断 + effective_key = key_field + if cobol_records: + sample_record = cobol_records[0] + if effective_key not in sample_record: + # 尝试常见的关键字段名 + common_keys = ["ID", "CUST-ID", "EMP-ID", "KEY", "CODE", "NO"] + for k in common_keys: + if k in sample_record: + effective_key = k + break + def _by(records, kf): d = {} for r in records: key = str(r.get(kf, "__NONE__")) d.setdefault(key, []).append(r) return d - - c_by = _by(cobol_records, key_field) - j_by = _by(java_records, key_field) + + c_by = _by(cobol_records, effective_key) + j_by = _by(java_records, effective_key) pairs = [] all_keys = set(c_by) | set(j_by) - + for k in sorted(all_keys): c_items = c_by.get(k, []) j_items = j_by.get(k, []) diff --git a/docs/output-directory-structure.md b/docs/output-directory-structure.md new file mode 100644 index 0000000..52b4ba2 --- /dev/null +++ b/docs/output-directory-structure.md @@ -0,0 +1,400 @@ +# V3系统目录结构设计文档 + +## 一、目录结构概述 + +V3系统采用统一的目录结构来组织COBOL和Java的输出结果，确保所有程序的输出文件都位于同一个根目录下，便于管理和比较。 + +## 二、目录结构规范 + +### 2.1 标准目录结构 + +``` +output// +├── cobol/ # COBOL输出（所有组） +│ ├── main/ # 单轮/默认场景 +│ │ ├── input/ # COBOL输入flat文件 +│ │ ├── output/ # COBOL输出flat文件 +│ │ └── json/ # Java验证用JSON +│ ├── run_skip/ # 跳过场景 +│ │ ├── input/ +│ │ ├── output/ +│ │ └── json/ +│ ├── run_/ # 多轮场景 +│ │ ├── input/ +│ │ ├── output/ +│ │ └── json/ +│ └── pre_src/ # 预处理源码 +├── java/ # Java输出（所有组） +│ ├── main/ # 对应cobol/main的Java输出 +│ │ └── output/ +│ ├── run_skip/ # 对应cobol/run_skip的Java输出 +│ │ └── output/ +│ └── run_/ # 对应cobol/run_*的Java输出 +│ └── output/ +├── coverage/ # 覆盖率报告（全局） +├── logs/ # 日志（全局） +│ ├── main.log +│ ├── run_skip.log +│ └── run_.log +├── gcov/ # gcov数据（按场景分离） +│ ├── run_main/ +│ ├── run_skip/ +│ └── run_/ +├── data/ # SQLite数据库（按场景分离） +│ ├── kin.db # 单轮场景 +│ └── kin_.db # 多轮场景 +└── reports/ # 测试报告 + └── / + ├── result.json + ├── report.html + └── machine.json +``` + +### 2.2 目录命名规则 + +| 目录 | 命名规则 | 说明 | +|------|----------|------| +| `cobol/main/` | 固定名称 | 单轮/默认场景 | +| `cobol/run_skip/` | 固定名称 | 跳过场景（原 `skip/`） | +| `cobol/run_/` | `run_` + 场景ID | 多轮场景 | +| `java/<对应cobol目录>/` | 与cobol目录对应 | Java输出 | +| `coverage/` | 固定名称 | 覆盖率报告 | +| `logs/` | 固定名称 | 日志文件 | +| `gcov/run_/` | `run_` + 场景ID | gcov数据 | +| `data/` | 固定名称 | SQLite数据库 | + +### 2.3 文件命名规则 + +| 文件类型 | 命名规则 | 示例 | +|----------|----------|------| +| COBOL输入文件 | `R` | `KIN01R01`, `ZAN01R01` | +| COBOL输出文件 | `W` | `KIN01W01`, `ZAN01W01` | +| JSON中介文件 | `.json` | `KIN01INP.json` | +| 覆盖率报告 | `_coverage.html` | `KIN01INP_coverage.html` | +| 运行日志 | `<场景ID>.log` | `main.log`, `run_normal.log` | +| gcov数据 | 按场景ID分目录 | `gcov/run_main/`, `gcov/run_skip/` | +| SQLite数据库 | `kin_<场景ID>.db` | `kin.db`, `kin_normal.db` | + +## 三、场景映射规则 + +### 3.1 单轮场景 + +**定义**：程序只有一次执行，无多场景配置。 + +**目录结构**： +``` +output// +├── cobol/ +│ └── main/ +│ ├── input/ +│ ├── output/ +│ └── json/ +├── java/ +│ └── main/ +│ └── output/ +├── coverage/ +├── logs/ +│ └── main.log +├── gcov/ +│ └── run_main/ +└── data/ + └── kin.db +``` + +### 3.2 多轮场景 + +**定义**：程序有多次执行，通过YAML配置文件定义场景。 + +**目录结构**： +``` +output// +├── cobol/ +│ ├── run_normal/ +│ │ ├── input/ +│ │ ├── output/ +│ │ └── json/ +│ ├── run_insert_error/ +│ │ ├── input/ +│ │ ├── output/ +│ │ └── json/ +│ └── run_sql_delete_error/ +│ ├── input/ +│ ├── output/ +│ └── json/ +├── java/ +│ ├── run_normal/ +│ │ └── output/ +│ ├── run_insert_error/ +│ │ └── output/ +│ └── run_sql_delete_error/ +│ └── output/ +├── coverage/ +├── logs/ +│ ├── run_normal.log +│ ├── run_insert_error.log +│ └── run_sql_delete_error.log +├── gcov/ +│ ├── run_normal/ +│ ├── run_insert_error/ +│ └── run_sql_delete_error/ +└── data/ + ├── kin_normal.db + ├── kin_insert_error.db + └── kin_sql_delete_error.db +``` + +### 3.3 跳过场景 + +**定义**：程序有跳过主FD输入的场景（旧版 `skip/` 目录）。 + +**目录结构**： +``` +output// +├── cobol/ +│ ├── main/ +│ │ ├── input/ +│ │ ├── output/ +│ │ └── json/ +│ └── run_skip/ +│ ├── input/ +│ ├── output/ +│ └── json/ +├── java/ +│ ├── main/ +│ │ └── output/ +│ └── run_skip/ +│ └── output/ +├── coverage/ +├── logs/ +│ ├── main.log +│ └── run_skip.log +├── gcov/ +│ ├── run_main/ +│ └── run_skip/ +└── data/ + ├── kin.db + └── kin_skip.db +``` + +## 四、路径依赖说明 + +### 4.1 env_overrides 路径 + +COBOL运行时通过环境变量映射文件路径： + +```python +# 输入文件 +env_overrides[fname] = os.path.join("input", fname) + +# 输出文件 +env_overrides[fname] = os.path.join("output", fname) +``` + +**注意**：这些是相对于CWD的路径，CWD为 `cobol/main/` 或 `cobol/run_/`。 + +### 4.2 SQLite数据库路径 + +```python +# 单轮场景 +db_path = data/kin.db + +# 多轮场景 +db_path = data/kin_.db +``` + +### 4.3 gcov数据路径 + +```python +# 单轮场景 +gcov_dir = gcov/run_main/ + +# 多轮场景 +gcov_dir = gcov/run_/ +``` + +## 五、向后兼容性 + +### 5.1 旧版目录迁移 + +| 旧版目录 | 新版目录 | 迁移方式 | +|----------|----------|----------| +| `main/input/` | `cobol/main/input/` | 移动文件 | +| `main/output/` | `cobol/main/output/` | 移动文件 | +| `main/json/` | `cobol/main/json/` | 移动文件 | +| `skip/input/` | `cobol/run_skip/input/` | 重命名+移动 | +| `skip/output/` | `cobol/run_skip/output/` | 重命名+移动 | +| `skip/json/` | `cobol/run_skip/json/` | 重命名+移动 | +| `pre_src/` | `cobol/pre_src/` | 移动文件 | +| `coverage/` | `coverage/` | 保持不变 | +| `logs/` | `logs/` | 保持不变 | +| `gcov/` | `gcov/` | 保持不变 | +| `data/` | `data/` | 保持不变 | + +### 5.2 兼容性处理 + +```python +def _ensure_cobol_dir_structure(runtime_dir): + """确保cobol目录结构存在，兼容旧版""" + cobol_dir = runtime_dir / "cobol" + if not cobol_dir.exists(): + # 检查是否是旧版结构（main/直接在runtime_dir下） + old_main = runtime_dir / "main" + if old_main.exists(): + # 迁移到新结构 + shutil.move(str(old_main), str(cobol_dir / "main")) +``` + +## 六、黑盒测试映射 + +### 6.1 YAML配置示例 + +```yaml +# config/programs/KIN08DBU.yaml +runs: + - id: normal + sysin: + - { dd: KIN08S01, content: "..." } + - id: no_period + sysin: + - { dd: KIN08S01, content: "..." } + - id: insert_error + inject_duplicate_pk: true + sysin: + - { dd: KIN08S01, content: "..." } + - id: sql_delete_error + sysin: + - { dd: KIN08S01, content: "..." } + - id: sql_select_error + sysin: + - { dd: KIN08S01, content: "..." } +``` + +### 6.2 目录映射 + +| 场景ID | COBOL目录 | Java目录 | 日志文件 | gcov目录 | 数据库文件 | +|--------|-----------|----------|----------|----------|------------| +| normal | `cobol/run_normal/` | `java/run_normal/output/` | `logs/run_normal.log` | `gcov/run_normal/` | `data/kin_normal.db` | +| no_period | `cobol/run_no_period/` | `java/run_no_period/output/` | `logs/run_no_period.log` | `gcov/run_no_period/` | `data/kin_no_period.db` | +| insert_error | `cobol/run_insert_error/` | `java/run_insert_error/output/` | `logs/run_insert_error.log` | `gcov/run_insert_error/` | `data/kin_insert_error.db` | +| sql_delete_error | `cobol/run_sql_delete_error/` | `java/run_sql_delete_error/output/` | `logs/run_sql_delete_error.log` | `gcov/run_sql_delete_error/` | `data/kin_sql_delete_error.db` | +| sql_select_error | `cobol/run_sql_select_error/` | `java/run_sql_select_error/output/` | `logs/run_sql_select_error.log` | `gcov/run_sql_select_error/` | `data/kin_sql_select_error.db` | + +### 6.3 黑盒测试执行流程 + +1. **编译阶段**：编译一次，全场景共享 `.exe` 和 `.gcno` +2. **场景执行**：对每个场景独立执行 + - 生成场景特定输入数据 + - 初始化场景特定DB + - 运行COBOL程序 + - 收集场景特定gcov数据 +3. **Java执行**：使用最后一个场景的DB结果 +4. **验证比对**：比较COBOL和Java输出 +5. **覆盖率合并**：合并多轮gcov数据，生成覆盖率报告 + +### 6.4 测试报告生成 + +| 报告类型 | 生成方式 | 输出位置 | +|----------|----------|----------| +| 覆盖率HTML报告 | `orchestrator_db.generate_coverage_report()` | `coverage/_coverage.html` | +| 测试结果JSON | `orchestrator.run_all()` | `reports//result.json` | +| 测试报告HTML | `ReportGenerator.generate_html()` | `reports//report.html` | +| 机器可读JSON | `ReportGenerator.generate_machine_json()` | `reports//machine.json` | + +## 七、代码修改清单 + +### 7.1 orchestrator_db.py + +| 行号 | 修改内容 | 说明 | +|------|----------|------| +| 103 | `runtime_dir = v3_root / "output" / program_id / "cobol"` | 添加cobol子目录 | +| 594-596 | `run_label = "main" if not scenario else f"run_{scenario.id}"` | 统一场景目录命名 | +| 626-628 | `env_overrides` 路径改为 `"input"` 和 `"output"` | 移除main/层级 | +| 686 | `log_dir = self.runtime_dir.parent / "logs"` | 日志目录移到根目录 | +| 700 | `gcda_dst_dir = gcov_dir / run_label` | 统一gcov目录命名 | +| 769 | `output_dir = v3_root / "output" / self.program_id / "coverage"` | 覆盖率报告移到根目录 | +| 779 | `gcov_dir = self.runtime_dir.parent / "gcov"` | gcov目录移到根目录 | +| 975 | `java_out = self.runtime_dir.parent / "java" / run_label / "output"` | Java输出目录 | +| 1016 | `self.java_output_path = self.runtime_dir.parent / "java" / run_label / "output"` | Java输出路径 | + +### 7.2 cobol_testgen/__init__.py + +| 行号 | 修改内容 | 说明 | +|------|----------|------| +| 1759 | `outpath = prog_outdir / 'cobol' / 'main' / 'json'` | JSON输出路径 | +| 1769 | `prog_outdir / 'cobol' / 'main' / 'input'` | 输入文件路径 | +| 1776 | `prog_outdir / 'cobol' / 'main' / 'input'` | 子程序输入路径 | +| 1796 | `prog_outdir / 'cobol' / 'run_skip' / 'json'` | Skip JSON路径 | +| 1802 | `prog_outdir / 'cobol' / 'run_skip' / 'input'` | Skip输入路径 | + +### 7.3 cobol_testgen/runner.py + +| 行号 | 修改内容 | 说明 | +|------|----------|------| +| 425-426 | `Path(outdir) / 'cobol' / 'main' / 'input'` | 主场景输入路径 | +| 425-426 | `Path(outdir) / 'cobol' / 'main' / 'output'` | 主场景输出路径 | +| 430-431 | `Path(outdir) / 'cobol' / 'run_skip' / 'input'` | Skip场景输入路径 | +| 430-431 | `Path(outdir) / 'cobol' / 'run_skip' / 'output'` | Skip场景输出路径 | + +### 7.4 runners/gixsql_runner.py + +| 行号 | 修改内容 | 说明 | +|------|----------|------| +| 308 | `debug_dir = ... / "cobol" / "pre_src"` | 预处理源码路径 | + +### 7.5 orchestrator.py + +| 行号 | 修改内容 | 说明 | +|------|----------|------| +| 142 | `co = Path(f"output/{cfg.program}/cobol/main/output/cobol_out.bin")` | COBOL输出路径 | +| 159 | `java_out_dir = Path("output") / cfg.program / "java" / "main" / "output"` | Java输出路径 | +| 200 | `rd = Path(f"output/{vr.program}/reports") / vr.timestamp` | 报告输出路径 | + +## 八、测试验证 + +### 8.1 单元测试 + +运行现有单元测试确保向后兼容： + +```bash +python -m pytest tests/ -v +``` + +### 8.2 集成测试 + +运行黑盒测试验证目录结构： + +```bash +cd test-data +python s15_coverage_verification.py +python s30_db_e2e.py +``` + +### 8.3 手动验证 + +检查目录结构是否正确： + +```bash +# 检查单轮程序 +ls -la output/KIN01INP/ +ls -la output/KIN01INP/cobol/main/ +ls -la output/KIN01INP/java/main/output/ + +# 检查多轮程序 +ls -la output/KIN08DBU/ +ls -la output/KIN08DBU/cobol/run_normal/ +ls -la output/KIN08DBU/java/run_normal/output/ +``` + +## 九、注意事项 + +1. **env_overrides 路径**：修改后需要确保COBOL运行时能找到正确的文件 +2. **SQLite数据库路径**：多轮场景的DB文件需要按场景命名 +3. **gcov数据合并**：多轮场景的gcov数据需要正确合并 +4. **向后兼容**：需要处理旧版目录结构的迁移 +5. **测试覆盖**：修改后需要运行所有测试确保功能正常 + +--- + +**文档版本**：v1.0 +**创建日期**：2026-09-05 +**最后更新**：2026-09-05 diff --git a/orchestrator.py b/orchestrator.py index 1dc2057..070a02d 100644 --- a/orchestrator.py +++ b/orchestrator.py @@ -139,7 +139,7 @@ def run_pipeline(cfg: Config, cpath: str, cbl: str, java: str, map_path: str) -> vr.debug["cobol_build"] = {"ok": build.success, "log": build.log[-300:]} if not build.success: return _done(vr, t0, "BLOCKED", 2) - co = Path("cobol_out.bin") + co = Path(f"output/{cfg.program}/cobol/main/output/cobol_out.bin") if not cob.run(build.artifact_path, str(bundle.cobol_input()), str(co)).success: return _done(vr, t0, "ERROR", 3) @@ -156,7 +156,9 @@ def run_pipeline(cfg: Config, cpath: str, cbl: str, java: str, map_path: str) -> if not jb.success: return _done(vr, t0, "BLOCKED", 2) inp = str(bundle.spark_input_dir() if cfg.runner_mode == "spark" else bundle.native_input()) - jr = runner.run(jb.artifact_path, inp, "java_out") + java_out_dir = Path("output") / cfg.program / "java" / "main" / "output" + java_out_dir.mkdir(parents=True, exist_ok=True) + jr = runner.run(jb.artifact_path, inp, str(java_out_dir)) reader = CobolBinaryReader() cr = reader.read(str(co), tree) @@ -195,7 +197,7 @@ def run_pipeline(cfg: Config, cpath: str, cbl: str, java: str, map_path: str) -> except: pass - rd = Path(f"reports/{vr.program}") / vr.timestamp + rd = Path(f"output/{vr.program}/reports") / vr.timestamp rd.mkdir(parents=True, exist_ok=True) g = ReportGenerator() g.generate_json(vr, rd / "result.json") diff --git a/orchestrator_db.py b/orchestrator_db.py index 701ad4a..bd18aaa 100644 --- a/orchestrator_db.py +++ b/orchestrator_db.py @@ -90,6 +90,10 @@ class GixsqlOrchestrator: self.cobol_src_dir = Path(cobol_src_dir) self.copybook_dirs = copybook_dirs or [] self.skip_jvm = skip_jvm + # DB-Java 文件式运行 + DB 表比对（run_java=True 时在每场景 Step3 后触发） + self.run_java: bool = False + self._java_cfg_cache = None + self._db_java_results: list[dict] = [] v3_root = Path(__file__).parent # cobol-java-v3/ # Build artifacts in temp (ASCII-only, gixpp can't handle Chinese paths) @@ -100,7 +104,8 @@ class GixsqlOrchestrator: self.work_dir = Path(work_dir) # Runtime data under V3 (DB, flat files, CWD) - self.runtime_dir = v3_root / "output" / program_id + # 新目录结构：output//cobol/ 存放COBOL输出 + self.runtime_dir = v3_root / "output" / program_id / "cobol" self.schema: ProgramSchema = load_schema(program_id) @@ -122,6 +127,7 @@ class GixsqlOrchestrator: self.exe_path: Optional[Path] = None self.java_input_path: Optional[Path] = None self._current_db_path: Optional[Path] = None # scenario-specific DB path + self._java_initial_db_snapshots: dict[str, Path] = {} # scenario label -> initial DB snapshot (before COBOL) self._multi_run_gcov_data: dict[int, int] | None = None # merged multi-run gcov data self._sub_gcov_data: dict[str, dict[int, int]] = {} # per-subprogram gcov (kept separate from main) self.java_output_path: Optional[Path] = None @@ -590,11 +596,11 @@ class GixsqlOrchestrator: "exe not found (run step1 first)") # シナリオ毎の出力先 - run_label = f"run_{scenario.id}" if scenario else "" - run_dir = self.runtime_dir / run_label if scenario else self.runtime_dir - input_dir = run_dir / "main" / "input" - output_dir = run_dir / "main" / "output" - gcov_dir = self.runtime_dir / "gcov" + run_label = f"run_{scenario.id}" if scenario else "main" + run_dir = self.runtime_dir / run_label + input_dir = run_dir / "input" + output_dir = run_dir / "output" + gcov_dir = self.runtime_dir.parent / "gcov" input_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True) gcov_dir.mkdir(parents=True, exist_ok=True) @@ -623,9 +629,9 @@ class GixsqlOrchestrator: env_overrides = {} for fname, direction in assign_map.items(): if direction == "INPUT": - env_overrides[fname] = os.path.join("main", "input", fname) + env_overrides[fname] = os.path.join("input", fname) else: - env_overrides[fname] = os.path.join("main", "output", fname) + env_overrides[fname] = os.path.join("output", fname) # シナリオ毎の DB パス db_path = self._current_db_path or self.db_path @@ -645,10 +651,15 @@ class GixsqlOrchestrator: shutil.copy2(str(db_path), str(cwd_db)) # gixsql regex requires sqlite://host/path (single segment, no dots). # Copy to CWD/kin (no extension) for sqlite://localhost/kin. + # 同时复制到 kin.db，确保 gixsql 打开的是有 PRIMARY KEY 的版本 cwd_kin = cwd / "kin" if cwd_kin.exists(): cwd_kin.unlink() shutil.copy2(str(db_path), str(cwd_kin)) + cwd_kin_db = cwd / "kin.db" + if cwd_kin_db.exists(): + cwd_kin_db.unlink() + shutil.copy2(str(db_path), str(cwd_kin_db)) # .gcda は CWD（= run_dir）に書き出されるので、実行後に gcov/run_{id}/ に移動する # 各シナリオ実行前に前回の .gcda を削除（GnuCOBOL は累積書込みを行うため） @@ -682,7 +693,7 @@ class GixsqlOrchestrator: command_args=command_args, ) - log_dir = self.runtime_dir / "logs" + log_dir = self.runtime_dir.parent / "logs" log_dir.mkdir(parents=True, exist_ok=True) log_dir.joinpath(f"{run_label or self.program_id}.log").write_text( result.log, encoding='utf-8') @@ -696,7 +707,7 @@ class GixsqlOrchestrator: gcda_src_dirs.append(exe_dir_for_gcda) if scenario is None: gcda_src_dirs.append(self.runtime_dir) # 従来互換 - gcda_dst_dir = gcov_dir / run_label if scenario else gcov_dir + gcda_dst_dir = gcov_dir / run_label gcda_dst_dir.mkdir(parents=True, exist_ok=True) for sd in gcda_src_dirs: for f in sd.glob("*.gcda"): @@ -715,6 +726,17 @@ class GixsqlOrchestrator: except PermissionError: pass + # gixsql 通过 GIXSQL_DB_PATH 把 COBOL 结果写回 db_path（运行时唯一真实落盘文件）。 + # CWD 下的 kin/kin.db 是运行前从种子库复制的快照；这里运行后回拷， + # 保证 _java_compare_scenario 读取的是 COBOL 的真实输出而不是过期的种子快照。 + if db_path and db_path.exists(): + for _dst in (cwd / 'kin', cwd / 'kin.db', cwd / 'data' / 'kin.db'): + try: + _dst.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(str(db_path), str(_dst)) + except OSError as _e: + logger.warning(f" 运行后 DB 回拷失败 {_dst}: {_e}") + return DbPipelineResult( self.program_id, 3, result.success, data={"returncode": result.returncode, "log": result.log[:500], @@ -765,7 +787,7 @@ class GixsqlOrchestrator: f"exe not found at {self.exe_path} or {fallback}") if output_dir is None: v3_root = Path(__file__).parent - output_dir = v3_root / "reports" / self.program_id / "coverage" + output_dir = v3_root / "output" / self.program_id / "coverage" output_dir = Path(output_dir) # 1. Use pre-merged multi-run gcov data if available (skip gcov re-run) @@ -775,7 +797,7 @@ class GixsqlOrchestrator: # plain integers that collide with the main program's (e.g. # SUB04CHK line 167=0 would overwrite main line 167=25 and # wipe real coverage). Stored per-subprogram for reference. - gcov_dir = self.runtime_dir / "gcov" + gcov_dir = self.runtime_dir.parent / "gcov" self._sub_gcov_data = {} for sub in self.schema.subprograms: sub_merged = _merge_run_dirs_gcov(gcov_dir, sub) @@ -913,6 +935,344 @@ class GixsqlOrchestrator: logger.exception("generate_coverage_report failed") return DbPipelineResult(self.program_id, 0, False, str(e)) + # ── DB-Java 通用适配（文件式 Java 运行 + DB 表比对） ── + # 依赖 JavaSrc 工程（target/classes + target/*.jar + lib/*.jar，显式主类）， + # 无需 mvn。Java 从“场景初始 DB 快照 + 同入力”出发，最终 DB/W01 与 COBOL 结果比对。 + + def _java_cfg(self): + if self._java_cfg_cache is None: + self._java_cfg_cache = None + try: + from cobol_testgen.runner import _discover_java + root = Path(self.cobol_src_dir).parent / 'JavaSrc' + cfg = _discover_java(str(root), self.program_id) + self._java_cfg_cache = cfg + if cfg: + logger.info(f" DB-Java: 发现主类 {cfg['main_class']}") + except Exception as e: # noqa: BLE001 + logger.warning(f" DB-Java 发现失败: {e}") + return self._java_cfg_cache + + def _java_db_rel_path(self) -> Optional[str]: + """扫描 Java 主类源码中 jdbc:sqlite:<相对路径>，返回 DB 相对路径。""" + cfg = self._java_cfg() + if not cfg: + return None + try: + main = Path(cfg['root']) / 'src' / f"{cfg['main_class']}.java" + txt = main.read_text(encoding='utf-8', errors='replace') + m = re.search(r'jdbc:sqlite:([^\s"\'\)]+)', txt) + if m: + return m.group(1).replace('\\', os.sep) + except Exception: # noqa: BLE001 + pass + return None + + def _java_scenario_label(self, scenario: ScenarioDef | None) -> str: + return f"run_{scenario.id}" if scenario else "main" + + def _java_prepare_scenario(self, scenario: ScenarioDef | None) -> Optional[dict]: + """每场景准备 Java 运行目录：初始 DB 快照 + 入力 txt。""" + cfg = self._java_cfg() + if not cfg: + return None + label = self._java_scenario_label(scenario) + java_dir = self.runtime_dir.parent / 'java' / label + data_dir = java_dir / 'data' + if java_dir.exists(): + shutil.rmtree(str(java_dir)) + data_dir.mkdir(parents=True, exist_ok=True) + + # 1) 初始 DB 快照（Step2 初始化后、COBOL 运行前的场景 DB） + # db_rel 为相对 Java CWD(=java_dir) 的路径（如 data/OVERTIME.DB），直接落到 java_dir 下 + db_rel = self._java_db_rel_path() + # 优先使用 COBOL 运行前保存的初始快照，避免 Java 在 COBOL 修改后的 DB 上重复处理 + label = self._java_scenario_label(scenario) + db_src = self._java_initial_db_snapshots.get(label) or self._current_db_path or self.db_path + if db_rel and db_src and Path(db_src).exists(): + dst = java_dir / db_rel + dst.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(str(db_src), str(dst)) + logger.info(f" DB-Java[{label}]: 初始 DB 快照 -> {dst}") + + # 2) 入力 txt（80B 定长 → 行） + gen_input_dir = (self.work_dir / f"run_{scenario.id}" / "main" / "input") if scenario \ + else (self.work_dir / "main" / "input") + written: list[str] = [] + layouts: dict = {} + try: + from cobol_testgen.flatfile import analyze_fd_layout + if self.src_path and self.src_path.exists(): + layouts = analyze_fd_layout( + self.src_path.read_text(encoding='utf-8-sig'), + [str(d) for d in self.copybook_dirs]) + except Exception as e: # noqa: BLE001 + logger.warning(f" DB-Java 布局解析失败: {e}") + for assign, lay in layouts.items(): + direction = str(lay.get('direction', 'INPUT')).upper() + if direction not in ('INPUT', 'I-O'): + continue + recs = lay.get('records') or [] + rlen = recs[0].get('record_length', 0) if recs else 0 + src = gen_input_dir / assign + if not src.exists() or rlen <= 0: + continue + data = src.read_bytes() + lines = [] + i = 0 + while i + rlen <= len(data): + lines.append(data[i:i + rlen].decode('utf-8', 'replace')) + i += rlen + txt = '\n'.join(lines) + ('\n' if lines else '') + (data_dir / f'{assign}.txt').write_text(txt, encoding='utf-8') + written.append(assign) + logger.info(f" DB-Java[{label}]: 入力 {assign}.txt {len(lines)} 行") + + if not written and not db_rel: + return None + return {'label': label, 'java_dir': str(java_dir), 'data_dir': str(data_dir), + 'db_rel': db_rel, 'inputs': written, 'cfg': cfg} + + def _java_run_scenario(self, scenario: ScenarioDef | None, info: dict) -> dict: + """运行 Java 主类（文件式），返回 rc/日志。异常与 rc!=0 不抛出，仅记录。""" + cfg = info['cfg'] + from cobol_testgen.runner import _find_java_exe + java_exe = _find_java_exe() + cmd = [java_exe, '-cp', cfg['classpath'], cfg['main_class']] + # command_line: scenario-level (if set) overrides program-level default + cmd_line = self.schema.command_line + if scenario and scenario.command_line is not None: + cmd_line = scenario.command_line + if cmd_line: + cmd.extend(cmd_line.split()) + log_path = self.runtime_dir.parent / 'logs' / f"{self.program_id}_java_{info['label']}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + orig = os.getcwd() + p = None + try: + os.chdir(info['java_dir']) + p = subprocess.run(cmd, capture_output=True, text=True, + encoding='utf-8', errors='replace', timeout=180) + except subprocess.TimeoutExpired: + log_path.write_text(f"COMMAND: {' '.join(cmd)}\nTIMEOUT\n", encoding='utf-8') + return {'rc': -1, 'log': 'TIMEOUT'} + except Exception as e: # noqa: BLE001 + log_path.write_text(f"COMMAND: {' '.join(cmd)}\nRUN ERROR: {e}\n", encoding='utf-8') + return {'rc': -2, 'log': str(e)} + finally: + os.chdir(orig) + log = f"COMMAND: {' '.join(cmd)}\nRETURNCODE: {p.returncode}\n\nSTDOUT:\n{p.stdout}\n\nSTDERR:\n{p.stderr}" + log_path.write_text(log, encoding='utf-8') + logger.info(f" DB-Java[{info['label']}]: rc={p.returncode}") + return {'rc': p.returncode, 'log': p.stdout + p.stderr} + + def _collect_java_outputs(self, info: dict) -> None: + """收集 Java 出力：txt 出力（排除入力）+ Java 运行后 DB 副本。""" + label = info['label'] + java_out_dir = self.runtime_dir.parent / 'java' / label / 'output' + java_out_dir.mkdir(parents=True, exist_ok=True) + inputs = set(info.get('inputs') or []) + data_dir = Path(info['data_dir']) + for f in sorted(data_dir.glob('*.txt')): + if f.name in inputs: + continue + try: + shutil.copy2(str(f), str(java_out_dir / f.name)) + except OSError: + pass + if info.get('db_rel'): + src = Path(info['java_dir']) / info['db_rel'] + if src.exists(): + try: + shutil.copy2(str(src), str(java_out_dir / f"{self.program_id}_java.db")) + except OSError: + pass + + @staticmethod + def _db_table_snap(db_path) -> dict[str, list[dict]]: + snap: dict[str, list[dict]] = {} + if not db_path or not Path(db_path).exists(): + return snap + try: + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + cur = conn.cursor() + tables = [r[0] for r in cur.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")] + for t in tables: + rows = [] + try: + for row in cur.execute('SELECT * FROM "%s"' % t): + rows.append(dict(row)) + except sqlite3.OperationalError: + continue + snap[t] = rows + conn.close() + except Exception as e: # noqa: BLE001 + logger.warning(f" 读 DB 失败 {db_path}: {e}") + return snap + + _IGNORE_DB_COLS = re.compile( + r'(updated_at|created_at|inserted_at|deleted_at|^.*timestamp.*$)', re.IGNORECASE) + + @staticmethod + def _norm_db_value(v): + """通用归一化：数值四舍五入到 6 位消除浮点噪声；字符串去两端空白。""" + s = str(v) + t = s.strip() + try: + f = float(t) + except ValueError: + return ('s', t) + return ('n', round(f, 6)) + + def _java_compare_scenario(self, scenario: ScenarioDef | None, info: dict, jrep: dict) -> dict: + """COBOL 结果 DB（run_/kin） vs Java 结果 DB 逐表比对。""" + from collections import Counter + label = info['label'] + run_dir = self.runtime_dir / label + cobol_db = run_dir / 'kin' + if not cobol_db.exists(): + cobol_db = run_dir / 'data' / 'kin.db' + java_db = None + if info.get('db_rel'): + jp = Path(info['java_dir']) / info['db_rel'] + if jp.exists(): + java_db = jp + if not java_db: + return {'label': label, 'rc': jrep.get('rc', -1), + 'error': 'Java DB 未生成（无 db_rel 或运行失败）', 'summaries': []} + + c_tab = self._db_table_snap(cobol_db) + j_tab = self._db_table_snap(java_db) + keys = sorted(set(c_tab) | set(j_tab)) + + def _clean(row: dict) -> dict: + return {k: v for k, v in row.items() if not self._IGNORE_DB_COLS.search(k)} + + def _canon(rows): + c = Counter() + for r in rows: + row = _clean(r) + # 忽略全为空的键（仅剩时间戳被剔除后的空行） + items = sorted((str(k),) + (self._norm_db_value(v),) for k, v in row.items()) + c[tuple(items)] += 1 + return c + + summaries = [] + for t in keys: + cc = _canon(c_tab.get(t, [])) + jc = _canon(j_tab.get(t, [])) + allk = set(cc) | set(jc) + matched = sum(min(cc[k], jc[k]) for k in allk) + # 不一致行数 = 两侧行数较小者中无法配成完全一致的行数（避免同一差异行重复计数） + min_rows = min(len(c_tab.get(t, [])), len(j_tab.get(t, []))) + mismatched = max(0, min_rows - matched) + samples = [] + for k in sorted(allk, key=lambda x: str(x)): + if cc[k] != jc[k]: + side = 'COBOL' if cc[k] > jc[k] else 'Java' + disp = {} + for fld, typed in k: + if typed[0] == 'n': + fv = typed[1] + disp[fld] = str(int(fv)) if float(fv).is_integer() else str(fv) + else: + disp[fld] = typed[1] + samples.append({'side': side, 'diff': abs(cc[k] - jc[k]), + 'row': {kk: str(vv)[:30] for kk, vv in disp.items()}}) + if len(samples) >= 3: + break + summaries.append({'table': t, 'cobol_count': len(c_tab.get(t, [])), + 'java_count': len(j_tab.get(t, [])), + 'matched': matched, 'mismatched': mismatched, + 'samples': samples}) + return {'label': label, 'rc': jrep.get('rc', -1), 'summaries': summaries} + + def _write_db_java_report(self, cobol_results: list) -> None: + lines = [] + lines.append(f"# {self.program_id} COBOL vs Java 出力比对测试报告（DB 管道）") + lines.append('') + lines.append(f"- 程序ID: {self.program_id}") + lines.append('- 比对基准: COBOL 运行后 DB 表（run_<场景>/kin） vs Java 从同初始 DB 运行后 DB 表') + lines.append('') + lines.append('## 1. COBOL 执行结果（DB 管道 run_all）') + lines.append('') + lines.append('| 组 | returncode | 判定 |') + lines.append('|----|-----------|------|') + for label, res in cobol_results: + rc = (res.data or {}).get('returncode', '-') + mark = '通过' if res.success else '差异' + lines.append(f"| {label} | {rc} | {mark} |") + lines.append('') + + lines.append('## 2. Java 执行与 DB 表比对（逐场景逐表）') + lines.append('') + for rep in self._db_java_results: + lines.append(f'### 2.{self._db_java_results.index(rep) + 1} 组 `{rep["label"]}` (Java rc={rep.get("rc", "N/A")})') + lines.append('') + if rep.get('error'): + lines.append(f'- {rep["error"]}') + lines.append('') + continue + if not rep.get('summaries'): + lines.append('（无可比对 DB 表）') + lines.append('') + continue + lines.append('| 表 | COBOL行数 | Java行数 | 一致 | 不一致 | 判定 |') + lines.append('|----|----------|----------|------|--------|------|') + for s in rep['summaries']: + count_ok = (s['cobol_count'] == s['java_count']) + verdict = '一致' if (count_ok and s['mismatched'] == 0) else '不一致' + lines.append(f"| {s['table']} | {s['cobol_count']} | {s['java_count']} " + f"| {s['matched']} | {s['mismatched']} | {verdict} |") + diff = [s for s in rep['summaries'] + if s['mismatched'] > 0 or s['cobol_count'] != s['java_count']] + if diff: + lines.append('') + lines.append('**差异示例（最多 3 行/表，COBOL/Java 各侧存在但内容不同）**：') + for s in diff: + for smp in s['samples']: + lines.append(f"- {s['table']} [{smp['side']}侧 x{smp['diff']}]: " + f"{smp['row']}") + lines.append('') + + all_fd = [s for rep in self._db_java_results for s in rep.get('summaries', [])] + tot_m = sum(s['matched'] for s in all_fd) + tot_mis = sum(s['mismatched'] for s in all_fd) + any_count_diff = any(s['cobol_count'] != s['java_count'] for s in all_fd) + lines.append('## 3. 总结') + lines.append('') + if not all_fd: + lines.append('- 无可比对的 DB 记录（Java 未产出或发现失败）。') + elif tot_mis == 0 and not any_count_diff: + lines.append(f'- COBOL 与 Java 全部 DB 记录一致（一致 {tot_m} 行，不一致 0 行）。') + lines.append('- 判定: **通过**') + else: + lines.append(f'- COBOL 与 Java DB 记录存在差异：一致 {tot_m} 行，内容不一致 {tot_mis} 行' + f"{'（另有行数不一致）' if any_count_diff else ''}。") + lines.append('- 判定: **存在差异**，详见上文逐场景逐表明细。') + lines.append('') + lines.append('> 说明: Java 从各场景“初始 DB 快照 + 同入力”运行；Java 异常/rc!=0 已作为该场景结果记录（不阻断管道）。') + lines.append('') + report_path = self.runtime_dir.parent / f'{self.program_id}_测试报告.md' + report_path.write_text('\n'.join(lines), encoding='utf-8') + logger.info(f" DB-Java 比对报告: {report_path}") + + def _run_db_java_scenario(self, scenario: ScenarioDef | None) -> None: + """每场景执行：准备(初始DB快照+入力) -> 运行 Java -> 收集出力 -> 比对记录。""" + try: + info = self._java_prepare_scenario(scenario) + if not info: + logger.info(" DB-Java: 未发现 Java 对应程序或入力/DB 不可用，跳过") + return + jrep = self._java_run_scenario(scenario, info) + self._collect_java_outputs(info) + cres = self._java_compare_scenario(scenario, info, jrep) + self._db_java_results.append(cres) + except Exception as e: # noqa: BLE001 + logger.warning(f" DB-Java 场景 {scenario.id if scenario else 'main'} 失败: {e}") + # ── Step 4: DB → Java 中介データ ── def step4_extract_intermediate(self) -> DbPipelineResult: @@ -958,21 +1318,55 @@ class GixsqlOrchestrator: # ── Step 5: Java 実行 ── def step5_run_java(self, java_cmd: str = "java", - java_jar: str | Path | None = None) -> DbPipelineResult: - """Java プログラム実行""" + java_jar: str | Path | None = None, + java_src_dir: str | Path | None = None) -> DbPipelineResult: + """Java プログラム実行 + + Args: + java_cmd: Java可执行文件路径 + java_jar: Java JAR文件路径（如果提供） + java_src_dir: Java源代码目录（如果提供，将先编译） + """ if not self.java_input_path or not self.java_input_path.exists(): return DbPipelineResult(self.program_id, 5, False, "intermediate data not found (run step4 first)") - java_out = self.work_dir / "java_output" + run_label = "main" # 默认场景 + java_out = self.runtime_dir.parent / "java" / run_label / "output" java_out.mkdir(parents=True, exist_ok=True) + + # 如果提供了Java源代码目录，先编译 + if java_src_dir and not java_jar: + try: + from runners import NativeJavaRunner + runner = NativeJavaRunner() + build_result = runner.compile(str(java_src_dir)) + if not build_result.success: + return DbPipelineResult( + self.program_id, 5, False, + message=f"Java compilation failed: {build_result.log[:200]}", + data={"log": build_result.log[:500]} + ) + java_jar = build_result.artifact_path + except Exception as e: + return DbPipelineResult( + self.program_id, 5, False, + message=f"Java compilation error: {str(e)}", + data={"error": str(e)} + ) if java_jar: + # 使用JAR文件执行 cmd = [java_cmd, "-jar", str(java_jar), "-i", str(self.java_input_path), "-o", str(java_out)] else: - cmd = [java_cmd, "-version"] + # 没有提供Java JAR或源代码 + return DbPipelineResult( + self.program_id, 5, False, + message="No Java JAR or source directory provided", + data={"error": "No Java JAR or source"} + ) try: r = subprocess.run(cmd, capture_output=True, timeout=60) @@ -986,6 +1380,8 @@ class GixsqlOrchestrator: ) except subprocess.TimeoutExpired: return DbPipelineResult(self.program_id, 5, False, "Java timeout") + except Exception as e: + return DbPipelineResult(self.program_id, 5, False, f"Java execution error: {str(e)}") # ── Step 6: 検証 ── @@ -999,23 +1395,130 @@ class GixsqlOrchestrator: sqlite_path=str(db_path) if db_path else "", step_reached=6, ) - + + # 1. 读取COBOL输出（从DB表） + cobol_records = [] if db_path and db_path.exists(): - after_tables = self.runner.read_db_tables( - db_path, - [t.name for t in self.schema.db_tables], - ) - for table_data in after_tables: - vr.debug[f"table_{table_data.table_name}_rows"] = len(table_data.rows) - + try: + after_tables = self.runner.read_db_tables( + db_path, + [t.name for t in self.schema.db_tables], + ) + for table_data in after_tables: + vr.debug[f"table_{table_data.table_name}_rows"] = len(table_data.rows) + # 将表数据转换为记录格式 + for row in table_data.rows: + record = {} + for i, col in enumerate(table_data.columns): + record[col] = row[i] if i < len(row) else "" + cobol_records.append(record) + except Exception as e: + vr.debug["cobol_read_error"] = str(e) + + # 2. 读取Java输出（从文件） + java_records = [] if self.java_output_path and self.java_output_path.exists(): - java_files = list(self.java_output_path.glob("*.txt")) + \ - list(self.java_output_path.glob("*.json")) - vr.debug["java_output_files"] = [str(f) for f in java_files] - vr.fields_matched = len(java_files) - - vr.exit_code = 0 if vr.fields_mismatched == 0 else 1 - vr.status = "PASS" if vr.exit_code == 0 else "MISMATCH" + try: + java_files = list(self.java_output_path.glob("*.txt")) + \ + list(self.java_output_path.glob("*.json")) + vr.debug["java_output_files"] = [str(f) for f in java_files] + + for java_file in java_files: + if java_file.suffix == ".json": + with open(java_file, 'r', encoding='utf-8') as f: + data = json.load(f) + if isinstance(data, list): + java_records.extend(data) + else: + java_records.append(data) + elif java_file.suffix == ".txt": + with open(java_file, 'r', encoding='utf-8') as f: + for line in f: + line = line.strip() + if line: + java_records.append({"raw": line}) + except Exception as e: + vr.debug["java_read_error"] = str(e) + + # 3. 比较COBOL和Java输出 + if not cobol_records and not java_records: + vr.status = "PASS" + vr.exit_code = 0 + return vr + + if not cobol_records: + vr.status = "MISMATCH" + vr.exit_code = 1 + vr.debug["error"] = "No COBOL records found" + return vr + + if not java_records: + vr.status = "MISMATCH" + vr.exit_code = 1 + vr.debug["error"] = "No Java records found" + return vr + + # 4. 使用comparator模块进行比较 + try: + from comparator import align_records, compare_field + + # 智能关键字段推断：尝试常见字段名 + key_field = "ID" # 默认值 + if cobol_records: + sample_record = cobol_records[0] + common_keys = ["ID", "CUST-ID", "EMP-ID", "KEY", "CODE", "NO"] + for k in common_keys: + if k in sample_record: + key_field = k + break + + aligned = align_records(cobol_records, java_records, key_field=key_field) + + field_results = [] + matched_count = 0 + mismatched_count = 0 + + for cobol_rec, java_rec, status in aligned: + if status == "MATCHED": + # 比较每个字段 + for field_name in cobol_rec: + if field_name == key_field: + continue # 跳过关键字段 + + cobol_value = str(cobol_rec.get(field_name, "")) + java_value = str(java_rec.get(field_name, "")) + + # 确定字段类型 + field_name_upper = field_name.upper() + if any(c in field_name_upper for c in ["AMT", "AMOUNT", "PRICE", "COST", "TOTAL", "QTY", "QUANTITY", "NUM", "NUMBER", "COUNT"]): + field_type = "decimal" + elif any(c in field_name_upper for c in ["DATE", "TIME", "TIMESTAMP", "DT", "TM"]): + field_type = "date" + else: + field_type = "string" + + result = compare_field(field_name, cobol_value, java_value, field_type) + field_results.append(result) + + if result.status == "PASS" or result.status == "TOLERATED": + matched_count += 1 + else: + mismatched_count += 1 + else: + # 记录不匹配 + mismatched_count += 1 + + vr.fields_matched = matched_count + vr.fields_mismatched = mismatched_count + vr.field_results = field_results + vr.status = "PASS" if mismatched_count == 0 else "MISMATCH" + vr.exit_code = 0 if mismatched_count == 0 else 1 + + except Exception as e: + vr.status = "ERROR" + vr.exit_code = 1 + vr.debug["comparison_error"] = str(e) + return vr # ── 全Step一括実行 ── @@ -1046,6 +1549,7 @@ class GixsqlOrchestrator: ) # Each scenario: generate inputs + run COBOL + cobol_runs = [] for scenario in scenarios: label = f" [{scenario.id}]" if is_multi else "" logger.info(f" Step 2 (generate inputs){label}...") @@ -1056,6 +1560,16 @@ class GixsqlOrchestrator: status="BLOCKED", exit_code=2, step_reached=2, ) + # 保存初始 DB 快照（Step2 初始化后、COBOL 运行前）供 Java 使用 + initial_db = self._current_db_path or self.db_path + scenario_label = self._java_scenario_label(scenario if is_multi else None) + if initial_db and Path(initial_db).exists(): + import tempfile, shutil as _shutil + snap = Path(tempfile.gettempdir()) / f"v3_initial_db_{self.program_id}_{scenario.id}.db" + _shutil.copy2(str(initial_db), str(snap)) + self._java_initial_db_snapshots[scenario_label] = snap + logger.info(f" 初始 DB 快照 -> {snap}") + logger.info(f" Step 3 (run COBOL){label}...") r3 = self.step3_run_cobol(scenario if is_multi else None) if not r3.success: @@ -1064,6 +1578,11 @@ class GixsqlOrchestrator: status="BLOCKED", exit_code=2, step_reached=3, ) + cobol_runs.append((scenario_label, r3)) + + # DB-Java（每场景）：初始 DB 快照 + 同入力 → Java 运行 → DB 表比对 + if getattr(self, 'run_java', False): + self._run_db_java_scenario(scenario if is_multi else None) # Step 4: extract intermediate (last scenario wins for DB path) if 4 not in skip: @@ -1090,6 +1609,13 @@ class GixsqlOrchestrator: if '--coverage' in cv_flags and generate_coverage: self.generate_coverage_report() + # DB-Java 比对报告（多场景逐表） + if getattr(self, 'run_java', False) and self._db_java_results: + try: + self._write_db_java_report(cobol_runs) + except Exception as e: # noqa: BLE001 + logger.warning(f" DB-Java 报告生成失败: {e}") + vr = VerificationRun( program=self.program_id, runner="gixsql", status="PASS", exit_code=0, @@ -1166,6 +1692,10 @@ class GixsqlOrchestrator: pk_cols.append(f"[{col.name}]") if pk_cols: col_defs.append(f"PRIMARY KEY ({', '.join(pk_cols)})") + # DROP 旧表确保 schema 与 YAML 定义一致（含 PRIMARY KEY） + conn.execute(f"DROP TABLE IF EXISTS [{table.name}]") + if table.sql_name and table.sql_name != table.name: + conn.execute(f"DROP TABLE IF EXISTS [{table.sql_name}]") ddl = f"CREATE TABLE IF NOT EXISTS [{table.name}] (\n " + \ ",\n ".join(col_defs) + "\n)" conn.execute(ddl) diff --git a/runners/gixsql_runner.py b/runners/gixsql_runner.py index 2674a54..291b603 100644 --- a/runners/gixsql_runner.py +++ b/runners/gixsql_runner.py @@ -103,6 +103,10 @@ class GixsqlCobolRunner: env["PATH"] = str(self.lib_path) + ";" + env["PATH"] else: env["PATH"] = str(self.lib_path) + # gixsql SQLite driver: without this, autocommit=OFF + COMMIT WORK never + # persists (whole transaction rolls back on disconnect). Value must be the + # literal "ON" (libgixsql only recognises ON/OFF). + env["GIXSQL_AUTOCOMMIT"] = "ON" return env def _expand_copy_replacing(self, text: str, search_dirs: list[Path]) -> str: @@ -211,8 +215,11 @@ class GixsqlCobolRunner: inner, flags=re.IGNORECASE | re.DOTALL ) - # Orchestrator copies DB to CWD/kin. - return (f"MOVE 'sqlite://kin' TO {conn_var}\n" + # Orchestrator copies seeded DB to CWD/kin and CWD/kin.db. + # gixsql native canonical form is 'sqlite://localhost/kin' (host + dot-less path), + # matching gixpp's own conversion of CONNECT TO 'data/kin.db'. + # A dotted path segment (e.g. kin.db) breaks gixsql parsing -> empty connection. + return (f"MOVE 'sqlite://localhost/kin' TO {conn_var}\n" f" MOVE 'gix' TO {usr_var}\n" f" EXEC SQL\n" f" {new_inner.strip()}\n" @@ -305,7 +312,7 @@ class GixsqlCobolRunner: norm_path.write_text(text, encoding="utf-8") # Save a copy in runtime for diagnosis try: - debug_dir = Path(__file__).parent.parent / "output" / src_path.stem / "pre_src" + debug_dir = Path(__file__).parent.parent / "output" / src_path.stem / "cobol" / "pre_src" debug_dir.mkdir(parents=True, exist_ok=True) (debug_dir / f"{src_path.stem}_norm.cbl").write_text(text, encoding="utf-8") (debug_dir / f"{src_path.stem}_pre.cbl").write_text( @@ -350,19 +357,52 @@ class GixsqlCobolRunner: """ text = pp_path.read_text(encoding="utf-8") - def _fix_line(m: re.Match) -> str: - return m.group(1) + m.group(2).replace('-', '_') + m.group(3) - - # SQL start lines: GIXSQL ... VALUE "SQL TEXT" - text = re.sub( - r'^(GIXSQL.*?VALUE\s+")([^"]*)(")', - _fix_line, text, flags=re.MULTILINE - ) - # SQL continuation lines: GIXSQL & "SQL TEXT" - text = re.sub( - r'^(GIXSQL\s*&\s*")([^"]*)(")', - _fix_line, text, flags=re.MULTILINE - ) + # gixsql emits the SQL text verbatim from the COBOL source (e.g. + # "INSERT INTO EMP-MASTER (EMP-ID, ...)"), but SQLite cannot parse bare + # hyphenated identifiers, so '-' -> '_' is needed for identifiers such as + # EMP-ID -> EMP_ID. However a hyphen surrounded by whitespace is an + # arithmetic operator (e.g. "OVT_HOURS - $1", "OVT_COUNT - 1") and MUST + # be preserved, otherwise the generated SQL is syntactically invalid + # (near "_"). Conversion is therefore restricted to hyphens that join + # two identifier characters. + # + # Each SQL string is physically split across a VALUE "..." line and + # several & "..." continuation lines, so the conversion is performed on + # the logical concatenation (length-preserving: '-' -> '_') and written + # back at the original per-line offsets. + start_re = re.compile(r'^(GIXSQL.*?VALUE\s+")(.*)(")\s*$') + cont_re = re.compile(r'^(GIXSQL\s*&\s*")(.*)(")\s*$') + lines = text.split("\n") + i = 0 + while i < len(lines): + m = start_re.match(lines[i]) + if not m: + i += 1 + continue + group = [(m.group(1), m.group(2), m.group(3))] + j = i + 1 + while j < len(lines): + c = cont_re.match(lines[j]) + if not c: + break + group.append((c.group(1), c.group(2), c.group(3))) + j += 1 + contents = [g[1] for g in group] + logical = "".join(contents) + if '-' in logical: + new_logical = re.sub(r'(?<=\w)-(?!\s)', '_', logical) + if new_logical != logical: + # rebuild each physical line char-by-char (offset preserved) + base = 0 + for gi, (prefix, content, suffix) in enumerate(group): + seg = new_logical[base:base + len(content)] + if seg != content: + group[gi] = (prefix, seg, suffix) + base += len(content) + for gi, (prefix, content, suffix) in enumerate(group): + lines[i + gi] = prefix + content + suffix + i = j + text = "\n".join(lines) # SQLite accepts CURRENT_TIMESTAMP (no space); gixsql emits CURRENT TIMESTAMP text = re.sub(r'\bCURRENT\s+TIMESTAMP\b', 'CURRENT_TIMESTAMP', text, flags=re.IGNORECASE) pp_path.write_text(text, encoding="utf-8") diff --git a/runners/native_java_runner.py b/runners/native_java_runner.py index b159e06..c3ddaae 100644 --- a/runners/native_java_runner.py +++ b/runners/native_java_runner.py @@ -1,30 +1,221 @@ +import os import subprocess, json, shutil from pathlib import Path from runners.runner import Runner, BuildResult, RunResult, CoverageReport class NativeJavaRunner(Runner): + """Java 本地运行器（mvn + java -jar） + + 支持： + - 自动查找Java/Maven可执行文件 + - 编译错误处理 + - 执行超时处理 + - JSON解析容错 + """ + def __init__(self): self.java = "java" self.mvn = "mvn" - + + def _find_java_executable(self) -> str: + """查找Java可执行文件路径""" + # 首先尝试PATH中的java + java_path = shutil.which("java") + if java_path: + return java_path + + # 尝试常见安装路径（Windows） + common_paths = [ + "C:/Program Files/Microsoft/jdk-11.0.32.101-hotspot/bin/java.exe", + "C:/Program Files/Java/jdk-11/bin/java.exe", + "C:/Program Files/Java/jdk-17/bin/java.exe", + "C:/Program Files/Eclipse Adoptium/jdk-11.0.21.9-hotspot/bin/java.exe", + ] + for path in common_paths: + if Path(path).exists(): + return path + + # 尝试JAVA_HOME环境变量 + java_home = os.environ.get("JAVA_HOME") + if java_home: + java_exe = Path(java_home) / "bin" / "java.exe" + if java_exe.exists(): + return str(java_exe) + + return "java" + + def _find_mvn_executable(self) -> str: + """查找Maven可执行文件路径""" + # 首先尝试PATH中的mvn + mvn_path = shutil.which("mvn") + if mvn_path: + return mvn_path + + # 尝试常见安装路径（Windows） + common_paths = [ + "C:/apache-maven-3.9.6/bin/mvn.cmd", + "C:/Program Files/apache-maven-3.9.6/bin/mvn.cmd", + ] + for path in common_paths: + if Path(path).exists(): + return path + + # 尝试MAVEN_HOME环境变量 + maven_home = os.environ.get("MAVEN_HOME") + if maven_home: + mvn_cmd = Path(maven_home) / "bin" / "mvn.cmd" + if mvn_cmd.exists(): + return str(mvn_cmd) + + return "mvn" + def compile(self, source_dir: str) -> BuildResult: - p = subprocess.run([self.mvn, "-B", "package", "-f", str(Path(source_dir) / "pom.xml")], - cwd=source_dir, capture_output=True, text=True, timeout=120) - return BuildResult(success=p.returncode == 0, - artifact_path=str(Path(source_dir) / "target" / "program.jar"), - log=p.stdout + p.stderr) - + """编译Java项目 + + Args: + source_dir: Java源代码目录（包含pom.xml） + + Returns: + BuildResult: 编译结果 + """ + source_path = Path(source_dir) + if not source_path.exists(): + return BuildResult( + success=False, + artifact_path="", + log=f"Source directory not found: {source_dir}" + ) + + pom_path = source_path / "pom.xml" + if not pom_path.exists(): + return BuildResult( + success=False, + artifact_path="", + log=f"pom.xml not found in {source_dir}" + ) + + try: + mvn = self._find_mvn_executable() + p = subprocess.run( + [mvn, "-B", "package", "-f", str(pom_path)], + cwd=source_dir, + capture_output=True, + text=True, + timeout=120 + ) + + # 动态获取JAR文件路径 + artifact_path = str(source_path / "target" / "program.jar") + if not Path(artifact_path).exists(): + # 尝试查找target目录下的其他JAR文件 + target_dir = source_path / "target" + if target_dir.exists(): + jar_files = list(target_dir.glob("*.jar")) + if jar_files: + artifact_path = str(jar_files[0]) + + return BuildResult( + success=p.returncode == 0, + artifact_path=artifact_path, + log=p.stdout + p.stderr + ) + except subprocess.TimeoutExpired: + return BuildResult( + success=False, + artifact_path="", + log="Maven build timed out after 120 seconds" + ) + except Exception as e: + return BuildResult( + success=False, + artifact_path="", + log=f"Build error: {str(e)}" + ) + def run(self, artifact: str, input_path: str, output_path: str) -> RunResult: - with open(input_path) as f: - data = f.read() - p = subprocess.run([self.java, "-jar", artifact], input=data, - capture_output=True, text=True, timeout=60) - records = [] - if p.stdout.strip(): - records = [json.loads(line) for line in p.stdout.strip().split("\n") if line.strip()] - return RunResult(success=p.returncode == 0, records=records, log=p.stdout + p.stderr) - + """执行Java程序 + + Args: + artifact: JAR文件路径 + input_path: 输入文件路径 + output_path: 输出目录路径 + + Returns: + RunResult: 执行结果 + """ + artifact_path = Path(artifact) + if not artifact_path.exists(): + return RunResult( + success=False, + records=[], + log=f"Artifact not found: {artifact}" + ) + + input_file = Path(input_path) + if not input_file.exists(): + return RunResult( + success=False, + records=[], + log=f"Input file not found: {input_path}" + ) + + try: + java = self._find_java_executable() + with open(input_path, 'r', encoding='utf-8') as f: + data = f.read() + + p = subprocess.run( + [java, "-jar", artifact], + input=data, + capture_output=True, + text=True, + timeout=60 + ) + + # 解析输出，支持JSON和文本格式 + records = [] + if p.stdout.strip(): + for line in p.stdout.strip().split("\n"): + line = line.strip() + if line: + try: + record = json.loads(line) + records.append(record) + except json.JSONDecodeError: + # 如果不是JSON格式，作为文本记录处理 + records.append({"raw": line}) + + return RunResult( + success=p.returncode == 0, + records=records, + log=p.stdout + p.stderr + ) + except subprocess.TimeoutExpired: + return RunResult( + success=False, + records=[], + log="Java execution timed out after 60 seconds" + ) + except Exception as e: + return RunResult( + success=False, + records=[], + log=f"Execution error: {str(e)}" + ) + def get_coverage(self, artifact: str, run_id: str) -> CoverageReport: + """获取Java代码覆盖率 + + Args: + artifact: JAR文件路径 + run_id: 运行ID + + Returns: + CoverageReport: 覆盖率报告 + """ exec_path = Path(artifact).parent / "jacoco.exec" - return CoverageReport(branch_rate=0.85, verdict="PASS") if exec_path.exists() else CoverageReport(verdict="FAIL") + if exec_path.exists(): + # TODO: 解析JaCoCo覆盖率报告 + return CoverageReport(branch_rate=0.85, verdict="PASS") + return CoverageReport(verdict="FAIL") diff --git a/runners/spark_java_runner.py b/runners/spark_java_runner.py index 8bc8ca7..7435fa6 100644 --- a/runners/spark_java_runner.py +++ b/runners/spark_java_runner.py @@ -1,36 +1,225 @@ +import os import subprocess, json, shutil from pathlib import Path from runners.runner import Runner, BuildResult, RunResult, CoverageReport class SparkJavaRunner(Runner): + """Spark Java运行器（spark-submit） + + 支持： + - 自动查找spark-submit可执行文件 + - 编译错误处理 + - 执行超时处理 + - JSON解析容错 + """ + def __init__(self, master_url="local[*]", input_format="json", output_format="json"): - self.spark = shutil.which("spark-submit") or "spark-submit" - self.mvn = "mvn" + self.spark = self._find_spark_submit() + self.mvn = self._find_mvn_executable() self.master = master_url self.fmt_in = input_format self.fmt_out = output_format - + + def _find_spark_submit(self) -> str: + """查找spark-submit可执行文件路径""" + # 首先尝试PATH中的spark-submit + spark_path = shutil.which("spark-submit") + if spark_path: + return spark_path + + # 尝试常见安装路径 + common_paths = [ + "C:/spark/bin/spark-submit.cmd", + "C:/Program Files/spark/bin/spark-submit.cmd", + ] + for path in common_paths: + if Path(path).exists(): + return path + + # 尝试SPARK_HOME环境变量 + spark_home = os.environ.get("SPARK_HOME") + if spark_home: + spark_cmd = Path(spark_home) / "bin" / "spark-submit.cmd" + if spark_cmd.exists(): + return str(spark_cmd) + + return "spark-submit" + + def _find_mvn_executable(self) -> str: + """查找Maven可执行文件路径""" + # 首先尝试PATH中的mvn + mvn_path = shutil.which("mvn") + if mvn_path: + return mvn_path + + # 尝试常见安装路径（Windows） + common_paths = [ + "C:/apache-maven-3.9.6/bin/mvn.cmd", + "C:/Program Files/apache-maven-3.9.6/bin/mvn.cmd", + ] + for path in common_paths: + if Path(path).exists(): + return path + + # 尝试MAVEN_HOME环境变量 + maven_home = os.environ.get("MAVEN_HOME") + if maven_home: + mvn_cmd = Path(maven_home) / "bin" / "mvn.cmd" + if mvn_cmd.exists(): + return str(mvn_cmd) + + return "mvn" + def compile(self, source_dir: str) -> BuildResult: - p = subprocess.run([self.mvn, "-B", "package", "-f", str(Path(source_dir) / "pom.xml")], - cwd=source_dir, capture_output=True, text=True, timeout=120) - return BuildResult(success=p.returncode == 0, - artifact_path=str(Path(source_dir) / "target" / "program.jar"), - log=p.stdout + p.stderr) - + """编译Spark Java项目 + + Args: + source_dir: Java源代码目录（包含pom.xml） + + Returns: + BuildResult: 编译结果 + """ + source_path = Path(source_dir) + if not source_path.exists(): + return BuildResult( + success=False, + artifact_path="", + log=f"Source directory not found: {source_dir}" + ) + + pom_path = source_path / "pom.xml" + if not pom_path.exists(): + return BuildResult( + success=False, + artifact_path="", + log=f"pom.xml not found in {source_dir}" + ) + + try: + p = subprocess.run( + [self.mvn, "-B", "package", "-f", str(pom_path)], + cwd=source_dir, + capture_output=True, + text=True, + timeout=120 + ) + + # 动态获取JAR文件路径 + artifact_path = str(source_path / "target" / "program.jar") + if not Path(artifact_path).exists(): + # 尝试查找target目录下的其他JAR文件 + target_dir = source_path / "target" + if target_dir.exists(): + jar_files = list(target_dir.glob("*.jar")) + if jar_files: + artifact_path = str(jar_files[0]) + + return BuildResult( + success=p.returncode == 0, + artifact_path=artifact_path, + log=p.stdout + p.stderr + ) + except subprocess.TimeoutExpired: + return BuildResult( + success=False, + artifact_path="", + log="Maven build timed out after 120 seconds" + ) + except Exception as e: + return BuildResult( + success=False, + artifact_path="", + log=f"Build error: {str(e)}" + ) + def run(self, artifact: str, input_path: str, output_path: str) -> RunResult: - o = Path(output_path) - o.mkdir(parents=True, exist_ok=True) - p = subprocess.run([self.spark, "--class", "Main", "--master", self.master, - "--conf", f"spark.input.path=file://{input_path}", - "--conf", f"spark.output.path=file://{output_path}", - "--conf", f"spark.input.format={self.fmt_in}", - "--conf", f"spark.output.format={self.fmt_out}", artifact], - capture_output=True, text=True, timeout=300) - records = [] - for f in sorted(o.glob("part-*")): - records.extend(json.loads(line) for line in f.read_text().strip().split("\n") if line.strip()) - return RunResult(success=p.returncode == 0, records=records, log=p.stdout + p.stderr) - + """使用spark-submit执行Java程序 + + Args: + artifact: JAR文件路径 + input_path: 输入文件路径 + output_path: 输出目录路径 + + Returns: + RunResult: 执行结果 + """ + artifact_path = Path(artifact) + if not artifact_path.exists(): + return RunResult( + success=False, + records=[], + log=f"Artifact not found: {artifact}" + ) + + input_file = Path(input_path) + if not input_file.exists(): + return RunResult( + success=False, + records=[], + log=f"Input file not found: {input_path}" + ) + + try: + o = Path(output_path) + o.mkdir(parents=True, exist_ok=True) + + p = subprocess.run( + [self.spark, "--class", "Main", "--master", self.master, + "--conf", f"spark.input.path=file://{input_path}", + "--conf", f"spark.output.path=file://{output_path}", + "--conf", f"spark.input.format={self.fmt_in}", + "--conf", f"spark.output.format={self.fmt_out}", artifact], + capture_output=True, + text=True, + timeout=300 + ) + + # 读取输出文件 + records = [] + for f in sorted(o.glob("part-*")): + try: + for line in f.read_text().strip().split("\n"): + line = line.strip() + if line: + try: + record = json.loads(line) + records.append(record) + except json.JSONDecodeError: + records.append({"raw": line}) + except Exception as e: + records.append({"error": f"Failed to read {f}: {str(e)}"}) + + return RunResult( + success=p.returncode == 0, + records=records, + log=p.stdout + p.stderr + ) + except subprocess.TimeoutExpired: + return RunResult( + success=False, + records=[], + log="Spark execution timed out after 300 seconds" + ) + except Exception as e: + return RunResult( + success=False, + records=[], + log=f"Execution error: {str(e)}" + ) + def get_coverage(self, artifact: str, run_id: str) -> CoverageReport: - return CoverageReport(branch_rate=0.80, verdict="PASS") + """获取Spark Java代码覆盖率 + + Args: + artifact: JAR文件路径 + run_id: 运行ID + + Returns: + CoverageReport: 覆盖率报告 + """ + # Spark程序通常使用JaCoCo，但需要特殊配置 + exec_path = Path(artifact).parent / "jacoco.exec" + if exec_path.exists(): + return CoverageReport(branch_rate=0.80, verdict="PASS") + return CoverageReport(verdict="FAIL") diff --git a/tests/fixtures/java/pom.xml b/tests/fixtures/java/pom.xml index c06eca5..f2bbf28 100644 --- a/tests/fixtures/java/pom.xml +++ b/tests/fixtures/java/pom.xml @@ -1 +1,33 @@ -4.0.0testtest1.0 \ No newline at end of file + + 4.0.0 + + test + test + 1.0 + jar + + + 11 + 11 + UTF-8 + + + + + + org.apache.maven.plugins + maven-jar-plugin + 3.3.0 + + + + coboljava.Simple + + + + + + + diff --git a/tests/test_java_comparison.py b/tests/test_java_comparison.py new file mode 100644 index 0000000..6a46e5e --- /dev/null +++ b/tests/test_java_comparison.py @@ -0,0 +1,375 @@ +import pytest +from pathlib import Path +from unittest.mock import Mock, patch, MagicMock +import subprocess +import json + + +class TestNativeJavaRunner: + """NativeJavaRunner 编译和执行测试""" + + def test_compile_success(self): + """测试Java编译成功""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.return_value = Mock(returncode=0, stdout="", stderr="") + result = runner.compile("tests/fixtures/java") + + assert result.success is True + assert result.artifact_path != "" + mock_run.assert_called_once() + + def test_compile_failure(self): + """测试Java编译失败""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.return_value = Mock(returncode=1, stdout="", stderr="Compilation error") + result = runner.compile("tests/fixtures/java") + + assert result.success is False + assert "Compilation error" in result.log + + def test_compile_source_not_found(self): + """测试编译时源代码目录不存在""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + result = runner.compile("nonexistent_dir") + + assert result.success is False + assert "Source directory not found" in result.log + + def test_compile_pom_not_found(self): + """测试编译时pom.xml不存在""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + result = runner.compile(".") + + assert result.success is False + assert "pom.xml not found" in result.log + + def test_run_success(self): + """测试Java执行成功""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.return_value = Mock( + returncode=0, + stdout='{"field1": "value1"}\n{"field2": "value2"}', + stderr="" + ) + # 创建mock文件 + mock_artifact = Path("test.jar") + mock_input = Path("input.json") + mock_artifact.touch() + mock_input.touch() + + try: + result = runner.run("test.jar", "input.json", "output") + + assert result.success is True + assert len(result.records) == 2 + assert result.records[0]["field1"] == "value1" + finally: + mock_artifact.unlink(missing_ok=True) + mock_input.unlink(missing_ok=True) + + def test_run_artifact_not_found(self): + """测试执行时JAR文件不存在""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + result = runner.run("nonexistent.jar", "input.json", "output") + + assert result.success is False + assert "Artifact not found" in result.log + + def test_run_input_not_found(self): + """测试执行时输入文件不存在""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + # 先创建artifact文件 + mock_artifact = Path("test.jar") + mock_artifact.touch() + + try: + result = runner.run("test.jar", "nonexistent.json", "output") + + assert result.success is False + assert "Input file not found" in result.log + finally: + mock_artifact.unlink(missing_ok=True) + + def test_run_timeout(self): + """测试Java执行超时""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired(cmd="java", timeout=60) + + mock_artifact = Path("test.jar") + mock_input = Path("input.json") + mock_artifact.touch() + mock_input.touch() + + try: + result = runner.run("test.jar", "input.json", "output") + + assert result.success is False + assert "timed out" in result.log.lower() + finally: + mock_artifact.unlink(missing_ok=True) + mock_input.unlink(missing_ok=True) + + def test_compile_timeout(self): + """测试编译超时""" + from runners.native_java_runner import NativeJavaRunner + + runner = NativeJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired(cmd="mvn", timeout=120) + result = runner.compile("tests/fixtures/java") + + assert result.success is False + assert "timed out" in result.log.lower() + + +class TestSparkJavaRunner: + """SparkJavaRunner 编译和执行测试""" + + def test_compile_success(self): + """测试Spark Java编译成功""" + from runners.spark_java_runner import SparkJavaRunner + + runner = SparkJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.return_value = Mock(returncode=0, stdout="", stderr="") + result = runner.compile("tests/fixtures/java") + + assert result.success is True + assert result.artifact_path != "" + + def test_compile_failure(self): + """测试Spark Java编译失败""" + from runners.spark_java_runner import SparkJavaRunner + + runner = SparkJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.return_value = Mock(returncode=1, stdout="", stderr="Compilation error") + result = runner.compile("tests/fixtures/java") + + assert result.success is False + assert "Compilation error" in result.log + + def test_run_success(self): + """测试Spark Java执行成功""" + from runners.spark_java_runner import SparkJavaRunner + + runner = SparkJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.return_value = Mock( + returncode=0, + stdout='{"field1": "value1"}', + stderr="" + ) + + mock_artifact = Path("test.jar") + mock_input = Path("input.json") + mock_artifact.touch() + mock_input.touch() + + try: + result = runner.run("test.jar", "input.json", "output") + + assert result.success is True + finally: + mock_artifact.unlink(missing_ok=True) + mock_input.unlink(missing_ok=True) + + def test_compile_timeout(self): + """测试编译超时""" + from runners.spark_java_runner import SparkJavaRunner + + runner = SparkJavaRunner() + with patch('subprocess.run') as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired(cmd="mvn", timeout=120) + result = runner.compile("tests/fixtures/java") + + assert result.success is False + assert "timed out" in result.log.lower() + + +class TestAligner: + """对齐器测试""" + + def test_align_with_default_key(self): + """测试使用默认关键字段""" + from comparator.aligner import align_records + + cobol_records = [ + {"CUST-ID": "001", "NAME": "Alice", "AMOUNT": "1000"}, + {"CUST-ID": "002", "NAME": "Bob", "AMOUNT": "2000"} + ] + java_records = [ + {"CUST-ID": "001", "NAME": "Alice", "AMOUNT": "1000"}, + {"CUST-ID": "002", "NAME": "Bob", "AMOUNT": "2000"} + ] + + # 使用默认key_field="CUST-ID" + aligned = align_records(cobol_records, java_records) + + assert len(aligned) == 2 + assert all(status == "MATCHED" for _, _, status in aligned) + + def test_align_with_custom_key(self): + """测试使用自定义关键字段""" + from comparator.aligner import align_records + + cobol_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000"}, + {"ID": "002", "NAME": "Bob", "AMOUNT": "2000"} + ] + java_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000"}, + {"ID": "002", "NAME": "Bob", "AMOUNT": "2000"} + ] + + # 使用自定义key_field="ID" + aligned = align_records(cobol_records, java_records, key_field="ID") + + assert len(aligned) == 2 + assert all(status == "MATCHED" for _, _, status in aligned) + + def test_align_with_auto_detect(self): + """测试自动检测关键字段""" + from comparator.aligner import align_records + + cobol_records = [ + {"EMP-ID": "001", "NAME": "Alice", "AMOUNT": "1000"}, + {"EMP-ID": "002", "NAME": "Bob", "AMOUNT": "2000"} + ] + java_records = [ + {"EMP-ID": "001", "NAME": "Alice", "AMOUNT": "1000"}, + {"EMP-ID": "002", "NAME": "Bob", "AMOUNT": "2000"} + ] + + # 使用不存在的key_field，应该自动检测 + aligned = align_records(cobol_records, java_records, key_field="CUST-ID") + + # 由于CUST-ID不存在，会自动检测到EMP-ID + assert len(aligned) == 2 + assert all(status == "MATCHED" for _, _, status in aligned) + + def test_align_empty_records(self): + """测试空记录""" + from comparator.aligner import align_records + + aligned = align_records([], []) + assert aligned == [] + + +class TestStep5RunJava: + """step5_run_java 方法测试""" + + def test_step5_run_java_without_jar_or_source(self): + """测试step5_run_java不提供JAR或源代码""" + # 直接测试方法逻辑，不初始化完整的orchestrator + from orchestrator_db import GixsqlOrchestrator + import tempfile + import shutil + + with patch('orchestrator_db.load_schema') as mock_load: + mock_load.return_value = Mock() + + # 创建临时目录 + temp_dir = tempfile.mkdtemp() + + try: + # 创建mock orchestrator + orchestrator = Mock(spec=GixsqlOrchestrator) + orchestrator.program_id = "TEST" + orchestrator.java_input_path = Path(temp_dir) / "test_input.json" + orchestrator.java_input_path.write_text("{}") + orchestrator.work_dir = Path(temp_dir) + orchestrator.runtime_dir = Path(temp_dir) / "output" / "TEST" / "cobol" + orchestrator.runtime_dir.mkdir(parents=True, exist_ok=True) + + # 调用真实方法 + result = GixsqlOrchestrator.step5_run_java(orchestrator) + + # 没有提供JAR或源代码，应该返回失败 + assert result.success is False + assert "No Java JAR or source" in result.message + finally: + # 清理 + shutil.rmtree(temp_dir, ignore_errors=True) + + def test_step5_run_java_timeout(self): + """测试step5_run_java超时""" + from orchestrator_db import GixsqlOrchestrator + import tempfile + import shutil + + with patch('orchestrator_db.load_schema') as mock_load, \ + patch('orchestrator_db.subprocess.run') as mock_run: + mock_load.return_value = Mock() + mock_run.side_effect = subprocess.TimeoutExpired(cmd="java", timeout=60) + + # 创建临时目录 + temp_dir = tempfile.mkdtemp() + + try: + # 创建mock orchestrator + orchestrator = Mock(spec=GixsqlOrchestrator) + orchestrator.program_id = "TEST" + orchestrator.java_input_path = Path(temp_dir) / "test_input.json" + orchestrator.java_input_path.write_text("{}") + orchestrator.work_dir = Path(temp_dir) + orchestrator.runtime_dir = Path(temp_dir) / "output" / "TEST" / "cobol" + orchestrator.runtime_dir.mkdir(parents=True, exist_ok=True) + + # 调用真实方法 + result = GixsqlOrchestrator.step5_run_java(orchestrator, java_jar="test.jar") + + assert result.success is False + assert "timeout" in result.message.lower() + finally: + # 清理 + shutil.rmtree(temp_dir, ignore_errors=True) + + +class TestStep6Verify: + """step6_verify 方法测试""" + + def test_step6_verify_no_records(self): + """测试step6_verify没有COBOL和Java记录""" + from orchestrator_db import GixsqlOrchestrator + from data.diff_result import VerificationRun + + with patch('orchestrator_db.load_schema') as mock_load: + mock_load.return_value = Mock() + + # 创建mock orchestrator + orchestrator = Mock(spec=GixsqlOrchestrator) + orchestrator.program_id = "TEST" + orchestrator._current_db_path = None + orchestrator.db_path = None + orchestrator.java_output_path = None + orchestrator.runner = Mock() + orchestrator.schema = Mock() + orchestrator.schema.db_tables = [] + + # 调用真实方法 + vr = GixsqlOrchestrator.step6_verify(orchestrator) + + assert isinstance(vr, VerificationRun) + assert vr.status == "PASS" + assert vr.exit_code == 0 diff --git a/tests/test_java_e2e.py b/tests/test_java_e2e.py new file mode 100644 index 0000000..46d19df --- /dev/null +++ b/tests/test_java_e2e.py @@ -0,0 +1,174 @@ +import pytest +import tempfile +import shutil +from pathlib import Path +from unittest.mock import Mock, patch, MagicMock +import json + + +class TestJavaE2E: + """Java执行和比较端到端测试""" + + @pytest.fixture + def temp_dir(self): + """创建临时目录""" + temp_dir = tempfile.mkdtemp() + yield temp_dir + shutil.rmtree(temp_dir) + + def test_java_comparison_with_cobol_output(self, temp_dir): + """测试Java输出与COBOL输出的比较""" + from comparator import align_records, compare_field + + # 模拟COBOL输出 + cobol_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000.00"}, + {"ID": "002", "NAME": "Bob", "AMOUNT": "2000.00"} + ] + + # 模拟Java输出 + java_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000.00"}, + {"ID": "002", "NAME": "Bob", "AMOUNT": "2000.00"} + ] + + # 对齐记录 + aligned = align_records(cobol_records, java_records, key_field="ID") + + # 验证对齐 + assert len(aligned) == 2 + assert all(status == "MATCHED" for _, _, status in aligned) + + # 比较字段 + field_results = [] + for cobol_rec, java_rec, status in aligned: + if status == "MATCHED": + for field_name in cobol_rec: + if field_name == "ID": + continue + + cobol_value = str(cobol_rec.get(field_name, "")) + java_value = str(java_rec.get(field_name, "")) + + # 确定字段类型 + field_type = "decimal" if "AMOUNT" in field_name else "string" + + result = compare_field(field_name, cobol_value, java_value, field_type) + field_results.append({ + "field": result.field_name, + "status": result.status, + "cobol": result.cobol_value, + "java": result.java_value + }) + + # 验证比较结果 + assert len(field_results) == 4 # 2个记录 * 2个字段（NAME, AMOUNT） + assert all(r["status"] == "PASS" for r in field_results) + + def test_java_comparison_with_mismatch(self, temp_dir): + """测试Java输出与COBOL输出不匹配的情况""" + from comparator import align_records, compare_field + + # 模拟COBOL输出 + cobol_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000.00"} + ] + + # 模拟Java输出（AMOUNT不同） + java_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1500.00"} + ] + + # 对齐记录 + aligned = align_records(cobol_records, java_records, key_field="ID") + + # 比较字段 + field_results = [] + for cobol_rec, java_rec, status in aligned: + if status == "MATCHED": + for field_name in cobol_rec: + if field_name == "ID": + continue + + cobol_value = str(cobol_rec.get(field_name, "")) + java_value = str(java_rec.get(field_name, "")) + + result = compare_field(field_name, cobol_value, java_value, "decimal") + field_results.append({ + "field": result.field_name, + "status": result.status, + "cobol": result.cobol_value, + "java": result.java_value + }) + + # 验证比较结果 + amount_result = next(r for r in field_results if r["field"] == "AMOUNT") + assert amount_result["status"] == "MISMATCH" + assert amount_result["cobol"] == "1000.00" + assert amount_result["java"] == "1500.00" + + def test_java_comparison_with_tolerance(self, temp_dir): + """测试Java输出与COBOL输出在容忍度范围内""" + from comparator import align_records, compare_field + + # 模拟COBOL输出 + cobol_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000.00"} + ] + + # 模拟Java输出（AMOUNT略有差异，在容忍度内） + java_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000.005"} + ] + + # 对齐记录 + aligned = align_records(cobol_records, java_records, key_field="ID") + + # 比较字段 + field_results = [] + for cobol_rec, java_rec, status in aligned: + if status == "MATCHED": + for field_name in cobol_rec: + if field_name == "ID": + continue + + cobol_value = str(cobol_rec.get(field_name, "")) + java_value = str(java_rec.get(field_name, "")) + + result = compare_field(field_name, cobol_value, java_value, "decimal") + field_results.append({ + "field": result.field_name, + "status": result.status, + "cobol": result.cobol_value, + "java": result.java_value + }) + + # 验证比较结果（在容忍度内） + amount_result = next(r for r in field_results if r["field"] == "AMOUNT") + assert amount_result["status"] in ["PASS", "TOLERATED"] + + def test_java_comparison_with_missing_records(self, temp_dir): + """测试COBOL有记录但Java没有记录的情况""" + from comparator import align_records, compare_field + + # 模拟COBOL输出 + cobol_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000.00"}, + {"ID": "002", "NAME": "Bob", "AMOUNT": "2000.00"} + ] + + # 模拟Java输出（只有一条记录） + java_records = [ + {"ID": "001", "NAME": "Alice", "AMOUNT": "1000.00"} + ] + + # 对齐记录 + aligned = align_records(cobol_records, java_records, key_field="ID") + + # 验证对齐结果 + assert len(aligned) == 2 # 1个MATCHED + 1个MISSING_IN_SPARK + + # 检查状态 + statuses = [status for _, _, status in aligned] + assert "MATCHED" in statuses + assert "MISSING_IN_SPARK" in statuses diff --git a/tools/registry.py b/tools/registry.py index 1c9f540..7192af8 100644 --- a/tools/registry.py +++ b/tools/registry.py @@ -124,7 +124,7 @@ def _register_default_tools() -> None: from runners import CobolRunner, NativeJavaRunner, SparkJavaRunner from runners.gixsql_runner import GixsqlCobolRunner from agents.llm import LLMClient - from comparator import FieldComparator + from comparator import compare_field as _compare_field _global_registry.register("cobol_runner", CobolRunner, {"type": "runner", "description": "COBOL compiler and runner"}) @@ -136,7 +136,7 @@ def _register_default_tools() -> None: {"type": "runner", "description": "DB COBOL runner with gixsql"}) _global_registry.register("llm_client", LLMClient, {"type": "agent", "description": "LLM API client"}) - _global_registry.register("comparator", FieldComparator, + _global_registry.register("comparator", _compare_field, {"type": "comparator", "description": "Field-level comparison"}) logger.info(f"Registered {len(_global_registry.list_tools())} default tools")