diff --git a/.gitignore b/.gitignore index d25ba26..ecd10a3 100644 --- a/.gitignore +++ b/.gitignore @@ -47,5 +47,20 @@ _test_flatfiles/ gixsql/ cobol-tna-system/ -# ԔOӋOӋԭ塢_ֹ -ԔOӋ/ +# Compilation artifacts +SUB*.c +SUB*.c.h +SUB*.c.l.h +SUB*.i + +# Temp scripts +diagnose_*.py + +# Generated reports/documents +SUMMARY.md +docs/prompt-template.md +docs/plans/ + +# Test output dirs +output/ +test_output/ diff --git a/cobol_testgen/__init__.py b/cobol_testgen/__init__.py index 89a6d38..cb1f2c4 100644 --- a/cobol_testgen/__init__.py +++ b/cobol_testgen/__init__.py @@ -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' diff --git a/cobol_testgen/core.py b/cobol_testgen/core.py index c489ac5..5678e6e 100644 --- a/cobol_testgen/core.py +++ b/cobol_testgen/core.py @@ -154,6 +154,7 @@ class _BrParser: self.assignments = assignments if assignments is not None else {} self.fields = fields self._goto_depth = goto_depth + self._cursors = {} def peek(self): if self.pos < len(self.lines): @@ -1311,6 +1312,64 @@ class _BrParser: self.assignments.setdefault(synthetic, []).append(info) return Assign(synthetic, info) + # 4a) DECLARE CURSOR ... FOR SELECT ... FROM ... + m = re.search( + r'DECLARE\s+(\w[\w-]*)\s+CURSOR\s+FOR\s+SELECT\s+(.*?)\s+FROM\s+(\w[\w-]*)\s*(.*)', + sql_text, re.IGNORECASE + ) + if m: + cursor_name = m.group(1).upper() + select_list = m.group(2).strip() + from_table = m.group(3).strip().upper() + remaining = m.group(4).strip() + + where_clause = '' + wm = self._RE_WHERE.search(remaining) + if wm: + where_clause = wm.group(1).strip() + + info = { + 'type': 'exec_sql_select', + 'cursor_name': cursor_name, + 'table': from_table, + 'select_list': select_list, + 'into_vars': [], + 'where': where_clause, + 'sql_text': sql_text, + } + synthetic = f'__SQL_CURSOR_{from_table}' + self.assignments.setdefault(synthetic, []).append(info) + self._cursors[cursor_name] = synthetic + return Assign(synthetic, info) + + # 4b) FETCH ... INTO ... + m = re.search(r'FETCH\s+(\w[\w-]*)\s+INTO\s+(.+)', sql_text, re.IGNORECASE) + if m: + cursor_name = m.group(1).upper() + into_raw = m.group(2).strip() + into_vars = [] + for v in re.split(r'\s*,\s*', into_raw): + v = v.strip().lstrip(':') + parts = v.split(':') + into_vars.append(parts[0].upper()) + + # Merge INTO vars into the corresponding DECLARE CURSOR entry + syn_key = self._cursors.get(cursor_name) + if syn_key: + for existing in self.assignments.get(syn_key, []): + if existing.get('cursor_name') == cursor_name: + existing['into_vars'] = into_vars + + info = { + 'type': 'exec_sql_fetch', + 'cursor_name': cursor_name, + 'into_vars': into_vars, + 'sql_text': sql_text, + } + for var in into_vars: + self.assignments.setdefault(var, []).append(info) + return Assign(into_vars[0], info) + # 4) UPDATE table SET ... WHERE ... m = self._RE_SQL_UPDATE.search(sql_text) if m: diff --git a/cobol_testgen/coverage.py b/cobol_testgen/coverage.py index 08ae4b6..9e92c41 100644 --- a/cobol_testgen/coverage.py +++ b/cobol_testgen/coverage.py @@ -373,6 +373,11 @@ def _mark_search(dp, cons, fields=None): if base_c == base_cond: branch_masks[i] = True break + # Also check reversed constraints (subject field references table element) + base_c2 = re.sub(r'\s*\(.*?\)\s*$', '', str(c[2])) + if base_c2 == base_cond: + branch_masks[i] = True + break else: leaves = list(collect_leaves(cond_tree)) assignment = {} @@ -458,15 +463,26 @@ def locate_decision_lines(decision_points, raw_source): for dp in decision_points: patterns = _build_search_patterns(dp) start = used_indices.get(dp.label, -1) + 1 + found = False for i in range(start, len(lines)): line = lines[i] for pat in patterns: if re.search(pat, line): dp.source_line = i + 1 used_indices[dp.label] = i + found = True break - if dp.source_line: + if found: break + # Multi-line fallback: compound conditions (AND/OR) may span >1 line + if not found and dp.kind == 'IF': + short_pat = _build_short_if_pattern(dp) + if short_pat: + for i in range(start, len(lines)): + if re.search(short_pat, lines[i]): + dp.source_line = i + 1 + used_indices[dp.label] = i + break def _normalize(text): @@ -501,6 +517,25 @@ def _build_search_patterns(dp): return patterns +def _build_short_if_pattern(dp): + """Build a pattern matching just the first clause of a compound IF condition.""" + if dp.kind != 'IF': + return None + cond = dp.label + if not cond: + return None + # Split on AND/OR to get first clause only + parts = re.split(r'\s+(AND|OR)\s+', cond, maxsplit=1, flags=re.IGNORECASE) + first_clause = parts[0].strip() + if first_clause == cond: + return None # Not a compound condition + norm = _normalize(first_clause) + esc = re.escape(norm) + esc = esc.replace(r'\ ', r'\s+') + esc = esc.replace(r'\'', r"['\"]") + return r'\bIF\b\s+' + esc + + # ── HTML 报告（详情页）── _DETAIL_HTML = ''' diff --git a/cobol_testgen/design.py b/cobol_testgen/design.py index 73f10f1..2958b44 100644 --- a/cobol_testgen/design.py +++ b/cobol_testgen/design.py @@ -1050,17 +1050,23 @@ def _dec_str(s, length): def _reconcile_unstring_fields(rec, left_field, operator, right_field, want_true, fields, left_chain, assignments, path_assign): - right_root, right_chain = trace_to_root(right_field, assignments, fields, path_assign) + right_root, right_chain = trace_to_root(right_field, assignments, fields, path_assign=None) if right_root not in rec: logger.debug(f"字段间比较协调：右侧根 {right_root} 不在 rec，跳过") return all_entries = (left_chain or []) + (right_chain or []) for _, asgn in all_entries: - if asgn.get('type') not in ('move', 'unstring_split'): + if asgn.get('type') not in ('move', 'unstring_split', 'move_literal'): logger.debug(f"字段间比较协调：链含非 MOVE 类型 {asgn.get('type')}，跳过") return left_val = str(rec.get(left_field, '')) if not left_val.strip(): + # 左字段无值时尝试反向：用右字段的值填充左字段 + right_root_val = str(rec.get(right_root, '')) + if right_root_val.strip() and operator in ('=', '==') and want_true: + rec[left_field] = right_root_val + logger.debug(f"字段间比较协调(反向)：{left_field}<={right_root}={right_root_val}") + return logger.debug(f"字段间比较协调：左侧 {left_field} 无值，跳过") return length = 0 @@ -1388,14 +1394,16 @@ def _enum_search_paths(node, fields): base = re.sub(r'\s*\(.*?\)\s*$', '', cond_tree.field) matching_val = cond_tree.value elem_key = f'{base}({i + 1})' + subj = cond_tree.value + subj_is_field = any(f['name'] == subj for f in fields) # 确保 match 值与字段 PIC 类型兼容 _fmt = next((f.get('pic_info', {}).get('type') for f in fields if f['name'] == elem_key), None) if _fmt in ('alphanumeric', 'alphabetic'): matching_val = str(matching_val).ljust( next((f['pic_info'].get('length', 1) for f in fields if f['name'] == elem_key and f.get('pic_info')), 1) )[:next((f['pic_info'].get('length', 1) for f in fields if f['name'] == elem_key and f.get('pic_info')), 1)] - if any(f['name'] == matching_val for f in fields): - extra_assign[elem_key] = [{'type': 'move', 'source_vars': [matching_val]}] + if subj_is_field: + extra_assign[elem_key] = [{'type': 'move', 'source_vars': [subj]}] else: extra_assign[elem_key] = [{'type': 'move_literal', 'literal': matching_val}] non_match = _non_match_for(cond_tree, fields) or ' ' @@ -1408,11 +1416,15 @@ def _enum_search_paths(node, fields): for k, v in sp_assign.items(): merged_assign.setdefault(k, []).extend(v if isinstance(v, list) else [v]) if cond_tree and isinstance(cond_tree, CondLeaf): - # Also set the subject field (right side of comparison) to match subj = cond_tree.value - if any(f['name'] == subj for f in fields): + subj_is_field = any(f['name'] == subj for f in fields) + if subj_is_field: + # Field-to-field: constrain subject to reference table element value + paths.append(([(subj, cond_tree.op, elem_key, True)] + sp_cons, merged_assign)) + else: + # Literal value: set subject field to literal, constrain table element merged_assign[subj] = [{'type': 'move_literal', 'literal': matching_val}] - paths.append(([(elem_key, cond_tree.op, matching_val.rstrip(), True)] + sp_cons, merged_assign)) + paths.append(([(elem_key, cond_tree.op, matching_val.rstrip(), True)] + sp_cons, merged_assign)) else: paths.append((sp_cons, merged_assign)) @@ -1436,32 +1448,76 @@ def _enum_search_paths(node, fields): return paths -def _rebuild_r01line_csv(rec, data_fields): - """直接基于 WRK-CSV 字段构建 CSV 字符串写入 rec['R01LINE']。 - 按 PIC 长度截断各字段，避免 _reconstruct_unstring_sources 污染导致字段过长的 bug。 +def _rebuild_r01line_csv(rec, data_fields, base_assignments=None): + """基于 UNSTRING 目标字段动态构建 CSV 字符串写入源字段（如 R01LINE）。 + 取代旧硬编码字段列表，支持任意程序的 UNSTRING 结构。 + 按各目标字段的实际 PIC 长度 padding，确保运行时 UNSTRING 正确解析。 """ - csv_fields = [ - ('WRK-CSV-APPL-ID', 8), ('WRK-CSV-EMP-ID', 8), ('WRK-CSV-APPL-DATE', 8), - ('WRK-CSV-START-TIME', 4), ('WRK-CSV-END-TIME', 4), ('WRK-CSV-STATUS', 1), - ('WRK-CSV-OVT-TYPE', 1), ('WRK-CSV-FILLER', 46), - ] - parts = [] - for fname, flen in csv_fields: - val = str(rec.get(fname, '')) - if len(val) > flen: - val = val[:flen] - elif len(val) < flen: - val = val.ljust(flen) - parts.append(val) - csv_value = ','.join(parts) - r01_len = 80 - for f in data_fields: - if f['name'] == 'R01LINE': - pi = f.get('pic_info', {}) - r01_len = pi.get('length', 80) or 80 - break - csv_value = csv_value.ljust(r01_len)[:r01_len] - rec['R01LINE'] = csv_value + if base_assignments is None: + return + + groups = {} + for tgt, asgn_list in base_assignments.items(): + for asgn in asgn_list: + if asgn.get('type') == 'unstring_split' and asgn.get('source_vars'): + src = asgn['source_vars'][0] + idx = asgn.get('index', 0) + groups.setdefault(src, []).append((idx, tgt)) + + if not groups: + return + + for src_var, targets in groups.items(): + targets.sort(key=lambda x: x[0]) + + resolved_src = src_var + for i, f in enumerate(data_fields): + if f['name'] == resolved_src: + grp_level = f.get('level', 0) + for f2 in data_fields[i + 1:]: + if f2.get('level', 0) <= grp_level or f2.get('level') == 77: + break + if f2.get('pic'): + resolved_src = f2['name'] + break + break + + if resolved_src not in rec: + continue + + csv_parts = [] + for idx, tgt in targets: + val = str(rec.get(tgt, '')) + length = 0 + for f in data_fields: + if f['name'] == tgt and f.get('pic_info'): + pi = f['pic_info'] + ftype = pi.get('type', '') + if ftype in ('alphanumeric', 'alphabetic'): + length = pi.get('length', 0) + elif ftype == 'numeric': + length = pi.get('digits', 0) + pi.get('decimal', 0) + break + if length > 0: + if len(val) > length: + val = val[:length] + elif len(val) < length: + val = val.ljust(length) + csv_parts.append(val) + + csv_value = ','.join(csv_parts) + + src_len = 0 + for f in data_fields: + if f['name'] == resolved_src: + pi = f.get('pic_info', {}) + if pi: + src_len = pi.get('length', 0) + break + if src_len > 0: + csv_value = csv_value.ljust(src_len)[:src_len] + + rec[resolved_src] = csv_value def generate_records(path_infos, data_fields, base_assignments=None, file_sec=None): @@ -1476,7 +1532,6 @@ def generate_records(path_infos, data_fields, base_assignments=None, file_sec=No records = [] kept_path_cons = [] term_types = [] - _zan01_emp_err_count = 0 if path_infos: for seq, (path_cons, path_assign, term_type) in enumerate(path_infos, start=1): path_cons = _filter_stop(path_cons) @@ -1595,29 +1650,54 @@ def generate_records(path_infos, data_fields, base_assignments=None, file_sec=No except (ValueError, TypeError): pass break - # Pass B.13: C01CHKRRC 约束与 SUB04CHK 输入字段同步 - # SUB04CHK 检查 C01CHKDAT(1:8) = SPACES → RC≠0。 - # 约束系统无法跨 CALL 追溯，需确保 WRK-CSV-EMP-ID/WRK-CSV-APPL-DATE - # 与预期的 C01CHKRRC 值一致，使运行时实际 CALL 返回正确结果。 - # want=False（C01CHKRRC=0，通过）→ EMP-ID 和 APPL-DATE 都有效 - # want=True（C01CHKRRC≠0，错误）→ 交替： - # 奇数个 → EMP-ID 空格 (#4-T) - # 偶数个 → EMP-ID有效 + DATE空格 (#5-T) - for c in path_cons: - if len(c) == 4 and c[0] == 'C01CHKRRC' and c[1] == '<>' and c[2] == 'ZERO': - if not c[3]: - if 'WRK-CSV-EMP-ID' in rec and str(rec.get('WRK-CSV-EMP-ID', '')).strip() == '': - rec['WRK-CSV-EMP-ID'] = '00000101' - if 'WRK-CSV-APPL-DATE' in rec and str(rec.get('WRK-CSV-APPL-DATE', '')).strip() == '': - rec['WRK-CSV-APPL-DATE'] = '20000101' - else: - _zan01_emp_err_count += 1 - if _zan01_emp_err_count % 2 == 0: - if 'WRK-CSV-EMP-ID' in rec: - rec['WRK-CSV-EMP-ID'] = '00000101' - if 'WRK-CSV-APPL-DATE' in rec: - rec['WRK-CSV-APPL-DATE'] = ' ' - break + # Pass B.13: C01CHKRRC 约束 → 按 IF 位置映射对应字段 + # SUB04CHK 校验 W01* 字段（非空格/非零为有效）。 + # C01CHKRRC 约束按 2030VALIDATESOR 的 IF 出现顺序对应： + # 0→EMP-ID, 1→START-DATE, 2→START-TIME, 3→END-DATE, 4→END-TIME + # BrSeq frozenset 去重导致只有 3 种 path type 保留： + # [True]、[False,True]、[False×5] + _c01_wants = [ + c[3] for c in path_cons + if len(c) == 4 and c[0] == 'C01CHKRRC' and c[1] == '<>' and c[2] == 'ZERO' + ] + if _c01_wants: + _c01_csv_fields = [ + 'WRK-CSV-EMP-ID', 'WRK-CSV-START-DATE', + 'WRK-CSV-START-TIME', 'WRK-CSV-END-DATE', 'WRK-CSV-END-TIME', + ] + _csv_to_w01 = {} + if base_assignments: + for tgt, alist in base_assignments.items(): + for a in alist: + if a.get('type') == 'move' and a.get('source_vars'): + _csv_to_w01.setdefault(a['source_vars'][0], tgt) + for idx, want in enumerate(_c01_wants): + if idx >= len(_c01_csv_fields): + break + csv_fld = _c01_csv_fields[idx] + w01_fld = _csv_to_w01.get(csv_fld) + for fname in (csv_fld, w01_fld): + if not fname or fname not in rec: + continue + if want: + pi = None + for f in data_fields: + if f['name'] == fname: + pi = f.get('pic_info', {}) + break + length = 8 + if pi: + length = pi.get('length', 8) or pi.get('digits', 8) + pi.get('decimal', 0) + rec[fname] = ' ' * length + else: + cur = str(rec.get(fname, '')).strip().rstrip('0').rstrip(' ') + if not cur or cur == '': + if 'TIME' in fname: + rec[fname] = '0900' + elif 'DATE' in fname: + rec[fname] = '20240115' + else: + rec[fname] = '00000101' # Pass B.8: UNSTRING source reconstruction (targets → source) if base_assignments: _reconstruct_unstring_sources(rec, base_assignments, data_fields) @@ -1638,7 +1718,7 @@ def generate_records(path_infos, data_fields, base_assignments=None, file_sec=No # 否则运行时 UNSTRING 会从 R01LINE 取有效值覆盖 WS 的无效值 # 直接基于 WRK-CSV 字段构建 CSV，避免 _reconstruct_unstring_sources 解析 # R01INNREC 时因组名在 rec 中而跳过子字段解析的 bug - _rebuild_r01line_csv(rec, data_fields) + _rebuild_r01line_csv(rec, data_fields, base_assignments) # Pass E: PIC 长度约束 — 模拟 COBOL 截断语义 for f in data_fields: @@ -1676,7 +1756,7 @@ def generate_records(path_infos, data_fields, base_assignments=None, file_sec=No rec2 = deepcopy(rec) rec2['WRK-CSV-EMP-ID'] = '00000101' rec2['WRK-CSV-APPL-DATE'] = ' ' - _rebuild_r01line_csv(rec2, data_fields) + _rebuild_r01line_csv(rec2, data_fields, base_assignments) records.append(rec2) kept_path_cons.append(path_cons) term_types.append(term_type) diff --git a/cobol_testgen/gcov.py b/cobol_testgen/gcov.py index 664466f..ed2f187 100644 --- a/cobol_testgen/gcov.py +++ b/cobol_testgen/gcov.py @@ -96,11 +96,18 @@ def _find_if_body_lines(source_lines: list[str], if_lineno_1: int): n = len(source_lines) for i in range(start, n): line = source_lines[i].upper().strip() - if re.match(r'\bIF\b', line) and not re.match(r'ELSE\s+IF', line, re.IGNORECASE): + if re.match(r'ELSE\s+IF', line, re.IGNORECASE): + if depth == 1: + else_start_0 = i + depth += 1 + continue + if re.match(r'\bIF\b', line): depth += 1 if re.match(r'END-IF', line): depth -= 1 - if depth == 0: + if '.' in line: + depth = 0 + if depth <= 0: end_if_0 = i break if depth == 1 and re.match(r'ELSE\b', line): @@ -172,12 +179,16 @@ def mark_from_gcov(decision_points: list, gcov_data: dict[int, int], dp.active_branches.add('F') elif dp.kind == 'EVALUATE': + for bn in dp.branch_names: + dp.active_branches.discard(bn) if count == 0: continue for bn in dp.branch_names: dp.active_branches.add(bn) elif dp.kind == 'PERFORM': + dp.active_branches.discard('Enter') + dp.active_branches.discard('Skip') if count > 1: dp.active_branches.add('Enter') dp.active_branches.add('Skip') diff --git a/cobol_testgen/pipeline_bridge.py b/cobol_testgen/pipeline_bridge.py index 83fe129..864eca2 100644 --- a/cobol_testgen/pipeline_bridge.py +++ b/cobol_testgen/pipeline_bridge.py @@ -180,5 +180,10 @@ def _assigns_list_to_dict(assigns_list: list) -> dict: tgt = a.get("tgt", "") src = a.get("src") or a.get("source_vars") if tgt and src: - result[tgt] = [a] + a_norm = dict(a) + if 'type' in a_norm: + a_norm['type'] = a_norm['type'].lower() + if 'src' in a_norm and not a_norm.get('source_vars'): + a_norm['source_vars'] = [a_norm['src']] + result[tgt] = [a_norm] return result diff --git a/cobol_testgen/runner.py b/cobol_testgen/runner.py index 15fadc1..7385d14 100644 --- a/cobol_testgen/runner.py +++ b/cobol_testgen/runner.py @@ -105,7 +105,8 @@ def _output_assign_names(select_info: dict, open_dir: dict, def compile_sub_modules(sub_dir: str, work_dir: str, - cpy_dir: str | None = None) -> list[str]: + cpy_dir: str | None = None, + log_dir: str | None = None) -> list[str]: sub_path = Path(sub_dir) work_path = Path(work_dir) work_path.mkdir(parents=True, exist_ok=True) @@ -131,9 +132,19 @@ def compile_sub_modules(sub_dir: str, work_dir: str, orig = os.getcwd() try: os.chdir(str(work_path)) - r = subprocess.run(cmd, capture_output=True, text=True, timeout=120) + r = subprocess.run(cmd, capture_output=True, text=True, timeout=120, errors='replace') finally: os.chdir(orig) + if log_dir: + log_path = Path(log_dir) / 'compile' / f"sub_{cbl.stem}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + f"COMMAND: {' '.join(cmd)}\n" + f"RETURNCODE: {r.returncode}\n\n" + f"STDOUT:\n{r.stdout}\n\n" + f"STDERR:\n{r.stderr}", + encoding='utf-8' + ) if r.returncode != 0: logger.warning(f" SUB编译失败 {cbl.name}: {r.stderr.strip()[:200]}") continue @@ -144,7 +155,8 @@ def compile_sub_modules(sub_dir: str, work_dir: str, def compile_program(program_name: str, source_dir: str, work_dir: str, sub_objects: list[str], - cpy_dir: str | None = None) -> str: + cpy_dir: str | None = None, + log_dir: str | None = None) -> str: work_path = Path(work_dir) work_path.mkdir(parents=True, exist_ok=True) @@ -164,9 +176,19 @@ def compile_program(program_name: str, source_dir: str, work_dir: str, orig = os.getcwd() try: os.chdir(str(work_path)) - r = subprocess.run(cmd, capture_output=True, text=True, timeout=120) + r = subprocess.run(cmd, capture_output=True, text=True, timeout=120, errors='replace') finally: os.chdir(orig) + if log_dir: + log_path = Path(log_dir) / 'compile' / f"{program_name}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + f"COMMAND: {' '.join(cmd)}\n" + f"RETURNCODE: {r.returncode}\n\n" + f"STDOUT:\n{r.stdout}\n\n" + f"STDERR:\n{r.stderr}", + encoding='utf-8' + ) if r.returncode != 0: raise RuntimeError(f"编译失败 {program_name}: {r.stderr.strip()[:500]}") return str(exe) @@ -302,7 +324,8 @@ def compare_outputs(actual: list[dict], expected: list[dict], # ── 单组执行（SOURCE 改 - Native）── -def run_group(group: GroupInfo, exe_path: str, temp_dir: str) -> GroupResult: +def run_group(group: GroupInfo, exe_path: str, temp_dir: str, + log_dir: str | None = None) -> GroupResult: logger.info(f" 执行组: {group.name} ({len(group.records)} 条记录)") exe = Path(exe_path).resolve() @@ -321,6 +344,17 @@ def run_group(group: GroupInfo, exe_path: str, temp_dir: str) -> GroupResult: finally: os.chdir(orig) + if log_dir: + log_path = Path(log_dir) / f"{group.name}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + f"COMMAND: {' '.join([str(exe)])}\n" + f"RETURNCODE: {result.returncode}\n\n" + f"STDOUT:\n{result.stdout}\n\n" + f"STDERR:\n{result.stderr}", + encoding='utf-8' + ) + rc = result.returncode all_pass = (rc == group.expected_returncode) all_details = [] @@ -362,7 +396,7 @@ def run_all(program_name: str, outdir: str, temp_dir: str, merged_gcov_data is None when no gcov runs. """ source_dir = source_dir or str(Path(outdir).parent) - work_dir = Path(temp_dir) + work_dir = Path(temp_dir).resolve() work_dir.mkdir(parents=True, exist_ok=True) expected = expected_records or [] path_infos = path_infos or [] @@ -370,6 +404,7 @@ def run_all(program_name: str, outdir: str, temp_dir: str, fd_field_dicts = _build_fd_field_dicts(fd_fields, fields_dict) assign_names = _input_assign_names(select_info, open_dir, fd_fields) + log_dir = os.path.join(outdir, 'logs') def _is_output_fd(fd_name: str) -> bool: dir_val = open_dir.get(fd_name, '') @@ -378,21 +413,21 @@ def run_all(program_name: str, outdir: str, temp_dir: str, # ── 1. SUB 编译（V3）── sub_dir = _resolve_sub_dir(source_dir) cpy_dir = _resolve_cpy_dir(source_dir) - sub_o = compile_sub_modules(sub_dir, str(work_dir), cpy_dir) + sub_o = compile_sub_modules(sub_dir, str(work_dir), cpy_dir, log_dir=log_dir) # ── 2. 主程序编译（V3）── exe_path = compile_program( - program_name, source_dir, str(work_dir), sub_o, cpy_dir + program_name, source_dir, str(work_dir), sub_o, cpy_dir, log_dir=log_dir ) # ── 3. 场景定义 ── scenes = [("main", records, term_types, expected, - Path(outdir) / 'input', Path(outdir) / 'output')] + Path(outdir) / 'main' / 'input', Path(outdir) / '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) / 'input_skip', Path(outdir) / 'run_skip' / 'output')) + Path(outdir) / 'skip' / 'input', Path(outdir) / 'skip' / 'output')) results = [] gcov_data_sets = [] @@ -428,7 +463,7 @@ def run_all(program_name: str, outdir: str, temp_dir: str, ) # ── 3d. 执行 ── - r = run_group(group, exe_path, str(work_dir)) + r = run_group(group, exe_path, str(work_dir), log_dir=log_dir) results.append(r) status = '✓' if r.passed else '✗' @@ -457,8 +492,7 @@ def run_all(program_name: str, outdir: str, temp_dir: str, for f in work_dir.glob("*.gcno"): if f.is_file() and f.stat().st_size > 0: dst = scene_gcov_dir / f.name - if not dst.exists(): - shutil.copy2(str(f), str(dst)) + shutil.copy2(str(f), str(dst)) # ── 3g. 收集该场景的 gcov 数据 ── from .gcov import run_gcov as _run_gcov @@ -506,6 +540,7 @@ def run_and_compare(program_name: str, outdir: str, temp_dir = os.path.join(outdir, '.run_cache') source_dir = os.path.join(outdir, '..', 'input') + log_dir = os.path.join(outdir, 'logs') work_dir = Path(temp_dir) work_dir.mkdir(parents=True, exist_ok=True) @@ -513,8 +548,8 @@ def run_and_compare(program_name: str, outdir: str, sub_dir = _resolve_sub_dir(source_dir) cpy_dir = _resolve_cpy_dir(source_dir) - sub_o = compile_sub_modules(sub_dir, temp_dir, cpy_dir) - exe_path = compile_program(program_name, source_dir, temp_dir, sub_o, cpy_dir) + sub_o = compile_sub_modules(sub_dir, temp_dir, cpy_dir, log_dir=log_dir) + exe_path = compile_program(program_name, source_dir, temp_dir, sub_o, cpy_dir, log_dir=log_dir) if normal_recs: group = GroupInfo( @@ -522,7 +557,7 @@ def run_and_compare(program_name: str, outdir: str, fd_field_dicts=fd_field_dicts, open_dir=open_dir, select_info=select_info, ) - r = run_group(group, exe_path, temp_dir) + r = run_group(group, exe_path, temp_dir, log_dir=log_dir) result['normal_pass'] = (r.returncode == 0) result['normal_returncode'] = r.returncode for fd_name in fd_field_dicts: @@ -545,7 +580,7 @@ def run_and_compare(program_name: str, outdir: str, fd_field_dicts=fd_field_dicts, open_dir=open_dir, select_info=select_info, ) - r = run_group(group, exe_path, temp_dir) + r = run_group(group, exe_path, temp_dir, log_dir=log_dir) if r.returncode != 0: result['abend_pass'] += 1 diff --git a/cobol_testgen/to_sql.py b/cobol_testgen/to_sql.py index 1680e40..82b95de 100644 --- a/cobol_testgen/to_sql.py +++ b/cobol_testgen/to_sql.py @@ -283,6 +283,8 @@ def collect_sql_meta(assignments: dict, declared_columns: dict, atype = asgn.get('type', '') if not atype.startswith('exec_sql_'): continue + if atype == 'exec_sql_fetch': + continue key = asgn.get('sql_text', '') if key in seen: continue @@ -323,6 +325,26 @@ def _infer_columns_from_where(where_cons: list) -> list[dict]: return list(seen.values()) +def _parse_select_columns(select_list: str) -> list[str]: + """Parse SELECT column list into individual column names. + Handles: 'COL1, COL2' → ['COL1', 'COL2'] + 'COL1, COL2 AS alias' → ['COL1', 'COL2'] + """ + if not select_list: + return [] + cols = [] + for part in select_list.split(','): + part = part.strip() + # Strip table alias prefix (T.COL → COL) + if '.' in part: + part = part.split('.')[1] if '.' in part else part + # Strip AS alias + m = re.search(r'^(\w[\w.-]*)', part) + if m: + cols.append(m.group(1).upper()) + return cols + + def build_db_input( branch_paths: list[tuple[list, dict]], fields_dict: list[dict], @@ -351,7 +373,9 @@ def build_db_input( for sql in sql_meta: atype = sql.get('type', '') - table = sql['table'] + table = sql.get('table', '') + if not table: + continue where_cons = sql.get('where_constraints', []) if table not in db_input: @@ -397,9 +421,18 @@ def build_db_input( if not col_infos: col_infos = _infer_columns_from_where(where_cons) into_vars = sql.get('into_vars', []) - for iv in into_vars: - if iv not in [c['name'] for c in col_infos]: - col_infos.append({'name': iv, 'db_type': 'CHAR', 'size': 20}) + + # Map INTO vars → actual SQL column names from SELECT clause + select_cols = _parse_select_columns(sql.get('select_list', '')) + into_to_col: dict[str, str] = {} + for i, iv in enumerate(into_vars): + if i < len(select_cols): + into_to_col[iv] = select_cols[i] + + # Use SQL column names (not INTO var names) for row keys + for sc in select_cols: + if sc not in [c['name'] for c in col_infos]: + col_infos.append({'name': sc, 'db_type': 'CHAR', 'size': 20}) for col_info in col_infos: col_name = col_info['name'] @@ -417,6 +450,19 @@ def build_db_input( if val is None and hv in rec: val = str(rec[hv]) + # Try to find value from INTO variable in the record + if val is None: + for iv, scola in into_to_col.items(): + if scola == col_name and iv in rec: + val = str(rec[iv]) + break + + # Try COBOL field name mapping + if val is None: + cobol_field = guess_cobol_field(col_name, table, declared_columns) + if cobol_field in rec: + val = str(rec[cobol_field]) + if val is not None: row[col_name] = _format_db_value(col_info, val) else: diff --git a/config/programs/ZAN06UPD.yaml b/config/programs/ZAN06UPD.yaml index d22c423..644b0fa 100644 --- a/config/programs/ZAN06UPD.yaml +++ b/config/programs/ZAN06UPD.yaml @@ -5,42 +5,63 @@ db_tables: - name: ZANTBL01 sql_name: OVT_APPLICATIONS columns: - - name: EMPNO - type: CHAR(6) - primary_key: true - - name: WORK_DATE + - name: APPL_ID type: CHAR(8) primary_key: true - - name: START_TIME - type: NUMERIC(4) - cobol_field: DB-START-TIME - - name: END_TIME - type: NUMERIC(4) - cobol_field: DB-END-TIME - - name: OVERTIME - type: NUMERIC(4) - cobol_field: DB-OVERTIME - - name: APPROVAL_FLAG + - name: EMP_ID + type: CHAR(8) + - name: APPL_DATE + type: CHAR(8) + - name: OVT_TYPE type: CHAR(1) - cobol_field: DB-APPROVAL-FLAG - - name: NOTES - type: VARCHAR(100) - cobol_field: DB-NOTES + - name: START_TIME + type: CHAR(4) + - name: END_TIME + type: CHAR(4) + - name: OVT_HOURS + type: NUMERIC(4,1) + - name: STATUS + type: CHAR(1) + - name: UPDATED_AT + type: TIMESTAMP - name: ZANTBL02 sql_name: OVT_MONTHLY columns: - - name: EMPNO + - name: EMP_ID + type: CHAR(8) + primary_key: true + - name: YEAR_MONTH type: CHAR(6) primary_key: true - - name: DEPTNO - type: CHAR(4) + - name: OVT_TYPE + type: CHAR(1) primary_key: true - - name: DEPT_NAME - type: VARCHAR(30) - cobol_field: DB-DEPT-NAME + - name: OVT_HOURS + type: NUMERIC(8,1) + - name: OVT_COUNT + type: NUMERIC(9) + - name: UPDATED_AT + type: TIMESTAMP subprograms: - SUB01DAT - SUB02MSG - SUB03END + - SUB04CHK + - SUB05TIM + +runs: + - id: normal + sysin: + period: "202607" + modes: ["NORMAL"] + - id: collision + sysin: + period: "202607" + modes: ["NORMAL"] + inject_duplicate_pk: true + - id: abnormal + sysin: + period: "202607" + modes: ["NORMAL"] diff --git a/orchestrator_db.py b/orchestrator_db.py index b21e942..8c68129 100644 --- a/orchestrator_db.py +++ b/orchestrator_db.py @@ -160,6 +160,10 @@ class GixsqlOrchestrator: copybook_dirs=[ascii_dir], extra_srcs=extra_srcs, ) + log_dir = self.runtime_dir / "logs" / "compile" + log_dir.mkdir(parents=True, exist_ok=True) + log_dir.joinpath(f"{self.program_id}.log").write_text( + result.log, encoding='utf-8') if result.success: self.exe_path = Path(result.exe_path) return DbPipelineResult( @@ -303,10 +307,70 @@ class GixsqlOrchestrator: if len(parts) == 2: rec['R01LINE'] = f"{dup_eid.ljust(8)},{parts[1]}" + # ── Coverage-driven data modifications (per-scenario) ── + # Normal scenario or legacy single-run: no modifications needed. + # Collision scenario: INSERT duplicate, OVT-MONTHLY match, COMMIT threshold. + # Abnormal scenario: orphan cancel ABEND (last, to avoid polluting other branches). + + apply_collision = scenario is not None and scenario.id == "collision" + apply_abnormal = scenario is not None and scenario.id == "abnormal" + + if apply_collision: + # #10 T: Ensure >= 50 R01 records for COMMIT threshold (CNS-COMMIT-CNT=50) + r01_recs = [r for r in recs if 'R01APPL-ID' in r] + r01_count = len(r01_recs) + if r01_count < 50: + template = r01_recs[-1].copy() if r01_recs else {} + needed = 50 - r01_count + for i in range(needed): + nr = {} + for key, val in template.items(): + if not key.startswith('R02'): + nr[key] = val + nr['R01APPL-ID'] = f"X50{str(i).zfill(5)}" + if 'R01EMP-ID' in nr: + nr['R01EMP-ID'] = str(int(str(nr.get('R01EMP-ID', '0') or '0')) + i + 10000).zfill(8) + recs.append(nr) + r01_recs = [r for r in recs if 'R01APPL-ID' in r] + logger.info(f" Coverage #10T: added {needed} R01-only records -> {len(r01_recs)} total") + + # #8 T: Two R01 records with same APPL-ID -> 2nd INSERT collides -> UPDATE + if len(r01_recs) >= 4: + dup_appl_id = 'COLISN01' + for idx in (2, 3): + r01_recs[idx]['R01APPL-ID'] = dup_appl_id + if 'R02APPL-ID' in r01_recs[idx]: + r01_recs[idx]['R02APPL-ID'] = dup_appl_id + logger.info(f" Coverage #8T: set APPL-ID={dup_appl_id} on records [2]&[3] for INSERT duplicate") + + # #11 T: Two R01 records with same (EMP-ID, APPL-DATE, OVT-TYPE) + if len(r01_recs) >= 2: + match_emp = r01_recs[1].get('R01EMP-ID', '00000000').strip() or '00000000' + match_date = r01_recs[1].get('R01APPL-DATE', '00000000').strip() or '00000000' + match_type = r01_recs[1].get('R01OVT-TYPE', '1').strip() or '1' + r01_recs[0]['R01EMP-ID'] = match_emp + r01_recs[0]['R01APPL-DATE'] = match_date + r01_recs[0]['R01OVT-TYPE'] = match_type + r01_recs[1]['R01EMP-ID'] = match_emp + r01_recs[1]['R01APPL-DATE'] = match_date + r01_recs[1]['R01OVT-TYPE'] = match_type + logger.info( + f" Coverage #11T: unified (EMP={match_emp} DATE={match_date}" + f" TYPE={match_type}) for R01 records [0]&[1]" + ) + + if apply_abnormal: + # #14 T: Last R02 record has non-existent APPL-ID -> orphan cancel ABEND + # NOTE: ABEND prevents 3000STPSOR (#19 T/F); covered by normal scenario. + r02_recs = [r for r in recs if 'R02APPL-ID' in r] + if r02_recs: + r02_recs[-1]['R02APPL-ID'] = 'ZZZZZZZZ' + logger.info(f" Coverage #14T: set last R02 APPL-ID='ZZZZZZZZ' for orphan cancel") + # 出力先ディレクトリ（シナリオ毎に分離） run_label = f"run_{scenario.id}" if scenario else "" output_root = self.work_dir / run_label if scenario else self.work_dir - input_dir = output_root / "input" + input_dir = output_root / "main" / "input" input_dir.mkdir(parents=True, exist_ok=True) # フラットファイル書き出し（全シナリオ同一） @@ -404,7 +468,7 @@ class GixsqlOrchestrator: sql_meta, declared_columns, records=recs) # Write main JSON（シナリオ毎に分離） - json_outdir = output_root / "json" + json_outdir = output_root / "main" / "json" json_outdir.mkdir(parents=True, exist_ok=True) json_path = json_outdir / f"{self.program_id}.json" output_json(json_records, json_path, roles, @@ -443,8 +507,8 @@ class GixsqlOrchestrator: # シナリオ毎の出力先 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 / "input" - output_dir = run_dir / "output" + input_dir = run_dir / "main" / "input" + output_dir = run_dir / "main" / "output" gcov_dir = self.runtime_dir / "gcov" input_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True) @@ -453,17 +517,17 @@ class GixsqlOrchestrator: # シナリオ毎の CWD = run_{id}/、単輪時は runtime_dir 直下 cwd = run_dir - # 入力ファイル（work_dir/run_{id}/input/ → runtime/run_{id}/input/） - gen_input_dir = self.work_dir / f"run_{scenario.id}" / "input" if scenario else self.work_dir / "input" + # 入力ファイル（work_dir/run_{id}/main/input/ → runtime/run_{id}/main/input/） + gen_input_dir = self.work_dir / f"run_{scenario.id}" / "main" / "input" if scenario else self.work_dir / "main" / "input" if gen_input_dir.exists(): for f in gen_input_dir.iterdir(): if f.is_file(): (input_dir / f.name).write_bytes(f.read_bytes()) - # JSON 出力（work_dir/run_{id}/json/ → runtime/run_{id}/json/） - gen_json_dir = self.work_dir / f"run_{scenario.id}" / "json" if scenario else self.work_dir / "json" + # JSON 出力（work_dir/run_{id}/main/json/ → runtime/run_{id}/main/json/） + gen_json_dir = self.work_dir / f"run_{scenario.id}" / "main" / "json" if scenario else self.work_dir / "main" / "json" if gen_json_dir.exists(): - json_dir = run_dir / "json" + json_dir = run_dir / "main" / "json" json_dir.mkdir(parents=True, exist_ok=True) for f in gen_json_dir.iterdir(): if f.is_file() and f.suffix.lower() == '.json': @@ -474,9 +538,9 @@ class GixsqlOrchestrator: env_overrides = {} for fname, direction in assign_map.items(): if direction == "INPUT": - env_overrides[fname] = os.path.join("input", fname) + env_overrides[fname] = os.path.join("main", "input", fname) else: - env_overrides[fname] = os.path.join("output", fname) + env_overrides[fname] = os.path.join("main", "output", fname) # シナリオ毎の DB パス db_path = self._current_db_path or self.db_path @@ -501,6 +565,11 @@ class GixsqlOrchestrator: env_overrides=env_overrides, ) + log_dir = self.runtime_dir / "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') + # .gcda を gcov/ にコピー（シナリオ毎に gcov/run_{id}/） # GnuCOBOL は .gcno が生成された CWD (= compile CWD = exe_dir) に .gcda を書き出す。 # 複数シナリオで .gcno は共有されるため COPY で行う（MOVE 不可）。 @@ -512,13 +581,22 @@ class GixsqlOrchestrator: gcda_src_dirs.append(self.runtime_dir) # 従来互換 gcda_dst_dir = gcov_dir / run_label if scenario else gcov_dir gcda_dst_dir.mkdir(parents=True, exist_ok=True) - for ext in (".gcda", ".gcno"): - for sd in gcda_src_dirs: - for f in sd.glob(f"*{ext}"): - if f.is_file() and f.stat().st_size > 0: - dst = gcda_dst_dir / f.name - if not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: + for sd in gcda_src_dirs: + for f in sd.glob("*.gcda"): + if f.is_file() and f.stat().st_size > 0: + dst = gcda_dst_dir / f.name + if not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: + try: shutil.copy2(str(f), str(dst)) + except PermissionError: + pass + for f in sd.glob("*.gcno"): + if f.is_file() and f.stat().st_size > 0: + dst = gcda_dst_dir / f.name + try: + shutil.copy2(str(f), str(dst)) + except PermissionError: + pass return DbPipelineResult( self.program_id, 3, result.success, @@ -600,12 +678,21 @@ class GixsqlOrchestrator: v3_root = Path(__file__).parent extra_search = list(gcov_dir.glob("run_*")) + [self.work_dir / "bin"] for search_dir in (v3_root, self.work_dir, self.runtime_dir, gcov_dir, Path.home(), *extra_search): - for ext in (".gcno", ".gcda"): - for f in search_dir.glob(f"*{ext}"): - if f.stat().st_size > 0: - dst = gcov_dir / f.name - if not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: + for f in search_dir.glob("*.gcda"): + if f.stat().st_size > 0: + dst = gcov_dir / f.name + if not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: + try: shutil.copy2(str(f), str(dst)) + except PermissionError: + pass + for f in search_dir.glob("*.gcno"): + if f.stat().st_size > 0: + dst = gcov_dir / f.name + try: + shutil.copy2(str(f), str(dst)) + except PermissionError: + pass # Count what we have gcno_gcda_count = 0 for ext in (".gcno", ".gcda"): @@ -915,26 +1002,37 @@ class GixsqlOrchestrator: select_to_file[sel_name] = fname assign_map[fname] = "UNKNOWN" - # Second pass: determine direction from OPEN statements - # Handle both simple (OPEN INPUT X) and compound (OPEN INPUT X OUTPUT Y) + # Second pass: determine direction from OPEN statements. + # COBOL allows multi-line OPEN where files listed without a direction + # keyword inherit the last stated direction: + # OPEN INPUT FILEA + # FILEB <-- inherits INPUT + # OUTPUT FILEC + # Strategy: extract OPEN body (up to terminating '.'), collapse + # whitespace, then parse direction→file pairs via splitting on + # direction keyword boundaries. for m in re.finditer( - r'OPEN\s+((?:INPUT|OUTPUT|I-O|EXTEND)\s+\w+)' - r'((?:\s+(?:INPUT|OUTPUT|I-O|EXTEND)\s+\w+)*)', - src_text, re.IGNORECASE + r'OPEN\s+(.+?)\.', src_text, re.IGNORECASE | re.DOTALL ): - # Parse the OPEN payload: "INPUT X" + " OUTPUT Y" - payload = m.group(1) + m.group(2) - for part in re.finditer( - r'(INPUT|OUTPUT|I-O|EXTEND)\s+(\w+)', payload, re.IGNORECASE - ): - direction = part.group(1).upper() - sel_name = part.group(2) - if sel_name in select_to_file: - fname = select_to_file[sel_name] - if direction in ("INPUT", "I-O"): - assign_map[fname] = "INPUT" - else: - assign_map[fname] = "OUTPUT" + full = re.sub(r'\s+', ' ', m.group(1)).strip() + # Split on direction keyword boundaries: "INPUT X Y OUTPUT Z" + # → ["INPUT X Y", "OUTPUT Z"] + tokens = re.split(r'\s+(?=(?:INPUT|OUTPUT|I-O|EXTEND)\s)', full, flags=re.IGNORECASE) + for seg in tokens: + seg = seg.strip() + if not seg: + continue + seg_m = re.match(r'(INPUT|OUTPUT|I-O|EXTEND)\s+([\w ]+)', seg, re.IGNORECASE) + if not seg_m: + continue + direction = seg_m.group(1).upper() + for fword in re.findall(r'\w+', seg_m.group(2)): + if fword in select_to_file: + fname = select_to_file[fword] + if direction in ("INPUT", "I-O"): + assign_map[fname] = "INPUT" + else: + assign_map[fname] = "OUTPUT" return assign_map @@ -1025,6 +1123,32 @@ class GixsqlOrchestrator: if not rows: logger.info(f" Table {table_name}: 0 initial rows (will be created at runtime)") continue + + # Debug + logger.info(f" Table {table_name}: {len(rows)} rows, cols={list(rows[0].keys()) if rows else []}") + + # Filter columns: only keep those that actually exist in the table + try: + pragma_cols = conn.execute( + f"PRAGMA table_info([{table_name}])" + ).fetchall() + valid_cols = {r[1].upper() for r in pragma_cols} + except Exception: + valid_cols = set() + + remapped_rows = [] + for row in rows: + new_row = {} + for k, v in row.items(): + if k.upper() in valid_cols: + new_row[k] = v + if new_row: + remapped_rows.append(new_row) + rows = remapped_rows + if not rows: + logger.info(f" Table {table_name}: all rows filtered out, skipping") + continue + col_names = list(rows[0].keys()) placeholders = ", ".join("?" for _ in col_names) quoted_cols = ", ".join(f"[{c}]" for c in col_names) diff --git a/runners/gixsql_runner.py b/runners/gixsql_runner.py index 5f0cb8a..d3aa5e4 100644 --- a/runners/gixsql_runner.py +++ b/runners/gixsql_runner.py @@ -290,10 +290,15 @@ class GixsqlCobolRunner: "-K", "GIXSQLExecSelectIntoOne", "-K", "GIXSQLEndSQL", "-K", "GIXSQLConnect", + "-K", "GIXSQLCursorDeclare", + "-K", "GIXSQLCursorDeclareParams", + "-K", "GIXSQLCursorFetchOne", + "-K", "GIXSQLCursorOpen", + "-K", "GIXSQLCursorClose", ] cmd = [ self.cobc_cmd, "-x", - "-L", str(self.lib_path), + "-L", str(self.lib_path.resolve()), *gixsql_k, "-l", "gixsql", ] diff --git a/test-data/s30_db_e2e.py b/test-data/s30_db_e2e.py new file mode 100644 index 0000000..46b1103 --- /dev/null +++ b/test-data/s30_db_e2e.py @@ -0,0 +1,67 @@ +"""S30: DB プログラム E2E — ZAN06UPD 6Step 実行""" +import sys, os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +P=0;F=0 +def ck(v,m=""): global P,F; (P:=P+1) if v else (F:=F+1, print(f" FAIL {m}")) +def sec(n): print(f"\n{'='*60}\n{n}\n{'='*60}") + +DESKTOP = os.path.join(os.environ['USERPROFILE'], "Desktop") +ROOT = os.path.join(DESKTOP, "2026技术大赛", "cobol-tna-system") + "/" +COPYBOOKS = [os.path.join(ROOT, "cpy"), os.path.join(ROOT, "src")] + +from config import Config +from orchestrator_db import GixsqlOrchestrator + +sec("Step 0: Init") +PROJECT = os.path.join(os.path.dirname(__file__), '..') +cfg = Config() +cfg.gixsql_path = os.path.join(PROJECT, "gixsql", "bin", "gixpp.exe") +cfg.gixsql_lib_path = os.path.join(PROJECT, "gixsql", "lib") + +orch = GixsqlOrchestrator( + cfg, "ZAN06UPD", + cobol_src_dir=os.path.join(ROOT, "src"), + copybook_dirs=COPYBOOKS, +) +ck(orch.program_id == "ZAN06UPD", "orchestrator init") +print(f" Tables: {[t.name for t in orch.schema.db_tables]}") +print(f" SUBs: {orch.schema.subprograms}") + +sec("Step 1: gixpp -> compile") +r1 = orch.step1_setup_environment() +ck(r1.success, f"step1: {r1.message}") +if r1.success: + ck(orch.exe_path and os.path.exists(orch.exe_path), "exe created") + +sec("Step 2: generate inputs") +r2 = orch.step2_generate_inputs() +ck(r2.success, f"step2: {r2.message}") + +sec("Step 3: run COBOL") +r3 = orch.step3_run_cobol() +ck(r3.success, f"step3: {r3.message}") + +sec("Coverage report") +cr = orch.generate_coverage_report() +ck(cr.success, f"coverage: {cr.message}") +if cr.success: + cov = cr.data.get("coverage", "?") + rep = cr.data.get("reports", "?") + print(f" Coverage: {cov}") + print(f" Reports: {rep}") + +sec("Step 4: extract intermediate data") +r4 = orch.step4_extract_intermediate() +ck(r4.success, f"step4: {r4.message}") + +sec("Step 5: run Java (placeholder)") +r5 = orch.step5_run_java() + +sec("Step 6: verify") +vr = orch.step6_verify() +print(f" Status: {vr.status}") +print(f" Step reached: {vr.step_reached}") + +sec("SUMMARY") +print(f"S30: {P} PASS / {F} FAIL") +if F > 0: sys.exit(1)