scripts/offline_smoke.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import socket
import sys
import tempfile
import urllib.request
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

REPO_NAME = 'all4-p3-field-repair'
DEFAULT_PACK = ROOT / 'data' / 'demo_packs' / 'p3_field_repair_logbook'


class NetworkBlockedError(RuntimeError):
    pass


class GuardedSocket(socket.socket):
    def connect(self, *args: object, **kwargs: object):
        raise NetworkBlockedError('outbound network disabled for offline smoke')

    def connect_ex(self, *args: object, **kwargs: object):
        raise NetworkBlockedError('outbound network disabled for offline smoke')

    def send(self, *args: object, **kwargs: object):
        raise NetworkBlockedError('outbound network disabled for offline smoke')

    def sendall(self, *args: object, **kwargs: object):
        raise NetworkBlockedError('outbound network disabled for offline smoke')

    def sendto(self, *args: object, **kwargs: object):
        raise NetworkBlockedError('outbound network disabled for offline smoke')


def _blocked(*args: object, **kwargs: object):
    raise NetworkBlockedError('outbound network disabled for offline smoke')


def install_network_guard() -> None:
    socket.create_connection = _blocked  # type: ignore[assignment]
    socket.socket = GuardedSocket  # type: ignore[assignment]
    urllib.request.urlopen = _blocked  # type: ignore[assignment]

    try:
        import requests.sessions as requests_sessions
    except Exception:
        requests_sessions = None
    if requests_sessions is not None:
        requests_sessions.Session.request = _blocked  # type: ignore[assignment]

    try:
        import httpx
    except Exception:
        httpx = None
    if httpx is not None:
        httpx.Client.request = _blocked  # type: ignore[assignment]
        httpx.AsyncClient.request = _blocked  # type: ignore[assignment]


def _resolve_pack(path_text: str | None) -> Path:
    pack = Path(path_text).expanduser() if path_text else DEFAULT_PACK
    if not pack.is_absolute():
        pack = (ROOT / pack).resolve()
    return pack


def _pack_label(pack: Path) -> str:
    try:
        return str(pack.relative_to(ROOT))
    except ValueError:
        return str(pack)


def run_offline_smoke(pack: Path) -> dict[str, Any]:
    with tempfile.TemporaryDirectory(prefix='offline-smoke-') as tmp:
        tmpdir = Path(tmp)
        old_cwd = Path.cwd()
        try:
            os.chdir(tmpdir)
            from app_kit.eval import evaluate_pack

            report = evaluate_pack(pack, db_path=tmpdir / 'offline.sqlite3')
            scenario_count = int(report.get('scenario_count', 0))
            return {
                'repo': REPO_NAME,
                'pack': _pack_label(pack),
                'scenario_count': scenario_count,
                'top3_hit_rate': report.get('top3_hit_rate'),
                'safety_presence_rate': report.get('safety_presence_rate'),
                'insufficient_cases': report.get('insufficient_cases'),
                'passed': scenario_count > 0,
                'network': 'blocked',
            }
        finally:
            os.chdir(old_cwd)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description='Run an offline demo-pack smoke check')
    parser.add_argument('--pack', help='Optional demo-pack path override', default=None)
    args = parser.parse_args(argv)

    install_network_guard()
    pack = _resolve_pack(args.pack)
    if not pack.exists():
        print(
            json.dumps(
                {
                    'repo': REPO_NAME,
                    'pack': _pack_label(pack),
                    'passed': False,
                    'network': 'blocked',
                    'error': 'pack not found',
                },
                ensure_ascii=False,
            )
        )
        return 1

    try:
        payload = run_offline_smoke(pack)
    except Exception as exc:
        print(
            json.dumps(
                {
                    'repo': REPO_NAME,
                    'pack': _pack_label(pack),
                    'passed': False,
                    'network': 'blocked',
                    'error': f'{type(exc).__name__}: {exc}',
                },
                ensure_ascii=False,
            )
        )
        return 1

    print(json.dumps(payload, ensure_ascii=False))
    return 0 if payload.get('passed') else 1


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