cobol-java-v3/test-data/s20_runtime_verification.py

"""S20: Runtime branch coverage verification via DISPLAY instrumentation

For each benchmark program:
1. Parse with our system → get expected decision points
2. Inject DISPLAY markers at each IF/ELSE/WHEN/AT_END branch in the COBOL source
3. Generate test data using our pipeline → write flat files
4. Compile INSTRUMENTED program with GnuCOBOL
5. Run it → capture stdout (DISPLAY lines = which branches were hit)
6. Compare: expected hits vs actual hits

If our parser says "200 decision points" but runtime only shows 150 hits,
we KNOW there's a gap — no way to fake this.
"""
import sys, os, re, subprocess, shutil, tempfile
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}")

ROOT = "D:/cobol-java/cobol-test-programs/"
COPYBOOKS = os.path.join(ROOT, "common", "copybooks")
COBC = "cobc"

from cobol_testgen import extract_structure, generate_data
from cobol_testgen.read import preprocess, resolve_copybooks, extract_data_division, extract_procedure_division, parse_data_division
from cobol_testgen.design_mcdc import enum_paths
from cobol_testgen.pipeline_bridge import build_branch_tree_fallback
from cobol_testgen.flatfile import write_all_files, analyze_fd_layout


def find_main_file(directory):
    cbls = [f for f in os.listdir(directory) if f.endswith('.cbl')]
    ws = [f for f in cbls if re.match(r'main-\d{2}-', f, re.IGNORECASE)]
    if ws:
        return max(ws, key=lambda f: os.path.getsize(os.path.join(directory, f)))
    return max(cbls, key=lambda f: os.path.getsize(os.path.join(directory, f))) if cbls else None


def instrument_source(source: str) -> tuple[str, list[dict]]:
    """Insert DISPLAY markers at each branch point.

    Returns (instrumented_source, list_of_marker_info).
    Each marker: {"id": int, "line": int, "kind": str, "label": str}
    """
    markers = []
    marker_id = [0]
    lines = source.split('\n')
    result = []
    in_pd = False

    for i, raw in enumerate(lines):
        line = raw
        upper = line.upper()

        # Detect PROCEDURE DIVISION (use search, not match — fixed format)
        if re.search(r'PROCEDURE\s+DIVISION', line, re.IGNORECASE):
            in_pd = True

        if not in_pd:
            result.append(line)
            continue

        # DISPLAY injection at decision points
        # IF line (not ELSE IF, not END-IF)
        if re.match(r'^\s*IF\b', upper) and not re.match(r'^\s*IF\s+\w+\s*>=\s*0', upper):
            marker_id[0] += 1
            mid = marker_id[0]
            markers.append({"id": mid, "line": i + 1, "kind": "IF"})
            # Insert DISPLAY before the IF's DOT or at end of line
            indent = line[:len(line) - len(line.lstrip())]
            # Find condition text for marker
            cond_match = re.match(r'^\s*IF\b\s*(.*)', line, re.IGNORECASE)
            cond = cond_match.group(1).strip()[:30] if cond_match else "?"
            display_line = f'{indent}           DISPLAY "BRANCH-MARKER:IF:{mid}:{cond}"'
            result.append(display_line)
            result.append(line)
            continue

        # ELSE (not ELSE IF)
        if re.match(r'^\s*ELSE\b', upper) and not re.match(r'^\s*ELSE\s+IF\b', upper):
            marker_id[0] += 1
            mid = marker_id[0]
            indent = line[:len(line) - len(line.lstrip())]
            markers.append({"id": mid, "line": i + 1, "kind": "ELSE"})
            display_line = f'{indent}           DISPLAY "BRANCH-MARKER:ELSE:{mid}"'
            result.append(display_line)
            result.append(line)
            continue

        # AT END
        if re.match(r'AT\s+END', line, re.IGNORECASE):
            marker_id[0] += 1
            mid = marker_id[0]
            markers.append({"id": mid, "line": i + 1, "kind": "AT_END"})
            indent = line[:len(line) - len(line.lstrip())]
            display_line = f'{indent}           DISPLAY "BRANCH-MARKER:AT_END:{mid}"'
            result.append(display_line)
            result.append(line)
            continue

        # NOT AT END
        if re.match(r'NOT\s+AT\s+END', line, re.IGNORECASE):
            marker_id[0] += 1
            mid = marker_id[0]
            markers.append({"id": mid, "line": i + 1, "kind": "NOT_AT_END"})
            indent = line[:len(line) - len(line.lstrip())]
            display_line = f'{indent}           DISPLAY "BRANCH-MARKER:NOT_AT_END:{mid}"'
            result.append(display_line)
            result.append(line)
            continue

        # WHEN (not WHEN OTHER)
        if re.match(r'WHEN\s+', line, re.IGNORECASE) and not re.match(r'WHEN\s+OTHER', line, re.IGNORECASE):
            marker_id[0] += 1
            mid = marker_id[0]
            markers.append({"id": mid, "line": i + 1, "kind": "WHEN"})
            indent = line[:len(line) - len(line.lstrip())]
            display_line = f'{indent}           DISPLAY "BRANCH-MARKER:WHEN:{mid}"'
            result.append(display_line)
            result.append(line)
            continue

        # WHEN OTHER
        if re.match(r'WHEN\s+OTHER', line, re.IGNORECASE):
            marker_id[0] += 1
            mid = marker_id[0]
            markers.append({"id": mid, "line": i + 1, "kind": "WHEN_OTHER"})
            indent = line[:len(line) - len(line.lstrip())]
            display_line = f'{indent}           DISPLAY "BRANCH-MARKER:WHEN_OTHER:{mid}"'
            result.append(display_line)
            result.append(line)
            continue

        result.append(line)

    return '\n'.join(result), markers


def count_unique_branch_hits(stdout: str) -> set[int]:
    """Extract unique BRANCH-MARKER IDs from stdout."""
    hits = set()
    for m in re.finditer(r'BRANCH-MARKER:([^:]+):(\d+)', stdout):
        mid = int(m.group(2))
        hits.add(mid)
    return hits


# ──────────────────────────────────────
# MAIN TEST
# ──────────────────────────────────────

# Pick 3 programs: matching (simple), sort (SORT), csv (complex logic)
test_programs = [
    ("01-matching-1-1", "01-matching-1-1", "Simple matching prog"),
    ("34-sort", "34-sort", "SORT with many branches"),
    ("28-sysin", "28-sysin", "SYSIN param dispatch, 200 branches"),
]

for dirname, expected_name, desc in test_programs:
    sec(f"Verifying {dirname}: {desc}")
    dp = os.path.join(ROOT, dirname)
    fname = find_main_file(dp)
    if not fname:
        ck(False, f"Can't find main file in {dp}")
        continue
    fpath = os.path.join(dp, fname)
    src = open(fpath, encoding='utf-8').read()

    # ── 1. Our static analysis ──
    print(f"\n[1/6] Static analysis...")
    st = extract_structure(src)
    static_branches = st.get('total_branches', 0)
    print(f"  Our parser finds: {static_branches} branches")

    # ── 2. Generate test data ──
    print(f"\n[2/6] Generating test data...")
    pp = resolve_copybooks(src, dp, extra_search_paths=[COPYBOOKS])
    pp_str = preprocess(pp)
    recs = generate_data(pp_str, st)
    print(f"  Generated {len(recs)} test records")

    # ── 3. Write flat files in temp directory ──
    print(f"\n[3/6] Writing flat files...")
    workdir = os.path.join(dp, f".tmp-runtime-{dirname}")
    if os.path.exists(workdir):
        shutil.rmtree(workdir)
    os.makedirs(workdir, exist_ok=True)
    layouts = analyze_fd_layout(pp_str)
    written = write_all_files(recs, pp_str, workdir)
    print(f"  Wrote {len(written)} flat files to {workdir}")

    # Also clean old .dat in the original dir
    for f in os.listdir(dp):
        if f.endswith('.dat') or f.endswith('.txt'):
            try: os.remove(os.path.join(dp, f))
            except: pass

    # Copy generated data to original dir (program expects files there)
    for fn, _, _ in written:
        src_f = os.path.join(workdir, fn)
        if os.path.exists(src_f):
            shutil.copy2(src_f, os.path.join(dp, fn))
            print(f"  Copied {fn} to {dp}")

    # ── 4. Instrument and compile ──
    print(f"\n[4/6] Instrumenting source...")
    # Need to instrument a copy with COPYBOOKS resolved (preprocessed)
    # But COBOL needs COPY statements to compile — instrument the ORIGINAL source
    instr_src, markers = instrument_source(src)
    print(f"  Injected {len(markers)} DISPLAY markers")

    instr_file = os.path.join(dp, f"__instrumented_{fname}")
    with open(instr_file, 'w', encoding='utf-8') as f:
        f.write(instr_src)
    exe_path = os.path.join(dp, f"__instrumented_{fname.replace('.cbl', '.exe')}")

    print(f"  Compiling instrumented program...")
    r = subprocess.run([COBC, '-x', '-Wall', instr_file, '-o', exe_path,
                        '-I', COPYBOOKS, '-I', dp],
                       capture_output=True, timeout=30, cwd=dp)
    out = r.stdout.decode('utf-8', errors='replace') if r.stdout else ''
    err = r.stderr.decode('utf-8', errors='replace') if r.stderr else ''
    if r.returncode != 0:
        ck(False, f"Instrumented compile FAIL: {err[:120]}")
        # Clean up
        try: os.remove(instr_file)
        except: pass
        continue
    print(f"  Compile OK: {os.path.getsize(exe_path)} bytes")

    # ── 5. Run ──
    print(f"\n[5/6] Running instrumented program...")
    run = subprocess.run([exe_path], capture_output=True, timeout=30,
                         cwd=dp, shell=True)
    run_out = run.stdout.decode('utf-8', errors='replace') if run.stdout else ''
    run_err = run.stderr.decode('utf-8', errors='replace') if run.stderr else ''
    rc = run.returncode

    print(f"  RC={rc}, stdout={len(run_out)} chars")

    # Extract branch markers from stdout
    actual_hits = count_unique_branch_hits(run_out)
    actual_count = len(actual_hits)
    expected_count = len(markers)

    print(f"  Branch markers injected: {expected_count}")
    print(f"  Branch markers hit at runtime: {actual_count}")

    # ── 6. Compare ──
    print(f"\n[6/6] Comparison:")
    print(f"  Our static branches:  {static_branches}")
    print(f"  Runtime DISPLAY hits: {actual_count}")
    print(f"  DISPLAY markers:      {expected_count}")

    # Our static branches = 2 per IF/EVAL/PERFORM ≈ markers / 2 roughly
    # But markers include IF + ELSE as separate, so total markers ≈ 2 * decision_points
    # The key check: runtime DISPLAY hits should equal expected markers
    # (every branch has a DISPLAY, so every branch hit = 1 DISPLAY)
    miss = expected_count - actual_count
    if miss > 0:
        print(f"\n  MISSING branches at runtime: {miss}")
        # Show which markers were NOT hit
        all_ids = set(m["id"] for m in markers)
        missed_ids = all_ids - actual_hits
        for m in markers:
            if m["id"] in missed_ids:
                print(f"    MISS: marker {m['id']}: {m['kind']} at line {m['line']}")

    # The relationship between our branches and runtime markers:
    # - Our branches = sum of all branch_names in decision points
    # - Runtime markers = DISPLAY statements that fired
    # - These should be similar (within margin for DISPLAY overhead)
    ratio = actual_count / max(static_branches, 1)
    ck(ratio > 0.7, f"Runtime coverage ratio: {ratio:.0%} ({actual_count}/{static_branches})")
    ck(miss <= expected_count * 0.3,
       f"Missing <= 30%: missed {miss}/{expected_count}")

    # ── Cleanup ──
    try:
        os.remove(instr_file)
        os.remove(exe_path)
        shutil.rmtree(workdir)
    except: pass
    print(f"  Cleanup done.")

# ── Summary ──
sec("FINAL SUMMARY")
print(f"\nThis test injects DISPLAY markers at every IF/ELSE/WHEN/AT_END branch")
print(f"in the COBOL source, compiles with REAL GnuCOBOL, and runs.")
print(f"The stdout shows exactly which branches were hit at runtime.")
print(f"This is INDEPENDENT verification — no Python involved after compilation.")
print(f"\n{'='*55}")
print(f"S20: {P} PASS / {F} FAIL")
print(f"{'='*55}")
if F > 0: sys.exit(1)