From 12d07797b7f156cb661e90fb7737f18a35e03955 Mon Sep 17 00:00:00 2001 From: zhang_taoming <854317252@qq.com> Date: Sun, 13 Sep 2026 14:55:38 +0800 Subject: [PATCH] feat(blackbox): add black-box JSON to internal record mapping --- cobol_testgen/blackbox.py | 138 +++++++++++++++++++++++++++ tests/cobol_testgen/test_blackbox.py | 77 +++++++++++++++ 2 files changed, 215 insertions(+) create mode 100644 cobol_testgen/blackbox.py create mode 100644 tests/cobol_testgen/test_blackbox.py diff --git a/cobol_testgen/blackbox.py b/cobol_testgen/blackbox.py new file mode 100644 index 0000000..22d0cb1 --- /dev/null +++ b/cobol_testgen/blackbox.py @@ -0,0 +1,138 @@ +"""黑盒（black-box-data-create）测试数据接入：JSON → cobol_testgen 内部 record。 + +黑盒 JSON 结构（每组一个文件）： + { "program": "ZAN04MAT", + "records": [ { "input": { "": { "<字段>": "<值>" } } } ] } + +FD 键与字段名由 LLM 生成，存在多种写法（FD名 / DD名 / 识别子，字段带或不带 '-'）， +因此这里做归一化容错匹配。 +""" + +import json +import logging +from dataclasses import dataclass, field +from pathlib import Path + +logger = logging.getLogger(__name__) + + +@dataclass +class BlackBoxGroup: + label: str + records: list[dict] + term_types: list[str] = field(default_factory=list) + source_path: str = '' + + +def norm_name(name) -> str: + """字段/FD 名归一化：大写 + 去除 '-' '_' 分隔符。""" + return str(name).upper().replace('-', '').replace('_', '') + + +def _fd_aliases(fd_name: str, select_info: dict) -> set: + aliases = {norm_name(fd_name)} + sel = select_info.get(fd_name, {}) + if isinstance(sel, dict): + assign = sel.get('assign') + if assign: + aliases.add(norm_name(assign)) + return aliases + + +def _resolve_fd(bb_key, fd_aliases: dict): + """bb_key → 内部 fd_name；先精确别名，再前缀容错（R01 → R01INNFIL）。""" + nk = norm_name(bb_key) + for fd_name, aliases in fd_aliases.items(): + if nk in aliases: + return fd_name + for fd_name, aliases in fd_aliases.items(): + for a in aliases: + if a and (nk.startswith(a) or a.startswith(nk)): + return fd_name + return None + + +def _field_alias_map(data_fields: list[dict]) -> dict: + m = {} + for f in data_fields or []: + if f.get('is_88'): + continue + m.setdefault(norm_name(f['name']), f['name']) + return m + + +def _iter_group_files(path: Path) -> list[Path]: + p = Path(path) + if p.is_file() and p.suffix.lower() == '.json': + return [p] + if not p.is_dir(): + return [] + files = sorted(p.glob('*_g*.json')) + if files: + return files + bb = p / 'black_box' + if bb.is_dir(): + return sorted(bb.rglob('*_g*.json')) + return [] + + +def _group_label(json_path: Path) -> str: + for token in json_path.stem.split('_'): + if token.lower().startswith('g') and token[1:].isdigit(): + return token + return json_path.stem + + +def _map_record(raw, fd_aliases, field_map, jf: Path) -> dict: + inp = raw.get('input') + if not isinstance(inp, dict): + return {} + rec = {} + for bb_fd, fields in inp.items(): + fd_name = _resolve_fd(bb_fd, fd_aliases) + if fd_name is None: + logger.debug(f"黑盒 FD 未匹配: {bb_fd} ({jf.name})") + continue + if not isinstance(fields, dict): + continue + for bb_field, val in fields.items(): + internal = field_map.get(norm_name(bb_field)) + if internal is None: + logger.debug(f"黑盒字段未匹配: {bb_field} ({jf.name})") + continue + rec[internal] = '' if val is None else str(val) + return rec + + +def load_black_box_groups(path, fd_fields: dict, select_info: dict, + data_fields: list[dict]) -> list[BlackBoxGroup]: + """读取黑盒 JSON，返回分组列表。 + + path 可以是：程序目录（自动找 black_box/）、black_box 目录、或单个 JSON 文件。 + """ + fd_aliases = {fd: _fd_aliases(fd, select_info) for fd in fd_fields} + field_map = _field_alias_map(data_fields) + groups = [] + for jf in _iter_group_files(Path(path)): + try: + obj = json.loads(jf.read_text(encoding='utf-8')) + except Exception as e: # noqa: BLE001 + logger.warning(f"黑盒 JSON 解析失败 {jf}: {e}") + continue + label = _group_label(jf) + records = [] + for raw in obj.get('records', []) or []: + rec = _map_record(raw, fd_aliases, field_map, jf) + if rec: + records.append(rec) + if not records: + logger.warning(f"黑盒组 {label} 无有效记录，跳过 ({jf.name})") + continue + groups.append(BlackBoxGroup( + label=label, + records=records, + term_types=['normal'] * len(records), + source_path=str(jf), + )) + logger.info(f" 黑盒组 {label}: {len(records)} 条记录 ({jf.name})") + return groups diff --git a/tests/cobol_testgen/test_blackbox.py b/tests/cobol_testgen/test_blackbox.py new file mode 100644 index 0000000..c54d852 --- /dev/null +++ b/tests/cobol_testgen/test_blackbox.py @@ -0,0 +1,77 @@ +"""黑盒 JSON → cobol_testgen 内部 record 映射测试""" + +import json +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from cobol_testgen.blackbox import norm_name, load_black_box_groups + + +def _write_group(tmp, name, payload): + d = Path(tmp) / "black_box" / "g1" + d.mkdir(parents=True, exist_ok=True) + p = d / f"{name}.json" + p.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + return p + + +def test_norm_name(): + assert norm_name("R01-APPL-ID") == "R01APPLID" + assert norm_name("R01APPL-ID") == "R01APPLID" + assert norm_name("r01_appl_id") == "R01APPLID" + + +def test_load_groups_fd_assign_and_field_hyphen(): + fd_fields = {"R01INNFIL": ["R01APPL-ID", "R01EMP-ID"]} + select_info = {"R01INNFIL": {"assign": "ZAN04R01"}} + data_fields = [ + {"name": "R01APPL-ID", "pic": "X(8)"}, + {"name": "R01EMP-ID", "pic": "9(8)"}, + ] + with tempfile.TemporaryDirectory() as tmp: + _write_group(tmp, "ZAN04MAT_g1", { + "program": "ZAN04MAT", + "records": [{ + "input": {"ZAN04R01": { + "R01-APPL-ID": "A0000001", + "R01-EMP-ID": "00000001", + }} + }], + }) + groups = load_black_box_groups(tmp, fd_fields, select_info, data_fields) + assert len(groups) == 1 + g = groups[0] + assert g.label == "g1" + assert g.records[0]["R01APPL-ID"] == "A0000001" + assert g.records[0]["R01EMP-ID"] == "00000001" + assert g.term_types == ["normal"] + + +def test_load_groups_fd_name_variant(): + fd_fields = {"R01INNFIL": ["R01APPL-ID"]} + select_info = {"R01INNFIL": {"assign": "ZAN04R01"}} + data_fields = [{"name": "R01APPL-ID", "pic": "X(8)"}] + with tempfile.TemporaryDirectory() as tmp: + _write_group(tmp, "ZAN04MAT_g1", { + "program": "ZAN04MAT", + "records": [{"input": {"R01INNFIL": {"R01-APPL-ID": "B0000002"}}}], + }) + groups = load_black_box_groups(tmp, fd_fields, select_info, data_fields) + assert groups[0].records[0]["R01APPL-ID"] == "B0000002" + + +def test_load_groups_skips_unmatched_and_empty(): + fd_fields = {"R01INNFIL": ["R01APPL-ID"]} + select_info = {"R01INNFIL": {"assign": "ZAN04R01"}} + data_fields = [{"name": "R01APPL-ID", "pic": "X(8)"}] + with tempfile.TemporaryDirectory() as tmp: + _write_group(tmp, "ZAN04MAT_g1", { + "program": "ZAN04MAT", + "records": [{"input": {"UNKNOWN-FD": {"X": "1"}}}], + }) + groups = load_black_box_groups(tmp, fd_fields, select_info, data_fields) + assert groups == []