# -*- coding: utf-8 -*-
"""Accessories for automated py.test runner."""
# standard imports
from __future__ import with_statement, print_function
import contextlib
import subprocess
import functools
import traceback
import termios
import codecs
import curses
import sys
import pty
import os

# local
from blessed import Terminal

# 3rd-party
import pytest
import six

TestTerminal = functools.partial(Terminal, kind='xterm-256color')
SEND_SEMAPHORE = SEMAPHORE = b'SEMAPHORE\n'
RECV_SEMAPHORE = b'SEMAPHORE\r\n'
many_lines_params = [40, 80]
# we must test a '1' column for conditional in _handle_long_word
many_columns_params = [1, 10]

if os.environ.get('TEST_QUICK'):
    many_lines_params = [80,]
    many_columns_params = [25,]

all_terms_params = 'xterm screen ansi vt220 rxvt cons25 linux'.split()

if os.environ.get('TEST_FULL'):
    try:
        all_terms_params = [
            # use all values of the first column of data in output of 'toe -a'
            _term.split(None, 1)[0] for _term in
            subprocess.Popen(('toe', '-a'),
                             stdout=subprocess.PIPE,
                             close_fds=True)
            .communicate()[0].splitlines()]
    except OSError:
        pass
elif os.environ.get('TEST_QUICK'):
    all_terms_params = 'xterm screen ansi linux'.split()


def init_subproc_coverage(run_note):
        try:
            import coverage
        except ImportError:
            return None
        _coveragerc = os.path.join(
            os.path.dirname(__file__),
            os.pardir, os.pardir,
            '.coveragerc')
        cov = coverage.Coverage(config_file=_coveragerc)
        cov.set_option("run:note", run_note)
        cov.start()
        return cov


class as_subprocess(object):
    """This helper executes test cases in a child process,
       avoiding a python-internal bug of _curses: setupterm()
       may not be called more than once per process.
    """
    _CHILD_PID = 0
    encoding = 'utf8'

    def __init__(self, func):
        self.func = func

    def __call__(self, *args, **kwargs):
        pid_testrunner = os.getpid()
        pid, master_fd = pty.fork()
        if pid == self._CHILD_PID:
            # child process executes function, raises exception
            # if failed, causing a non-zero exit code, using the
            # protected _exit() function of ``os``; to prevent the
            # 'SystemExit' exception from being thrown.
            cov = init_subproc_coverage(
                "@as_subprocess-{pid};{func_name}(*{args}, **{kwargs})"
                .format(pid=os.getpid(), func_name=self.func,
                        args=args, kwargs=kwargs))
            try:
                self.func(*args, **kwargs)
            except Exception:
                e_type, e_value, e_tb = sys.exc_info()
                o_err = list()
                for line in traceback.format_tb(e_tb):
                    o_err.append(line.rstrip().encode('utf-8'))
                o_err.append(('-=' * 20).encode('ascii'))
                o_err.extend([_exc.rstrip().encode('utf-8') for _exc in
                              traceback.format_exception_only(
                                  e_type, e_value)])
                os.write(sys.__stdout__.fileno(), b'\n'.join(o_err))
                os.close(sys.__stdout__.fileno())
                os.close(sys.__stderr__.fileno())
                os.close(sys.__stdin__.fileno())
                if cov is not None:
                    cov.stop()
                    cov.save()
                os._exit(1)
            else:
                if cov is not None:
                    cov.stop()
                    cov.save()
                os._exit(0)

        # detect rare fork in test runner, when bad bugs happen
        if pid_testrunner != os.getpid():
            print('TEST RUNNER HAS FORKED, {0}=>{1}: EXIT'
                  .format(pid_testrunner, os.getpid()), file=sys.stderr)
            os._exit(1)

        exc_output = six.text_type()
        decoder = codecs.getincrementaldecoder(self.encoding)()
        while True:
            try:
                _exc = os.read(master_fd, 65534)
            except OSError:
                # linux EOF
                break
            if not _exc:
                # bsd EOF
                break
            exc_output += decoder.decode(_exc)

        # parent process asserts exit code is 0, causing test
        # to fail if child process raised an exception/assertion
        pid, status = os.waitpid(pid, 0)
        os.close(master_fd)

        # Display any output written by child process
        # (esp. any AssertionError exceptions written to stderr).
        exc_output_msg = 'Output in child process:\n%s\n%s\n%s' % (
            u'=' * 40, exc_output, u'=' * 40,)
        assert exc_output == '', exc_output_msg

        # Also test exit status is non-zero
        assert os.WEXITSTATUS(status) == 0


def read_until_semaphore(fd, semaphore=RECV_SEMAPHORE,
                         encoding='utf8', timeout=10):
    """Read file descriptor ``fd`` until ``semaphore`` is found.

    Used to ensure the child process is awake and ready. For timing
    tests; without a semaphore, the time to fork() would be (incorrectly)
    included in the duration of the test, which can be very length on
    continuous integration servers (such as Travis-CI).
    """
    # note that when a child process writes xyz\\n, the parent
    # process will read xyz\\r\\n -- this is how pseudo terminals
    # behave; a virtual terminal requires both carriage return and
    # line feed, it is only for convenience that \\n does both.
    outp = six.text_type()
    decoder = codecs.getincrementaldecoder(encoding)()
    semaphore = semaphore.decode('ascii')
    while not outp.startswith(semaphore):
        try:
            _exc = os.read(fd, 1)
        except OSError:  # Linux EOF
            break
        if not _exc:     # BSD EOF
            break
        outp += decoder.decode(_exc, final=False)
    assert outp.startswith(semaphore), (
        'Semaphore not recv before EOF '
        '(expected: %r, got: %r)' % (semaphore, outp,))
    return outp[len(semaphore):]


def read_until_eof(fd, encoding='utf8'):
    """Read file descriptor ``fd`` until EOF. Return decoded string."""
    decoder = codecs.getincrementaldecoder(encoding)()
    outp = six.text_type()
    while True:
        try:
            _exc = os.read(fd, 100)
        except OSError:  # linux EOF
            break
        if not _exc:  # bsd EOF
            break
        outp += decoder.decode(_exc, final=False)
    return outp


@contextlib.contextmanager
def echo_off(fd):
    """Ensure any bytes written to pty fd are not duplicated as output."""
    try:
        attrs = termios.tcgetattr(fd)
        attrs[3] = attrs[3] & ~termios.ECHO
        termios.tcsetattr(fd, termios.TCSANOW, attrs)
        yield
    finally:
        attrs[3] = attrs[3] | termios.ECHO
        termios.tcsetattr(fd, termios.TCSANOW, attrs)


def unicode_cap(cap):
    """Return the result of ``tigetstr`` except as Unicode."""
    try:
        val = curses.tigetstr(cap)
    except curses.error:
        val = None
    if val:
        return val.decode('latin1')
    return u''


def unicode_parm(cap, *parms):
    """Return the result of ``tparm(tigetstr())`` except as Unicode."""
    try:
        cap = curses.tigetstr(cap)
    except curses.error:
        cap = None
    if cap:
        try:
            val = curses.tparm(cap, *parms)
        except curses.error:
            val = None
        if val:
            return val.decode('latin1')
    return u''


@pytest.fixture(params=all_terms_params)
def all_terms(request):
    """Common kind values for all kinds of terminals."""
    return request.param


@pytest.fixture(params=many_lines_params)
def many_lines(request):
    """Various number of lines for screen height."""
    return request.param


@pytest.fixture(params=many_columns_params)
def many_columns(request):
    """Various number of columns for screen width."""
    return request.param