hangshuo652/cobol-java-v3
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: phase2 review fixes

Browse Source 
- TIME injection: change from spaces to '2500' (hour>23) for NUMVAL trigger
- Runner: add .resolve() to work_dir to fix chdir+relative path breakage
- Coverage: per-target field overrides for DP#9-#12 (START-DATE=20240115 etc.)
- .gitignore: add compilation artifacts, temp scripts, test outputs
This commit is contained in:
hangshuo652
2026-07-15 21:46:28 +08:00
parent f3be17e5eb
commit 54d4e81240
13 changed files with 769 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cobol_testgen/__init__.py
+116 -38
View File
@@ -291,6 +291,78 @@ def _inject_empty_emp_rec(records, fields):
logger.info(f" injected empty-EMP-ID record at position 0")
def _inject_c01_coverage_records(records, fields, base_assignments):
"""Inject records for uncovered C01 decision points (DP#9-#12 T).
Takes a base record with valid EMP-ID and all dates valid, then
modifies ONE CSV field per copy to trigger a specific SUB04CHK failure.
"""
if not records or not base_assignments:
return
r01_len = 80
for f in fields:
if isinstance(f, dict) and f.get('name') == 'R01LINE' and f.get('pic_info'):
r01_len = f['pic_info'].get('length', 80)
break
unstring_items = []
for tgt, alist in base_assignments.items():
for a in alist:
if a.get('type') == 'unstring_split' and a.get('source_vars'):
unstring_items.append((a.get('index', 0), tgt))
if not unstring_items:
return
unstring_items.sort(key=lambda x: x[0])
unstring_fields = [tgt for _, tgt in unstring_items]
base = None
for rec in records:
emp = str(rec.get('WRK-CSV-EMP-ID', '')).strip()
if emp and emp != '0':
base = rec
lt = str(rec.get('WRK-CSV-LEAVE-TYPE', '')).strip()
if lt and lt != '':
break
if base is None:
return
# (dp_num, csv_field, invalid_value, extra_valid_fixes)
# For DATE fields: SUB04CHK checks month 01-12, day 01-31 → spaces make month=0 < 1 → error
# For TIME fields: SUB04CHK checks hour 00-23, minute 00-59 → spaces make NUMVAL=0 → valid!
# Need explicit invalid value like '2500' (hour=25 > 23) or '0060' (minute=60 > 59)
# For EMP-ID fields: SUB04CHK checks spaces + alpha/digit/special → '00000000' is valid
field_invalid = {
'WRK-CSV-START-DATE': ' ',
'WRK-CSV-START-TIME': '2500',
'WRK-CSV-END-DATE': ' ',
'WRK-CSV-END-TIME': '2500',
}
field_sizes = {
'WRK-CSV-START-DATE': 8,
'WRK-CSV-START-TIME': 4,
'WRK-CSV-END-DATE': 8,
'WRK-CSV-END-TIME': 4,
}
targets = [
(9, 'WRK-CSV-START-DATE', {'WRK-CSV-START-DATE': ' '}),
(10, 'WRK-CSV-START-TIME', {'WRK-CSV-START-TIME': '2500', 'WRK-CSV-START-DATE': '20240115', 'WRK-CSV-END-DATE': '20240115'}),
(11, 'WRK-CSV-END-DATE', {'WRK-CSV-END-DATE': ' '}),
(12, 'WRK-CSV-END-TIME', {'WRK-CSV-END-TIME': '2500', 'WRK-CSV-START-DATE': '20240115', 'WRK-CSV-END-DATE': '20240115'}),
]
for dp_num, csv_field, overrides in targets:
if csv_field not in base:
continue
new_rec = dict(base)
for k, v in overrides.items():
new_rec[k] = v
csv_parts = [str(new_rec.get(fname, '')) for fname in unstring_fields]
csv_value = ','.join(csv_parts).ljust(r01_len)[:r01_len]
new_rec['R01LINE'] = csv_value
records.append(new_rec)
logger.info(f" injected DP#{dp_num} T coverage record ({csv_field}={repr(overrides.get(csv_field,''))})")
# ── 入口 ──
def main():
@@ -302,7 +374,6 @@ def main():
do_run = False
gcov_mode = False
gixsql_mode = False
temp_dir = None
if '--run' in args:
do_run = True
@@ -313,9 +384,6 @@ def main():
if not _HAVE_RUNNER:
logger.warning("--gcov: runner.py not found. Compile/run will be skipped. "
"Use --gcov without runner only generates test data + static coverage.")
if '--gixsql' in args:
gixsql_mode = True
args.remove('--gixsql')
i = 0
while i < len(args):
if args[i] == '--temp-dir':
@@ -372,8 +440,16 @@ def main():
programs = []
if gixsql_mode:
# DB pipeline: GixsqlOrchestrator
# ── Auto-route: split DB / non-DB per file ──
db_files = []
non_db_files = []
for f in cobol_files:
if 'EXEC SQL' in f.read_text(encoding='utf-8-sig').upper():
db_files.append(f)
else:
non_db_files.append(f)
if db_files:
import sys as _sys
_v3_root = str(Path(__file__).parent.parent)
if _v3_root not in _sys.path:
@@ -381,25 +457,22 @@ def main():
from orchestrator_db import GixsqlOrchestrator
from config import Config
config = Config()
src_dir = cobol_files[0].parent if cobol_files else Path.cwd()
cpy_dirs = [src_dir / '..' / 'cpy']
_db_config = Config()
_src_dir = cobol_files[0].parent if cobol_files else Path.cwd()
_cpy_dirs = [str(_src_dir / '..' / 'cpy')]
for filepath in cobol_files:
for filepath in db_files:
pid = filepath.stem
prog_outdir = outdir / pid
prog_outdir.mkdir(parents=True, exist_ok=True)
(prog_outdir / 'logs').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'input').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'output').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'json').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'coverage').mkdir(parents=True, exist_ok=True)
logger.info(f"\n========== DB: {pid} ==========")
orch = GixsqlOrchestrator(
config=config, program_id=pid,
cobol_src_dir=str(src_dir),
copybook_dirs=[str(d) for d in cpy_dirs],
config=_db_config, program_id=pid,
cobol_src_dir=str(_src_dir),
copybook_dirs=[str(d) for d in _cpy_dirs],
skip_jvm=True,
)
vr = orch.run_all(generate_coverage=False)
@@ -407,38 +480,38 @@ def main():
# Copy output files to outdir
if orch.runtime_dir.exists():
for item in orch.runtime_dir.iterdir():
if item.is_file():
shutil.copy2(str(item), str(prog_outdir / item.name))
if item.name == "gixsql.log":
continue
dst = prog_outdir / item.name
if item.is_dir():
shutil.copytree(str(item), str(dst), dirs_exist_ok=True)
else:
try:
shutil.copy2(str(item), str(dst))
except PermissionError:
logger.warning(f" Skipping locked file: {item.name}")
logger.info(f" {pid}: rc={vr.exit_code} status={vr.status}")
# Coverage report (only once, with correct output_dir)
if '--coverage' in getattr(config, 'gixsql_compile_flags', ''):
# Coverage report
if gcov_mode and '--coverage' in getattr(_db_config, 'gixsql_compile_flags', ''):
cov_result = orch.generate_coverage_report(output_dir=str(prog_outdir / 'coverage'))
if cov_result.success:
cv = cov_result.data.get("coverage", "unknown")
logger.info(f" Coverage: {cv}")
cov_dict = cov_result.data.get("_cov_dict")
if cov_dict:
# Fix detail_relpath relative to top-level index
rel = Path(prog_outdir / 'coverage' / f"{pid}_coverage.html")
cov_dict['detail_relpath'] = str(rel.relative_to(outdir).as_posix())
programs.append(cov_dict)
else:
logger.warning(" --coverage not in gixsql_compile_flags; skipping coverage")
if programs:
from cobol_testgen.coverage import generate_coverage_index as _gen_idx
_gen_idx(programs, outdir / 'coverage')
logger.info(f"\n覆盖率总览：{outdir / 'coverage' / 'index.html'}")
return
for filepath in cobol_files:
for filepath in non_db_files:
if not filepath.exists():
logger.error(f"错误：文件不存在 {filepath}")
continue
source = filepath.read_text(encoding='utf-8')
orig_source = source # 用于行号定位（与 gcov 对齐）
source = resolve_copybooks(
source,
str(filepath.parent),
@@ -527,9 +600,6 @@ def main():
prog_outdir = outdir / filepath.stem
prog_outdir.mkdir(parents=True, exist_ok=True)
(prog_outdir / 'logs').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'input').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'output').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'json').mkdir(parents=True, exist_ok=True)
(prog_outdir / 'coverage').mkdir(parents=True, exist_ok=True)
logger.info(f"\n========== {filepath.name} ==========")
@@ -581,6 +651,12 @@ def main():
skip_path_infos = [p for p in path_infos if _is_skip(p[0])]
main_path_infos = [p for p in path_infos if not _is_skip(p[0])]
_c01_types = {}
for pi in path_infos:
wants = tuple(c[3] for c in pi[0] if len(c) == 4 and c[0] == 'C01CHKRRC' and c[1] == '<>' and c[2] == 'ZERO')
if wants:
_c01_types[wants] = _c01_types.get(wants, 0) + 1
logger.info(f" C01 path types: {dict(sorted(_c01_types.items()))}")
path_infos = main_path_infos
if skip_path_infos:
logger.info(f" Skip 路径: {len(skip_path_infos)} 条（将单独生成数据集）")
@@ -633,6 +709,8 @@ def main():
# P4: inject empty EMP-ID record to trigger R01EMP-ID = SPACE path
_inject_empty_emp_rec(records, fields_dict)
# P5: inject records for uncovered C01 decision points (DP#9-#12 T)
_inject_c01_coverage_records(records, fields_dict, assignments)
if _HAVE_TOSQL:
sql_meta = collect_sql_meta(assignments, declared_columns)
@@ -643,7 +721,7 @@ def main():
else:
db_input = None
outpath = prog_outdir / 'json' / (filepath.stem + '.json')
outpath = prog_outdir / 'main' / 'json' / (filepath.stem + '.json')
output_json(records, outpath, roles,
fd_fields=fd_fields, field_to_fd=field_to_fd,
open_dir=open_dir,
@@ -653,7 +731,7 @@ def main():
select_info = parse_file_control(preprocessed)
output_input_files(records, prog_outdir / 'input', filepath.stem, roles,
output_input_files(records, prog_outdir / 'main' / 'input', filepath.stem, roles,
fd_fields, field_to_fd, open_dir,
term_types=term_types,
data_fields=fields_dict, select_info=select_info)
@@ -675,13 +753,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 / 'json' / (filepath.stem + '_skip.json')
skip_outpath = prog_outdir / '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 / 'input_skip'
skip_input_dir = prog_outdir / 'skip' / 'input'
output_input_files(skip_records, skip_input_dir,
filepath.stem + '_skip', roles,
fd_fields, field_to_fd, open_dir,
@@ -791,7 +869,7 @@ 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,
source, cov_prefix, index_relpath='index.html',
orig_source, cov_prefix, index_relpath='index.html',
gcov_data=gcov_data)
programs.append(cov_result)
programs[-1]['detail_relpath'] = f'{filepath.stem}/coverage/{filepath.stem}_coverage.html'