feat(blackbox): add black-box JSON to internal record mapping
This commit is contained in:
@@ -0,0 +1,138 @@
|
||||
"""黑盒（black-box-data-create）测试数据接入：JSON → cobol_testgen 内部 record。
|
||||
|
||||
黑盒 JSON 结构（每组一个文件）：
|
||||
{ "program": "ZAN04MAT",
|
||||
"records": [ { "input": { "<FD>": { "<字段>": "<值>" } } } ] }
|
||||
|
||||
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
|
||||
@@ -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 == []
|
||||
Reference in New Issue
Block a user