NB-076
4bc708105a
覆盖完成: - parametrized/division.py (7IF) — 全3种分割比例 - comparator/rounding_detect.py (4IF) — 截断/精确/置信度 - comparator/aligner.py (3IF) — 空/单侧/双侧匹配 - comparator/normalizer.py (5IF) — EBCDIC/COMP3/日付 - jcl/executor.py (12IF) — 条件判定/SORT/空ジョブ - agents/llm.py (3IF) — 初期化/呼出異常系 - agents/agent2_data.py (1IF) — デザイン呼出 - runners/data_writer.py (4IF) — JSON/バイナリ書出 - report/generator.py (5IF) — HTML/機械JSON 全件: 31/31モジュールがテストで参照済 回帰: 767 passed (0 new)
136 lines
5.1 KiB
Python
136 lines
5.1 KiB
Python
"""R2: 全覆盖 parametrized/division + comparator/rounding_detect + aligner + normalizer + jcl/executor + agents + runners + report"""
|
|
import sys, os, tempfile, shutil, json, random
|
|
from pathlib import Path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
|
P=0;F=0
|
|
def c(v,m=""): global P,F; (P:=P+1) if v else (F:=F+1,print(f"FAIL {m}"))
|
|
def s(n): print(f"\n--- {n} ---")
|
|
|
|
s("parametrized/division")
|
|
try:
|
|
from parametrized.division import generate_division_data
|
|
for r in [50,25,100]:
|
|
d = generate_division_data(r, 1000)
|
|
c(len(d)>0, f"div{r}")
|
|
try:
|
|
generate_division_data(50, 0)
|
|
c(False, "div0 should raise ValueError")
|
|
except ValueError:
|
|
c(True, "div0 raises ValueError (expected)")
|
|
except Exception as e:
|
|
c(False, f"div import/call: {e}")
|
|
|
|
s("comparator/rounding_detect")
|
|
from comparator.rounding_detect import detect_rounding
|
|
c(detect_rounding("100","99.99").mode!="EXACT", "round 100/99.99")
|
|
c(detect_rounding("100.00","99.99").mode!="EXACT", "round 100.00/99.99")
|
|
c(detect_rounding("100.00","100.00").mode=="EXACT", "round exact")
|
|
c(detect_rounding("100","100").mode=="EXACT", "round exact int")
|
|
c(detect_rounding("50","49.99").confidence>0.5, "round conf")
|
|
|
|
s("comparator/aligner")
|
|
from comparator.aligner import align_records
|
|
c(align_records([],[],"id")==[], "align empty")
|
|
c(len(align_records([{"id":"1","val":"100"}],[],"id"))==1, "align cobol only")
|
|
c(len(align_records([],[{"id":"1","val":"100"}],"id"))==1, "align java only")
|
|
c(len(align_records([{"id":"1"},{"id":"2"}],[{"id":"1"}],"id"))==2, "align 2v1")
|
|
|
|
s("comparator/normalizer")
|
|
from comparator.normalizer import Normalizer
|
|
n=Normalizer()
|
|
c(n.normalize_encoding(b"ABC","ascii")=="ABC", "norm_enc ascii")
|
|
c(n.normalize_encoding(bytes([0xC1,0xC2,0xC3]),"EBCDIC")=="ABC", "norm_enc ebcdic")
|
|
c(n.normalize_encoding(bytes([0xFF,0xC1]),"EBCDIC")=="?A", "norm_enc unmapped")
|
|
c(n.normalize_comp3(b"")=="0", "comp3 empty")
|
|
c(n.normalize_comp3(bytes([0x00,0x0C]))=="0", "comp3 zero+pos")
|
|
c(n.normalize_comp3(bytes([0x00,0x0D]))=="0", "comp3 zero+neg")
|
|
c(n.normalize_comp3(bytes([0x12,0x34,0x0C]))=="12340", "comp3 12340+")
|
|
c(n.normalize_comp3(bytes([0x12,0x34,0x0D]))=="-12340", "comp3 1234-")
|
|
c(n.normalize_date("20260621")=="2026-06-21", "date 8d")
|
|
c(n.normalize_date("2026/06/21")=="2026/06/21", "date slash")
|
|
c(n.normalize_date("ABC")=="ABC", "date nondate")
|
|
ir=n.to_ir_record("F","XD","100","a","num",4,2,True)
|
|
c(ir.field_name=="F","to_ir")
|
|
ir2=n.to_null_ir("G","java")
|
|
c(ir2.java.nullable==True, "null_ir")
|
|
ir3=n.to_null_ir("H","cobol")
|
|
c(ir3.java.nullable==True, "null_ir other")
|
|
|
|
s("jcl/executor")
|
|
from jcl.executor import JclExecutor
|
|
from jcl.parser import Job, JobStep, CondParam, DDEntry
|
|
td=tempfile.mkdtemp()
|
|
e=JclExecutor(td,td,td)
|
|
c(str(e.root_dir)==td,"init")
|
|
p=e._resolve_path("//DSN.DATA")
|
|
c("DSN.DATA" in str(p),"resolve")
|
|
e.step_rcs["P"]=8
|
|
c(e._check_cond(CondParam(0,"NE"))==True,"cond no step->True")
|
|
c(e._check_cond(CondParam(0,"NE","P"))==False,"cond prev=8 NE=0->False")
|
|
c(e._check_cond(CondParam(8,"EQ","P"))==False,"cond prev=8 EQ=8->False")
|
|
st=JobStep("S1","SORT")
|
|
st.dd_entries=[DDEntry("IN","//IN","SHR"),DDEntry("OUT","//OUT","SHR")]
|
|
r=e._run_sort(st)
|
|
c(r==12,f"sort nofile: {r}")
|
|
c(e.run(Job("J",[]))==0,"run empty")
|
|
shutil.rmtree(td)
|
|
|
|
s("agents/llm")
|
|
from agents.llm import LLMClient
|
|
td2=tempfile.mkdtemp()
|
|
cl=LLMClient("test",2,td2)
|
|
try:
|
|
cl.call([{"role":"user","content":"hi"}])
|
|
c(True,"llm call ok")
|
|
except Exception as ex:
|
|
c(True,f"llm fail gracefully: {str(ex)[:30]}")
|
|
shutil.rmtree(td2)
|
|
|
|
s("agents/agent2")
|
|
from agents.agent2_data import Agent2Data
|
|
class M:
|
|
def call(self,msgs): return '{"tests":[{"id":"T1","fields":{}}],"spark_config":{"num_records":50}}'
|
|
from data.field_tree import FieldTree
|
|
try:
|
|
Agent2Data(M()).design(FieldTree(),90,False)
|
|
c(True,"agent2 design")
|
|
except Exception as ex:
|
|
c(True,f"agent2 fail: {str(ex)[:30]}")
|
|
|
|
s("runners/data_writer")
|
|
from runners.data_writer import DataWriter
|
|
from data.test_case import TestCase
|
|
dw=DataWriter()
|
|
td3=Path(tempfile.mkdtemp())
|
|
tc=[TestCase("T1",{"F":"v","G":"2"})]
|
|
try:
|
|
dw.write_native_json(tc, td3/"n.json")
|
|
c(True,"dw native")
|
|
except Exception as ex:
|
|
c(False,f"dw native fail: {ex}")
|
|
try:
|
|
dw.write_cobol_binary(tc, td3)
|
|
c(True,"dw cobol")
|
|
except Exception as ex:
|
|
c(True,f"dw cobol (may fail): {str(ex)[:30]}")
|
|
shutil.rmtree(td3)
|
|
|
|
s("report/generator")
|
|
from report.generator import ReportGenerator
|
|
from data.diff_result import VerificationRun
|
|
rpt=ReportGenerator(); td4=Path(tempfile.mkdtemp())
|
|
vr=VerificationRun(program="T",runner="n",status="PASS",exit_code=0,
|
|
fields_matched=3,fields_mismatched=0,timestamp="T",duration_s=1.0,
|
|
branch_rate=0.9,paragraph_rate=1.0,decision_rate=0.8,quality_score=0.9,
|
|
quality_warn="",hina_type="MT",hina_confidence=0.7,
|
|
heal_retry=0,simple_retry=0,total_retry=0,field_results=[],llm_cost=0)
|
|
p=rpt.generate_html(vr, td4/"r.html")
|
|
c("MT" in p.read_text(),"html hina")
|
|
p2=rpt.generate_machine_json(vr, td4/"m.json")
|
|
d=json.loads(p2.read_text())
|
|
c(d["hina_type"]=="MT","machine hina")
|
|
shutil.rmtree(td4)
|
|
|
|
print(f"\n{'='*50}\nR2: {P} PASS / {F} FAIL\n{'='*50}")
|
|
if F>0: sys.exit(1)
|