# -*- coding: utf-8 -*-
# tag: ipython

"""Tests for the Cython magics extension."""

from __future__ import absolute_import

import os
import io
import sys
from contextlib import contextmanager
from unittest import skipIf

from Cython.Build import IpythonMagic
from Cython.TestUtils import CythonTest
from Cython.Compiler.Annotate import AnnotationCCodeWriter

try:
    import IPython.testing.globalipapp
except ImportError:
    # Disable tests and fake helpers for initialisation below.
    def skip_if_not_installed(_):
        return None
else:
    def skip_if_not_installed(c):
        return c

# not using IPython's decorators here because they depend on "nose"
skip_win32 = skipIf(sys.platform == 'win32', "Skip on Windows")
skip_py27 = skipIf(sys.version_info[:2] == (2,7), "Disabled in Py2.7")

try:
    # disable IPython history thread before it gets started to avoid having to clean it up
    from IPython.core.history import HistoryManager
    HistoryManager.enabled = False
except ImportError:
    pass


@contextmanager
def capture_output():
    backup = sys.stdout, sys.stderr
    try:
        replacement = [
             io.TextIOWrapper(io.BytesIO(), encoding=sys.stdout.encoding),
             io.TextIOWrapper(io.BytesIO(), encoding=sys.stderr.encoding),
        ]
        sys.stdout, sys.stderr = replacement
        output = []
        yield output
    finally:
        sys.stdout, sys.stderr = backup
        for wrapper in replacement:
            wrapper.seek(0)  # rewind
            output.append(wrapper.read())
            wrapper.close()


code = u"""\
def f(x):
    return 2*x
"""

cython3_code = u"""\
def f(int x):
    return 2 / x

def call(x):
    return f(*(x,))
"""

pgo_cython3_code = cython3_code + u"""\
def main():
    for _ in range(100): call(5)
main()
"""

compile_error_code = u'''\
cdef extern from *:
    """
    xxx a=1;
    """
    int a;
def doit():
    return a
'''

compile_warning_code = u'''\
cdef extern from *:
    """
    #pragma message ( "CWarning" )
    int a = 42;
    """
    int a;
def doit():
    return a
'''


@skip_if_not_installed
class TestIPythonMagic(CythonTest):

    @classmethod
    def setUpClass(cls):
        CythonTest.setUpClass()
        cls._ip = IPython.testing.globalipapp.get_ipython()

    def setUp(self):
        CythonTest.setUp(self)
        self._ip.extension_manager.load_extension('cython')

    def test_cython_inline(self):
        ip = self._ip
        ip.ex('a=10; b=20')
        result = ip.run_cell_magic('cython_inline', '', 'return a+b')
        self.assertEqual(result, 30)

    @skip_win32
    def test_cython_pyximport(self):
        ip = self._ip
        module_name = '_test_cython_pyximport'
        ip.run_cell_magic('cython_pyximport', module_name, code)
        ip.ex('g = f(10)')
        self.assertEqual(ip.user_ns['g'], 20.0)
        ip.run_cell_magic('cython_pyximport', module_name, code)
        ip.ex('h = f(-10)')
        self.assertEqual(ip.user_ns['h'], -20.0)
        try:
            os.remove(module_name + '.pyx')
        except OSError:
            pass

    def test_cython(self):
        ip = self._ip
        ip.run_cell_magic('cython', '', code)
        ip.ex('g = f(10)')
        self.assertEqual(ip.user_ns['g'], 20.0)

    def test_cython_name(self):
        # The Cython module named 'mymodule' defines the function f.
        ip = self._ip
        ip.run_cell_magic('cython', '--name=mymodule', code)
        # This module can now be imported in the interactive namespace.
        ip.ex('import mymodule; g = mymodule.f(10)')
        self.assertEqual(ip.user_ns['g'], 20.0)

    def test_cython_language_level(self):
        # The Cython cell defines the functions f() and call().
        ip = self._ip
        ip.run_cell_magic('cython', '', cython3_code)
        ip.ex('g = f(10); h = call(10)')
        if sys.version_info[0] < 3:
            self.assertEqual(ip.user_ns['g'], 2 // 10)
            self.assertEqual(ip.user_ns['h'], 2 // 10)
        else:
            self.assertEqual(ip.user_ns['g'], 2.0 / 10.0)
            self.assertEqual(ip.user_ns['h'], 2.0 / 10.0)

    def test_cython3(self):
        # The Cython cell defines the functions f() and call().
        ip = self._ip
        ip.run_cell_magic('cython', '-3', cython3_code)
        ip.ex('g = f(10); h = call(10)')
        self.assertEqual(ip.user_ns['g'], 2.0 / 10.0)
        self.assertEqual(ip.user_ns['h'], 2.0 / 10.0)

    def test_cython2(self):
        # The Cython cell defines the functions f() and call().
        ip = self._ip
        ip.run_cell_magic('cython', '-2', cython3_code)
        ip.ex('g = f(10); h = call(10)')
        self.assertEqual(ip.user_ns['g'], 2 // 10)
        self.assertEqual(ip.user_ns['h'], 2 // 10)

    def test_cython_compile_error_shown(self):
        ip = self._ip
        with capture_output() as out:
            ip.run_cell_magic('cython', '-3', compile_error_code)
        captured_out, captured_err = out

        # it could be that c-level output is captured by distutil-extension
        # (and not by us) and is printed to stdout:
        captured_all = captured_out + "\n" + captured_err
        self.assertTrue("error" in captured_all, msg="error in " + captured_all)

    def test_cython_link_error_shown(self):
        ip = self._ip
        with capture_output() as out:
            ip.run_cell_magic('cython', '-3 -l=xxxxxxxx', code)
        captured_out, captured_err = out

        # it could be that c-level output is captured by distutil-extension
        # (and not by us) and is printed to stdout:
        captured_all = captured_out + "\n!" + captured_err
        self.assertTrue("error" in captured_all, msg="error in " + captured_all)

    def test_cython_warning_shown(self):
        ip = self._ip
        with capture_output() as out:
            # force rebuild, otherwise no warning as after the first success
            # no build step is performed
            ip.run_cell_magic('cython', '-3 -f', compile_warning_code)
        captured_out, captured_err = out

        # check that warning was printed to stdout even if build hasn't failed
        self.assertTrue("CWarning" in captured_out)

    @skip_py27  # Not strictly broken in Py2.7 but currently fails in CI due to C compiler issues.
    @skip_win32
    def test_cython3_pgo(self):
        # The Cython cell defines the functions f() and call().
        ip = self._ip
        ip.run_cell_magic('cython', '-3 --pgo', pgo_cython3_code)
        ip.ex('g = f(10); h = call(10); main()')
        self.assertEqual(ip.user_ns['g'], 2.0 / 10.0)
        self.assertEqual(ip.user_ns['h'], 2.0 / 10.0)

    @skip_win32
    def test_extlibs(self):
        ip = self._ip
        code = u"""
from libc.math cimport sin
x = sin(0.0)
        """
        ip.user_ns['x'] = 1
        ip.run_cell_magic('cython', '-l m', code)
        self.assertEqual(ip.user_ns['x'], 0)


    def test_cython_verbose(self):
        ip = self._ip
        ip.run_cell_magic('cython', '--verbose', code)
        ip.ex('g = f(10)')
        self.assertEqual(ip.user_ns['g'], 20.0)

    def test_cython_verbose_thresholds(self):
        @contextmanager
        def mock_distutils():
            class MockLog:
                DEBUG = 1
                INFO = 2
                thresholds = [INFO]

                def set_threshold(self, val):
                    self.thresholds.append(val)
                    return self.thresholds[-2]


            new_log = MockLog()
            old_log = IpythonMagic.distutils.log
            try:
                IpythonMagic.distutils.log = new_log
                yield new_log
            finally:
                IpythonMagic.distutils.log = old_log

        ip = self._ip
        with mock_distutils() as verbose_log:
            ip.run_cell_magic('cython', '--verbose', code)
            ip.ex('g = f(10)')
        self.assertEqual(ip.user_ns['g'], 20.0)
        self.assertEqual([verbose_log.INFO, verbose_log.DEBUG, verbose_log.INFO],
                          verbose_log.thresholds)

        with mock_distutils() as normal_log:
            ip.run_cell_magic('cython', '', code)
            ip.ex('g = f(10)')
        self.assertEqual(ip.user_ns['g'], 20.0)
        self.assertEqual([normal_log.INFO], normal_log.thresholds)

    def test_cython_no_annotate(self):
        ip = self._ip
        html = ip.run_cell_magic('cython', '', code)
        self.assertTrue(html is None)

    def test_cython_annotate(self):
        ip = self._ip
        html = ip.run_cell_magic('cython', '--annotate', code)
        # somewhat brittle way to differentiate between annotated htmls
        # with/without complete source code:
        self.assertTrue(AnnotationCCodeWriter.COMPLETE_CODE_TITLE not in html.data)

    def test_cython_annotate_default(self):
        ip = self._ip
        html = ip.run_cell_magic('cython', '-a', code)
        # somewhat brittle way to differentiate between annotated htmls
        # with/without complete source code:
        self.assertTrue(AnnotationCCodeWriter.COMPLETE_CODE_TITLE not in html.data)

    def test_cython_annotate_complete_c_code(self):
        ip = self._ip
        html = ip.run_cell_magic('cython', '--annotate-fullc', code)
        # somewhat brittle way to differentiate between annotated htmls
        # with/without complete source code:
        self.assertTrue(AnnotationCCodeWriter.COMPLETE_CODE_TITLE in html.data)
