#!/usr/bin/env python3
"""Synthetic bounded asyncio subprocess tests; no user-controlled command or shell.

Run with: python3 mcp_subprocess_timeout_smoke.py --self-test
Uses only the Python interpreter running this script, in temporary child
processes that print known text / a known bounded output / sleep briefly.
No MCP, cloud, credentials, application or service is needed.
"""
from __future__ import annotations

import argparse
import asyncio
import json
import sys

MAX_OUTPUT = 256_000


async def run_fixed_python(script: str, deadline: float) -> dict[str, object]:
    # script is a constant from this module's self-test, not external input.
    proc = await asyncio.create_subprocess_exec(
        sys.executable, "-c", script,
        stdin=asyncio.subprocess.DEVNULL,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), deadline)
        if len(stdout) + len(stderr) > MAX_OUTPUT:
            return {"outcome": "OUTPUT_TOO_LARGE", "exit_code": proc.returncode}
        return {
            "outcome": "FINISHED",
            "exit_code": proc.returncode,
            "stdout_bytes": len(stdout),
            "stderr_bytes": len(stderr),
        }
    except asyncio.TimeoutError:
        if proc.returncode is None:
            proc.kill()
        await proc.communicate()  # drain known-bounded child output & reap child
        return {"outcome": "TIMED_OUT_CHILD_REAPED", "exit_code": proc.returncode}


async def self_test() -> dict[str, object]:
    cases = [
        ("print('ok')", 2.0, "FINISHED", 0, "short_stdout"),
        ("import sys; sys.stderr.write('diagnostic\\n')", 2.0, "FINISHED", 0, "short_stderr"),
        ("import sys; sys.stdout.write('A'*131072)", 2.0, "FINISHED", 0, "larger_than_common_pipe_buffer"),
        ("import sys; sys.exit(7)", 2.0, "FINISHED", 7, "nonzero_exit"),
        ("import time; time.sleep(2.0)", .2, "TIMED_OUT_CHILD_REAPED", None, "timeout_process_reaped"),
    ]
    assertions=[]
    for script,deadline,expected_status,expected_code,label in cases:
        result=await run_fixed_python(script,deadline)
        assert result["outcome"]==expected_status,(label,result)
        if expected_code is not None:assert result["exit_code"]==expected_code,(label,result)
        if label=="larger_than_common_pipe_buffer":
            assert result["stdout_bytes"]==131072
        if label=="timeout_process_reaped":
            assert isinstance(result["exit_code"],int) and result["exit_code"]!=0
        assertions.append(label)
    return {
        "status": "PASS",
        "synthetic_subprocess_checks": assertions,
        "total": len(assertions),
        "tested_mcp_server": False,
        "user_commands_executed": 0,
        "network_requests": 0,
        "wrote_user_files": False,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--self-test", action="store_true", required=True)
    parser.parse_args()
    print(json.dumps(asyncio.run(self_test()), indent=2))
    return 0


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