# Utilities for testing
import asyncio
import socket as stdlib_socket
import threading
import os
import sys
from typing import TYPE_CHECKING

import pytest
import warnings
from contextlib import contextmanager, closing

import gc

# See trio/tests/conftest.py for the other half of this
from trio.tests.conftest import RUN_SLOW

slow = pytest.mark.skipif(not RUN_SLOW, reason="use --run-slow to run slow tests")

# PyPy 7.2 was released with a bug that just never called the async
# generator 'firstiter' hook at all.  This impacts tests of end-of-run
# finalization (nothing gets added to runner.asyncgens) and tests of
# "foreign" async generator behavior (since the firstiter hook is what
# marks the asyncgen as foreign), but most tests of GC-mediated
# finalization still work.
buggy_pypy_asyncgens = (
    not TYPE_CHECKING
    and sys.implementation.name == "pypy"
    and sys.pypy_version_info < (7, 3)
)

try:
    s = stdlib_socket.socket(stdlib_socket.AF_INET6, stdlib_socket.SOCK_STREAM, 0)
except OSError:  # pragma: no cover
    # Some systems don't even support creating an IPv6 socket, let alone
    # binding it. (ex: Linux with 'ipv6.disable=1' in the kernel command line)
    # We don't have any of those in our CI, and there's nothing that gets
    # tested _only_ if can_create_ipv6 = False, so we'll just no-cover this.
    can_create_ipv6 = False
    can_bind_ipv6 = False
else:
    can_create_ipv6 = True
    with s:
        try:
            s.bind(("::1", 0))
        except OSError:
            can_bind_ipv6 = False
        else:
            can_bind_ipv6 = True

creates_ipv6 = pytest.mark.skipif(not can_create_ipv6, reason="need IPv6")
binds_ipv6 = pytest.mark.skipif(not can_bind_ipv6, reason="need IPv6")


def gc_collect_harder():
    # In the test suite we sometimes want to call gc.collect() to make sure
    # that any objects with noisy __del__ methods (e.g. unawaited coroutines)
    # get collected before we continue, so their noise doesn't leak into
    # unrelated tests.
    #
    # On PyPy, coroutine objects (for example) can survive at least 1 round of
    # garbage collection, because executing their __del__ method to print the
    # warning can cause them to be resurrected. So we call collect a few times
    # to make sure.
    for _ in range(4):
        gc.collect()


# Some of our tests need to leak coroutines, and thus trigger the
# "RuntimeWarning: coroutine '...' was never awaited" message. This context
# manager should be used anywhere this happens to hide those messages, because
# when expected they're clutter.
@contextmanager
def ignore_coroutine_never_awaited_warnings():
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", message="coroutine '.*' was never awaited")
        try:
            yield
        finally:
            # Make sure to trigger any coroutine __del__ methods now, before
            # we leave the context manager.
            gc_collect_harder()


def _noop(*args, **kwargs):
    pass


if sys.version_info >= (3, 8):

    @contextmanager
    def restore_unraisablehook():
        sys.unraisablehook, prev = sys.__unraisablehook__, sys.unraisablehook
        try:
            yield
        finally:
            sys.unraisablehook = prev

    @contextmanager
    def disable_threading_excepthook():
        if sys.version_info >= (3, 10):
            threading.excepthook, prev = threading.__excepthook__, threading.excepthook
        else:
            threading.excepthook, prev = _noop, threading.excepthook

        try:
            yield
        finally:
            threading.excepthook = prev

else:

    @contextmanager
    def restore_unraisablehook():  # pragma: no cover
        yield

    @contextmanager
    def disable_threading_excepthook():  # pragma: no cover
        yield


# template is like:
#   [1, {2.1, 2.2}, 3] -> matches [1, 2.1, 2.2, 3] or [1, 2.2, 2.1, 3]
def check_sequence_matches(seq, template):
    i = 0
    for pattern in template:
        if not isinstance(pattern, set):
            pattern = {pattern}
        got = set(seq[i : i + len(pattern)])
        assert got == pattern
        i += len(got)


# https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=246350
skip_if_fbsd_pipes_broken = pytest.mark.skipif(
    sys.platform != "win32"  # prevent mypy from complaining about missing uname
    and hasattr(os, "uname")
    and os.uname().sysname == "FreeBSD"
    and os.uname().release[:4] < "12.2",
    reason="hangs on FreeBSD 12.1 and earlier, due to FreeBSD bug #246350",
)


def create_asyncio_future_in_new_loop():
    with closing(asyncio.new_event_loop()) as loop:
        return loop.create_future()