# tag: ipython

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


import builtins
import os
import io
import sys
from contextlib import contextmanager
from unittest import skipIf, SkipTest

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


# not using IPython's decorators here because they depend on "nose"
skip_win32 = skipIf(sys.platform == 'win32', "Skip on Windows")


@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 = """\
def f(x):
    return 2*x
"""

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

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

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

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

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


class TestIPythonMagic(CythonTest):
    _orig_builtins = None

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        # IPython modifies the builtins, so keep a clean copy before the import.
        orig_builtins = dict(builtins.__dict__)

        try:
            import IPython.testing.globalipapp
        except ImportError as exc:
            # Disable tests and fake helpers for initialisation below.
            raise SkipTest(f"IPython is not installed: {exc}")

        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

        cls._orig_builtins = orig_builtins
        cls._ip = IPython.testing.globalipapp.get_ipython()

    @classmethod
    def tearDownClass(cls):
        # Clean up builtins from IPython left-overs.
        if cls._orig_builtins:
            for name in list(vars(builtins)):
                if name not in cls._orig_builtins:
                    delattr(builtins, name)
            for name, value in cls._orig_builtins.items():
                if getattr(builtins, name, None) is not value:
                    setattr(builtins, name, value)

        super().tearDownClass()

    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)')
        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_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 = """
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)