#!/usr/bin/python3
"""Exercise the installed runtime under every supported Python version."""
import os
import subprocess
import sys
import textwrap

CASES = {
    "installed-core": """
        import gevent
        import gevent.libev.corecext as core
        import gevent.resolver.cares as cares
        from gevent import queue

        for module in (gevent, core, cares):
            assert module.__file__.startswith('/usr/lib/python3/dist-packages/'), module.__file__
        assert gevent.config.loop is core.loop, gevent.config.loop
        assert type(gevent.get_hub().loop) is core.loop, gevent.get_hub().loop
        assert gevent.spawn(lambda: 42).get(timeout=10) == 42
        q = queue.Queue()
        gevent.spawn(q.put, 17)
        assert q.get(timeout=10) == 17
        print(gevent.__version__, gevent.__file__, core.__file__, cares.__file__)
    """,
    "monkey-no-hub": """
        from gevent import monkey
        monkey.patch_all()
        import gevent.hub
        assert gevent.hub._get_hub() is None, 'monkey.patch_all() initialized the hub'
    """,
    "ssl-api": """
        import ssl
        import gevent.ssl as green_ssl
        import gevent.socket as green_socket

        context = green_ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.check_hostname = False
        with context.wrap_socket(green_socket.socket(), do_handshake_on_connect=False) as sock:
            for name in ('group', 'client_sigalg', 'server_sigalg'):
                if hasattr(ssl.SSLSocket, name):
                    assert hasattr(green_ssl.SSLSocket, name), name
                    assert getattr(sock, name)() is None, name
    """,
    "thread-stack-size": """
        from gevent import thread
        assert thread.stack_size(0) == 0
        for size in (-1, 123):
            try:
                thread.stack_size(size)
            except ValueError:
                pass
            else:
                raise AssertionError(size)
    """,
    "socket-subprocess": """
        from gevent import monkey
        monkey.patch_all()
        import socket
        import subprocess
        import sys
        import gevent

        left, right = socket.socketpair()
        with left, right:
            sender = gevent.spawn(right.sendall, b'hello')
            with left.makefile('rb') as stream:
                assert stream.read(5) == b'hello'
            sender.get(timeout=10)
        output = subprocess.check_output([sys.executable, '-c', 'print(42)'], timeout=10)
        assert output.strip() == b'42'
    """,
    "subprocess-encoding": """
        import sys
        from gevent import subprocess

        result = subprocess.run(
            [sys.executable, '-c', "import os; os.write(1, bytes([255]))"],
            stdout=subprocess.PIPE, text=True, encoding='ascii',
            errors='replace', check=True,
        )
        assert result.stdout == chr(65533), repr(result.stdout)
    """,
    "subprocess-timeout": """
        import sys
        import time
        from gevent import subprocess

        started = time.monotonic()
        try:
            subprocess.check_output('sleep 3', shell=True, timeout=0.2)
        except subprocess.TimeoutExpired:
            assert time.monotonic() - started < 2
        else:
            raise AssertionError('shell timeout did not fire')

        code = "import os,time; os.write(1,b'out'+bytes([226])); time.sleep(10)"
        try:
            subprocess.run([sys.executable, '-c', code], timeout=2,
                           capture_output=True, text=True, encoding='utf-8')
        except subprocess.TimeoutExpired as error:
            assert error.output == b'out' + bytes([226]), repr(error.output)
        else:
            raise AssertionError('partial-output timeout did not fire')
    """,
    "subprocess-text-buffer": """
        import sys
        from gevent import subprocess

        with subprocess.Popen(
                [sys.executable, '-c', "print('hello')"],
                stdout=subprocess.PIPE, text=True) as process:
            assert process.stdout.read(1) == 'h'
            assert process.communicate(timeout=5) == ('ello' + chr(10), None)
            assert process.communicate() == ('ello' + chr(10), None)
    """,
    "http-wsgi": """
        from gevent import monkey
        monkey.patch_all()
        from gevent.pywsgi import WSGIServer
        import urllib.request

        def application(environ, start_response):
            start_response('200 OK', [('Content-Type', 'text/plain')])
            return [b'ok']

        server = WSGIServer(('127.0.0.1', 0), application, log=None)
        server.start()
        try:
            opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
            with opener.open('http://127.0.0.1:%d/' % server.server_port, timeout=10) as response:
                assert response.read() == b'ok'
        finally:
            server.stop()
    """,
    "executor-exit": """
        from gevent import monkey
        monkey.patch_all()
        from concurrent.futures import ThreadPoolExecutor

        # Keep the executor alive until interpreter shutdown to catch exit hangs.
        executor = ThreadPoolExecutor(max_workers=1)
        assert executor.submit(lambda: 42).result(timeout=10) == 42
    """,
}

# Exercise each optional event loop in a fresh interpreter. The remaining
# cases above test backend-independent APIs and need only the default loop.
CFFI_CASES = {
    name: CASES[name]
    for name in ('socket-subprocess', 'subprocess-timeout', 'executor-exit')
}
CFFI_CASES['loop-timer-thread-fork'] = """
    import os
    import signal
    import threading
    import gevent
    from gevent import os as green_os

    hub = gevent.get_hub()
    assert gevent.spawn_later(0.02, lambda: 42).get(timeout=10) == 42

    # Fork before starting native workers. The child must be able to run
    # timers on its reinitialized loop, and the parent must observe its exit.
    pid = green_os.fork()
    if pid == 0:
        try:
            assert gevent.spawn_later(0.02, lambda: 17).get(timeout=10) == 17
        except BaseException:
            os._exit(1)
        os._exit(0)
    reaped = False
    try:
        with gevent.Timeout(10):
            child, status = green_os.waitpid(pid, 0)
        reaped = True
        assert child == pid and os.waitstatus_to_exitcode(status) == 0, (child, status)
    finally:
        if not reaped:
            try:
                os.kill(pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            green_os.waitpid(pid, 0)

    # A native worker's completion must wake the hub via the loop's async
    # watcher, rather than merely running a second greenlet in this thread.
    main_thread = threading.get_ident()
    worker_thread = hub.threadpool.spawn(threading.get_ident).get(timeout=10)
    assert worker_thread != main_thread, (main_thread, worker_thread)
"""

CFFI_PREFIX = """
    import importlib
    import os
    import gevent

    expected_loop, extension_name = {
        'libev-cffi': ('gevent.libev.corecffi', 'gevent.libev._corecffi'),
        'libuv-cffi': ('gevent.libuv.loop', 'gevent.libuv._corecffi'),
    }[os.environ['GEVENT_LOOP']]
    assert gevent.config.loop.__module__ == expected_loop, gevent.config.loop
    loop_module = importlib.import_module(expected_loop)
    extension = importlib.import_module(extension_name)
    for module in (gevent, loop_module, extension):
        assert module.__file__.startswith('/usr/lib/python3/dist-packages/'), module.__file__
    print(os.environ['GEVENT_LOOP'], extension.__file__, flush=True)
"""
CFFI_SUFFIX = """
    assert type(gevent.get_hub().loop).__module__ == expected_loop, gevent.get_hub().loop
"""

versions = subprocess.check_output(["py3versions", "-s"], text=True).split()
assert versions, "No supported Python interpreter found"
base_environment = os.environ.copy()
base_environment.pop('GEVENT_LOOP', None)
failed = False
for version in versions:
    for backend, cases in ((None, CASES), ('libev-cffi', CFFI_CASES),
                           ('libuv-cffi', CFFI_CASES)):
        environment = base_environment.copy()
        if backend is not None:
            environment['GEVENT_LOOP'] = backend
        for name, code in cases.items():
            code = textwrap.dedent(code)
            if backend is not None:
                code = textwrap.dedent(CFFI_PREFIX) + code + textwrap.dedent(CFFI_SUFFIX)
            label = f"{version} {backend or 'default'}: {name}"
            try:
                subprocess.run(
                    [version, "-I", "-c", code],
                    check=True, timeout=30, env=environment,
                )
            except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as error:
                print(f"FAIL {label}: {error}", flush=True)
                failed = True
            else:
                print(f"PASS {label}", flush=True)
sys.exit(failed)
