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]
if (ROOT / 'src' / 'apps').is_dir():
    sys.path.insert(0, str(ROOT / 'src'))
sys.path.insert(0, str(ROOT))

REPO_NAME = 'all4-p1-elder-paperwork'
PROJECT_MODULE = 'apps.p1_elder_paperwork.app'
DEFAULT_PACK = ROOT / 'data' / 'demo_packs' / 'p1_elder_paperwork' / 'sample_appointment_notice'


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 _prepare_env(tmpdir: Path) -> None:
    os.environ.update(
        {
            'APP_ROOT_DIR': str(ROOT),
            'DATA_DIR': str(ROOT / 'data'),
            'MODEL_REGISTRY_PATH': str(ROOT / 'configs' / 'model_registry.yaml'),
            'MODEL_CACHE_DIR': '/tmp',
            'ARTIFACT_DIR': str(tmpdir / 'artifacts'),
            'SQLITE_PATH': str(tmpdir / 'offline.sqlite3'),
            'HF_HUB_OFFLINE': '1',
            'TRANSFORMERS_OFFLINE': '1',
            'HF_DATASETS_OFFLINE': '1',
            'HF_HUB_DISABLE_TELEMETRY': '1',
            'GRADIO_ANALYTICS_ENABLED': 'False',
            'DO_NOT_TRACK': '1',
        }
    )


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

            result = run_eval_for_project(PROJECT_MODULE, pack, db_path=tmpdir / 'offline.sqlite3')
            return {
                'repo': REPO_NAME,
                'project': getattr(result, 'project', None),
                'pack_id': getattr(result, 'pack_id', None),
                'pack': _pack_label(pack),
                'passed': bool(getattr(result, 'passed', False)),
                'findings': list(getattr(result, 'findings', [])),
                '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())