#!/usr/bin/env python3
"""Generate and optionally run HLDS emulator test suites.

This tool sits above the analyzer and dumper/media tooling:
  * analyzer emits device profiles
  * dumper/media tooling emits media profiles
  * session builder emits session configs
  * this tool emits repeatable emulator tests and can run them

It intentionally does not bake model facts into the emulator. It reads device,
media, and session profiles, writes JSONL request scripts, and optionally runs
those scripts through gdr_host_jsonl_bridge.py. Full GDR-8050L authentication
validation is delegated to the existing packet runner because it builds the
challenge/response sequence from the captured unlock log.
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable

from hlds_profiles import (
    load_device_profile,
    load_media_profile,
    load_session_profile,
    effective_session,
    session_value,
    write_json,
)


def hx(data: bytes) -> str:
    return ' '.join(f'{b:02X}' for b in data)


def cdb_read10(lba: int, blocks: int) -> str:
    lba = int(max(0, lba)) & 0xFFFFFFFF
    blocks = int(max(0, min(0xFFFF, blocks)))
    return hx(bytes([0x28, 0x00]) + lba.to_bytes(4, 'big') + bytes([0x00]) + blocks.to_bytes(2, 'big') + bytes([0x00]))


def cdb_read12(lba: int, blocks: int) -> str:
    lba = int(max(0, lba)) & 0xFFFFFFFF
    blocks = int(max(0, blocks)) & 0xFFFFFFFF
    return hx(bytes([0xA8, 0x00]) + lba.to_bytes(4, 'big') + blocks.to_bytes(4, 'big') + bytes([0x00, 0x00]))


def cdb_write10(lba: int, blocks: int) -> str:
    lba = int(max(0, lba)) & 0xFFFFFFFF
    blocks = int(max(0, min(0xFFFF, blocks)))
    return hx(bytes([0x2A, 0x00]) + lba.to_bytes(4, 'big') + bytes([0x00]) + blocks.to_bytes(2, 'big') + bytes([0x00]))


def cdb_write12(lba: int, blocks: int) -> str:
    lba = int(max(0, lba)) & 0xFFFFFFFF
    blocks = int(max(0, blocks)) & 0xFFFFFFFF
    return hx(bytes([0xAA, 0x00]) + lba.to_bytes(4, 'big') + blocks.to_bytes(4, 'big') + bytes([0x00, 0x00]))


def safe_name(text: str) -> str:
    out = ''.join(ch if ch.isalnum() or ch in '._-' else '_' for ch in str(text))
    return out.strip('_') or 'emutest'


def load_any_session(args: argparse.Namespace) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], str | None, str | None]:
    session = effective_session(
        args.session_profile,
        device_profile=args.device_profile,
        media_profile=args.media_profile,
        firmware_image=args.firmware_image,
    ) if (args.session_profile or args.device_profile or args.media_profile or args.firmware_image) else {}
    device_path = args.device_profile or session.get('device_profile')
    media_path = args.media_profile or session.get('media_profile')
    device = load_device_profile(device_path) if device_path else {}
    media = load_media_profile(media_path) if media_path else {}
    return session, device, media, str(device_path) if device_path else None, str(media_path) if media_path else None


def media_image(media: dict[str, Any]) -> dict[str, Any]:
    return dict(media.get('image') or {})


def media_layout(media: dict[str, Any]) -> dict[str, Any]:
    return dict((media.get('image') or {}).get('layout') or media.get('layout') or {})


def media_sector_count(media: dict[str, Any]) -> int:
    image = media_image(media)
    layout = media_layout(media)
    return int(image.get('sector_count') or media.get('sector_count') or layout.get('sector_count') or 0)


def media_sector_size(media: dict[str, Any]) -> int:
    image = media_image(media)
    return int(image.get('sector_size') or media.get('sector_size') or 2048)


def media_file_path(media: dict[str, Any], media_path: str | None) -> str | None:
    image = media_image(media)
    file = image.get('file') or media.get('iso') or None
    if not file:
        return None
    p = Path(str(file))
    if not p.is_absolute() and media_path:
        p = Path(media_path).resolve().parent / p
    return str(p)


def first_true_cap(caps: dict[str, Any], names: Iterable[str]) -> bool:
    for name in names:
        if caps.get(name):
            return True
    return False


def supports_writes(device: dict[str, Any]) -> bool:
    scsi = dict(device.get('scsi_capabilities') or {})
    mmc = dict(device.get('mmc_capabilities') or {})
    write_media = dict(mmc.get('write_media') or {})
    return first_true_cap(scsi, ('supports_write_10', 'supports_write_12', 'supports_write_and_verify_10')) or any(bool(v) for v in write_media.values())


def supports_erase(device: dict[str, Any]) -> bool:
    scsi = dict(device.get('scsi_capabilities') or {})
    mmc = dict(device.get('mmc_capabilities') or {})
    erasable = dict(mmc.get('erasable_media') or {})
    return first_true_cap(scsi, ('supports_blank', 'supports_format_unit')) or any(bool(v) for v in erasable.values())


def has_xbox_auth(device: dict[str, Any]) -> bool:
    return bool((device.get('xbox_auth') or device.get('xbox_auth_capabilities') or {}).get('supported'))


def base_lbas(media: dict[str, Any]) -> list[int]:
    layout = media_layout(media)
    total = media_sector_count(media)
    vals: list[int] = []
    def add(v: Any) -> None:
        try:
            n = int(v)
        except Exception:
            return
        if n >= 0 and n not in vals:
            vals.append(n)
    add(0)
    add(1)
    add(layout.get('game_unlocked_source_start_lba'))
    add(layout.get('visible_sector_count'))
    add(layout.get('game_output_start_lba'))
    layer = layout.get('layer_break_lba')
    if layer is not None:
        add(int(layer) - 1)
        add(layer)
        add(int(layer) + 1)
    if total:
        add(total - 2)
        add(total - 1)
    return vals


def req(tag: str, cdb_hex: str, data_out_hex: str | None = None, **extra: Any) -> dict[str, Any]:
    row: dict[str, Any] = {'tag': tag, 'cdb_hex': cdb_hex}
    if data_out_hex is not None:
        row['data_out_hex'] = data_out_hex
    row.update(extra)
    return row


def generate_profile_controls() -> list[dict[str, Any]]:
    return [
        {'op': 'state', 'tag': 'state_initial'},
        {'op': 'media_model', 'tag': 'media_model'},
        {'op': 'firmware_lba_state', 'tag': 'firmware_lba_state'},
    ]


def generate_scsi_basic_locked(device: dict[str, Any], media: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = [{'op': 'reset_auth', 'tag': 'reset_locked', 'stage': 'locked'}]
    rows += [
        req('locked_test_unit_ready', '00 00 00 00 00 00'),
        req('locked_inquiry_36', '12 00 00 00 24 00'),
        req('locked_inquiry_vpd_supported_pages', '12 01 00 00 FF 00'),
        req('locked_inquiry_vpd_serial', '12 01 80 00 FF 00'),
        req('locked_inquiry_vpd_device_id', '12 01 83 00 FF 00'),
        req('locked_request_sense_18', '03 00 00 00 12 00'),
        req('locked_read_capacity_10', '25 00 00 00 00 00 00 00 00 00'),
        req('locked_mode_sense10_3e', '5A 00 3E 00 00 00 00 00 1C 00'),
        req('locked_mode_sense6_3e', '1A 00 3E 00 18 00'),
        req('locked_get_configuration_64', '46 02 00 00 00 00 00 00 40 00'),
        req('locked_read_toc_20', '43 00 00 00 00 00 00 00 14 00'),
        req('locked_read_disc_information_34', '51 00 00 00 00 00 00 00 22 00'),
        req('locked_get_event_status_8', '4A 01 00 00 10 00 00 00 08 00'),
        req('locked_prevent_allow_false', '1E 00 00 00 00 00'),
        req('locked_start_stop_nop', '1B 00 00 00 00 00'),
    ]
    if has_xbox_auth(device):
        auth_cdb = (device.get('xbox_auth') or {}).get('read_dvd_structure_auth_cdb') or 'AD00FF02FDFFFE00066400C0'
        rows.append(req('locked_read_dvd_structure_ad_c0', str(auth_cdb)))
    rows += [
        req('locked_read_dvd_structure_pfi', 'AD 00 00 00 00 00 00 00 08 00 00 00'),
        req('locked_read_dvd_structure_dmi', 'AD 00 00 00 00 00 00 04 08 00 00 00'),
        req('locked_read10_lba0_1', cdb_read10(0, 1)),
    ]
    return rows


def generate_scsi_basic_unlocked(device: dict[str, Any], media: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = [{'op': 'reset_auth', 'tag': 'reset_unlocked', 'stage': 'unlocked'}]
    rows += [
        req('unlocked_test_unit_ready', '00 00 00 00 00 00'),
        req('unlocked_inquiry_36', '12 00 00 00 24 00'),
        req('unlocked_request_sense_18', '03 00 00 00 12 00'),
        req('unlocked_read_capacity_10', '25 00 00 00 00 00 00 00 00 00'),
        req('unlocked_mode_sense10_3e', '5A 00 3E 00 00 00 00 00 1C 00'),
        req('unlocked_mode_sense6_3e', '1A 00 3E 00 18 00'),
        req('unlocked_get_configuration_64', '46 02 00 00 00 00 00 00 40 00'),
        req('unlocked_read_toc_20', '43 00 00 00 00 00 00 00 14 00'),
        req('unlocked_read_disc_information_34', '51 00 00 00 00 00 00 00 22 00'),
    ]
    for lba in base_lbas(media)[:10]:
        rows.append(req(f'unlocked_read10_lba{lba}_1', cdb_read10(lba, 1)))
    return rows


def generate_media_read_matrix(media: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = [{'op': 'reset_auth', 'tag': 'reset_unlocked_for_reads', 'stage': 'unlocked'}]
    for lba in base_lbas(media):
        rows.append(req(f'read10_lba{lba}_1', cdb_read10(lba, 1)))
        rows.append(req(f'read12_lba{lba}_1', cdb_read12(lba, 1)))
    layout = media_layout(media)
    for name in ('game_unlocked_source_start_lba', 'game_output_start_lba', 'layer_break_lba'):
        if layout.get(name) is not None:
            lba = int(layout[name])
            rows.append(req(f'read10_{name}_16', cdb_read10(lba, 16)))
    return rows


def generate_media_write_capture(device: dict[str, Any], media: dict[str, Any], include_unsupported: bool = False) -> list[dict[str, Any]]:
    if not include_unsupported and not supports_writes(device) and not supports_erase(device):
        return []
    block = bytes((i & 0xFF for i in range(media_sector_size(media))))
    payload = hx(block)
    rows: list[dict[str, Any]] = [{'op': 'reset_auth', 'tag': 'reset_unlocked_for_write_capture', 'stage': 'unlocked'}]
    rows += [
        req('capture_write10_lba0_1', cdb_write10(0, 1), payload, expected_policy='capture_only'),
        req('capture_write12_lba1_1', cdb_write12(1, 1), payload, expected_policy='capture_only'),
        req('capture_synchronize_cache_10', '35 00 00 00 00 00 00 00 00 00'),
        req('capture_reserve_track_16_blocks', '53 00 00 00 00 00 00 00 10 00'),
        req('capture_close_track_session', '5B 00 02 00 00 00 00 00 00 00'),
        req('capture_blank_minimal', 'A1 00 00 00 00 00 00 00 00 00 00 00'),
        req('capture_format_unit_minimal', '04 11 00 00 00 00'),
    ]
    return rows


def generate_firmware_update_capture(device: dict[str, Any], include_vendor_probe: bool = True) -> list[dict[str, Any]]:
    payload = bytes([0x48, 0x4C, 0x44, 0x53]) + bytes(range(60))
    rows: list[dict[str, Any]] = [
        req('fwprobe_inquiry_36', '12 00 00 00 24 00'),
        req('fwprobe_inquiry_vpd_supported_pages', '12 01 00 00 FF 00'),
        req('fwprobe_read_buffer_capacity', '5C 00 00 00 00 00 00 00 20 00'),
        req('fwprobe_write_buffer_capture', '3B 05 00 00 00 00 00 00 40 00', hx(payload), expected_policy='capture_only'),
        req('fwprobe_request_sense_after_write_buffer', '03 00 00 00 12 00'),
    ]
    if include_vendor_probe:
        # 0xE7 was useful in prior HLDS/GCC exploration. Keep payload tiny and capture-only.
        rows.append(req('fwprobe_vendor_e7_capture', 'E7 00 00 00 00 00 00 00 04 00', '48 4C 44 53', expected_policy='capture_only'))
        rows.append(req('fwprobe_vendor_e7_request_sense', '03 00 00 00 12 00'))
    return rows


SUITE_BUILDERS = {
    'profile_controls': lambda d, m, a: generate_profile_controls(),
    'scsi_basic_locked': lambda d, m, a: generate_scsi_basic_locked(d, m),
    'scsi_basic_unlocked': lambda d, m, a: generate_scsi_basic_unlocked(d, m),
    'media_read_matrix': lambda d, m, a: generate_media_read_matrix(m),
    'media_write_capture': lambda d, m, a: generate_media_write_capture(d, m, include_unsupported=a.include_unsupported_write_tests),
    'firmware_update_capture': lambda d, m, a: generate_firmware_update_capture(d, include_vendor_probe=not a.no_vendor_firmware_probe),
}


def expand_suites(suites: list[str]) -> list[str]:
    if not suites or 'default' in suites:
        suites = ['profile_controls', 'scsi_basic_locked', 'scsi_basic_unlocked', 'media_read_matrix']
    if 'all' in suites:
        suites = list(SUITE_BUILDERS.keys())
    out: list[str] = []
    for s in suites:
        if s == 'default':
            continue
        if s not in SUITE_BUILDERS:
            raise SystemExit(f'unknown suite: {s}; choices: {", ".join(SUITE_BUILDERS)}')
        if s not in out:
            out.append(s)
    return out


def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(''.join(json.dumps(r, separators=(',', ':')) + '\n' for r in rows), encoding='utf-8')


def run_cmd(cmd: list[str], cwd: Path | None, log_path: Path | None = None) -> int:
    print('[RUN]', ' '.join(str(x) for x in cmd))
    proc = subprocess.run(cmd, cwd=str(cwd) if cwd else None, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    if log_path:
        log_path.parent.mkdir(parents=True, exist_ok=True)
        log_path.write_text(proc.stdout, encoding='utf-8')
    sys.stdout.write(proc.stdout)
    return int(proc.returncode)


def run_jsonl_suite(args: argparse.Namespace, suite_name: str, script_path: Path, out_dir: Path, extra: dict[str, Any]) -> dict[str, Any]:
    resp = out_dir / f'{suite_name}.responses.jsonl'
    payload_dir = out_dir / f'{suite_name}.payloads'
    verify = out_dir / f'{suite_name}.verify.md'
    log = out_dir / f'{suite_name}.run.log'
    cmd = [args.python, args.jsonl_bridge, '--script', str(script_path), '--out', str(resp), '--write-response-dir', str(payload_dir), '--verify-md', str(verify)]
    if args.session_profile:
        cmd += ['--session-profile', args.session_profile]
    else:
        if args.device_profile:
            cmd += ['--device-profile', args.device_profile]
        if args.media_profile:
            cmd += ['--media-profile', args.media_profile]
        if args.media_profile:
            cmd += [args.media_profile]
        if args.iso:
            cmd += [args.iso]
    if args.read_map:
        cmd += ['--xdvd-read-map', args.read_map]
    if args.log:
        cmd += ['--log', args.log]
    rc = run_cmd(cmd, Path(args.cwd) if args.cwd else None, log)
    rows = []
    if resp.exists():
        rows = [json.loads(line) for line in resp.read_text(encoding='utf-8').splitlines() if line.strip()]
    return {
        'suite': suite_name,
        'returncode': rc,
        'requests': len(rows),
        'ok': sum(1 for r in rows if r.get('ok')),
        'check_condition': sum(1 for r in rows if r.get('scsi_status') == 'CHECK CONDITION'),
        'script': str(script_path),
        'responses': str(resp),
        'verify_md': str(verify),
        'run_log': str(log),
    }


def write_report(path: Path, manifest: dict[str, Any]) -> None:
    lines: list[str] = []
    lines.append('# HLDS emulator test report')
    lines.append('')
    lines.append(f"stem={manifest.get('stem')}")
    lines.append(f"session_profile={manifest.get('session_profile') or ''}")
    lines.append(f"device_profile={manifest.get('device_profile') or ''}")
    lines.append(f"media_profile={manifest.get('media_profile') or ''}")
    lines.append('')
    lines.append('| suite | requests | ok | check_condition | rc |')
    lines.append('|---|---:|---:|---:|---:|')
    for r in manifest.get('run_results') or []:
        lines.append(f"| `{r.get('suite')}` | {r.get('requests',0)} | {r.get('ok',0)} | {r.get('check_condition',0)} | {r.get('returncode', '')} |")
    if manifest.get('validation_runner'):
        vr = manifest['validation_runner']
        lines.append('')
        lines.append('## Full packet/session validation runner')
        lines.append(f"returncode={vr.get('returncode')}")
        lines.append(f"log={vr.get('run_log')}")
    path.write_text('\n'.join(lines) + '\n', encoding='utf-8')


def write_run_all_cmd(path: Path, manifest_name: str) -> None:
    path.write_text('@echo off\r\npython "%~dp0..\\hlds_emutest_builder.py" --manifest "%~dp0' + manifest_name + '" --run-jsonl %*\r\n', encoding='utf-8')


def main() -> int:
    ap = argparse.ArgumentParser(description='Generate and optionally run HLDS emulator tests from profiles/session config')
    ap.add_argument('--session-profile', default=None, help='session profile generated by hlds_session_builder.py')
    ap.add_argument('--device-profile', default=None, help='device profile generated by analyzer chain')
    ap.add_argument('--media-profile', default=None, help='media profile / XDVD sidecar generated by dumper/media tooling')
    ap.add_argument('--firmware-image', default=None)
    ap.add_argument('--iso', default=None, help='raw ISO override when not using a session profile')
    ap.add_argument('--log', default=None, help='unlock/auth log path override')
    ap.add_argument('--stem', default=None)
    ap.add_argument('--out-dir', default=None)
    ap.add_argument('--suite', action='append', choices=['default','all'] + list(SUITE_BUILDERS.keys()), help='suite to generate; repeatable; default generates non-mutating read/profile suites')
    ap.add_argument('--manifest', default=None, help='existing manifest to run instead of generating new scripts')
    ap.add_argument('--run-jsonl', action='store_true', help='run generated JSONL scripts through gdr_host_jsonl_bridge.py')
    ap.add_argument('--run-validation', action='store_true', help='run full packet/session validation runner too')
    ap.add_argument('--python', default=sys.executable)
    ap.add_argument('--jsonl-bridge', default='gdr_host_jsonl_bridge.py')
    ap.add_argument('--packet-runner', default='run_gdr8050l_auth_packet_v166_windows_iscsi_bridge.py')
    ap.add_argument('--cwd', default=None, help='working directory for subprocesses; default current process cwd')
    ap.add_argument('--read-map', default=None, help='override --xdvd-read-map for JSONL bridge runs')
    ap.add_argument('--include-unsupported-write-tests', action='store_true', help='generate write/erase capture tests even for read-only device profiles')
    ap.add_argument('--no-vendor-firmware-probe', action='store_true')
    args = ap.parse_args()

    if args.manifest:
        manifest_path = Path(args.manifest)
        manifest = json.loads(manifest_path.read_text(encoding='utf-8'))
        args.session_profile = args.session_profile or manifest.get('session_profile')
        args.device_profile = args.device_profile or manifest.get('device_profile')
        args.media_profile = args.media_profile or manifest.get('media_profile')
        args.stem = args.stem or manifest.get('stem')
        out_dir = manifest_path.parent
    else:
        session, device, media, device_path, media_path = load_any_session(args)
        stem = safe_name(args.stem or session.get('stem') or (Path(str(args.session_profile)).stem if args.session_profile else 'hlds_emutest'))
        out_dir = Path(args.out_dir or (Path('emutests') / stem))
        scripts_dir = out_dir / 'scripts'
        suites = expand_suites(args.suite or ['default'])
        manifest = {
            'format': 'hlds-emutest-manifest',
            'version': 1,
            'stem': stem,
            'session_profile': args.session_profile,
            'device_profile': args.device_profile or device_path,
            'media_profile': args.media_profile or media_path,
            'suites': [],
            'notes': [
                'Analyzer chain generates device profiles; dumper/media tooling generates media profiles; this manifest binds generated test scripts to a session/profile set.',
                'Full auth challenge validation is delegated to the packet runner; JSONL unlocked tests use reset_auth stage=unlocked for read-path coverage.',
            ],
        }
        for suite in suites:
            rows = SUITE_BUILDERS[suite](device, media, args)
            if not rows:
                manifest['suites'].append({'name': suite, 'skipped': True, 'reason': 'not supported by profile or not requested'})
                continue
            script_path = scripts_dir / f'{suite}.jsonl'
            write_jsonl(script_path, rows)
            manifest['suites'].append({'name': suite, 'script': str(script_path), 'requests': len(rows)})
        out_dir.mkdir(parents=True, exist_ok=True)
        manifest_path = out_dir / 'manifest.json'
        write_json(manifest_path, manifest)
        write_run_all_cmd(out_dir / 'run_all_emutests.cmd', manifest_path.name)
        print(f'[OK] wrote emutest manifest: {manifest_path}')
        print(f'[OK] wrote test scripts under: {scripts_dir}')

    run_results: list[dict[str, Any]] = []
    if args.run_jsonl:
        for suite in manifest.get('suites') or []:
            if suite.get('skipped'):
                continue
            script = Path(str(suite.get('script')))
            if not script.is_absolute():
                # Paths in freshly-generated manifests are usually relative to cwd;
                # paths in copied manifests may be relative to manifest dir.
                if script.exists():
                    pass
                else:
                    script = Path(args.manifest).parent / script if args.manifest else script
            run_results.append(run_jsonl_suite(args, str(suite.get('name')), script, out_dir / 'runs', manifest))
        manifest['run_results'] = run_results
        write_json(out_dir / 'manifest.json', manifest)
        write_report(out_dir / 'report.md', manifest)
        print(f'[OK] wrote emutest report: {out_dir / "report.md"}')

    if args.run_validation:
        log = out_dir / 'runs' / 'packet_validation.run.log'
        cmd = [args.python, args.packet_runner]
        if args.session_profile:
            cmd += ['--session-profile', args.session_profile]
        elif args.media_profile:
            cmd += [args.media_profile]
            if args.iso:
                cmd += [args.iso]
        cmd += ['--windows-probes']
        rc = run_cmd(cmd, Path(args.cwd) if args.cwd else None, log)
        manifest['validation_runner'] = {'returncode': rc, 'run_log': str(log)}
        write_json(out_dir / 'manifest.json', manifest)
        write_report(out_dir / 'report.md', manifest)
        print(f'[OK] packet validation rc={rc}')

    if args.run_jsonl or args.run_validation:
        # Return nonzero only for runner crashes, not expected CHECK CONDITION tests.
        bad_rc = [r for r in run_results if int(r.get('returncode') or 0) not in (0, 2)]
        if bad_rc:
            return 1
    return 0


if __name__ == '__main__':
    raise SystemExit(main())
