import importlib
import pickle
import threading
from textwrap import dedent
import unittest
import time

from test.support import import_helper
# Raise SkipTest if subinterpreters not supported.
_channels = import_helper.import_module('_interpchannels')
from test.support import interpreters
from test.support.interpreters import channels
from .utils import _run_output, TestBase


class LowLevelTests(TestBase):

    # The behaviors in the low-level module is important in as much
    # as it is exercised by the high-level module.  Therefore the
    # most # important testing happens in the high-level tests.
    # These low-level tests cover corner cases that are not
    # encountered by the high-level module, thus they
    # mostly shouldn't matter as much.

    # Additional tests are found in Lib/test/test__interpchannels.py.
    # XXX Those should be either moved to LowLevelTests or eliminated
    # in favor of high-level tests in this file.

    def test_highlevel_reloaded(self):
        # See gh-115490 (https://github.com/python/cpython/issues/115490).
        importlib.reload(channels)


class TestChannels(TestBase):

    def test_create(self):
        r, s = channels.create()
        self.assertIsInstance(r, channels.RecvChannel)
        self.assertIsInstance(s, channels.SendChannel)

    def test_list_all(self):
        self.assertEqual(channels.list_all(), [])
        created = set()
        for _ in range(3):
            ch = channels.create()
            created.add(ch)
        after = set(channels.list_all())
        self.assertEqual(after, created)

    def test_shareable(self):
        interp = interpreters.create()
        rch, sch = channels.create()

        self.assertTrue(
            interpreters.is_shareable(rch))
        self.assertTrue(
            interpreters.is_shareable(sch))

        sch.send_nowait(rch)
        sch.send_nowait(sch)
        rch2 = rch.recv()
        sch2 = rch.recv()

        interp.prepare_main(rch=rch, sch=sch)
        sch.send_nowait(rch)
        sch.send_nowait(sch)
        interp.exec(dedent("""
            rch2 = rch.recv()
            sch2 = rch.recv()
            assert rch2 == rch
            assert sch2 == sch

            sch.send_nowait(rch2)
            sch.send_nowait(sch2)
            """))
        rch3 = rch.recv()
        sch3 = rch.recv()

        self.assertEqual(rch2, rch)
        self.assertEqual(sch2, sch)
        self.assertEqual(rch3, rch)
        self.assertEqual(sch3, sch)

    def test_is_closed(self):
        rch, sch = channels.create()
        rbefore = rch.is_closed
        sbefore = sch.is_closed
        rch.close()
        rafter = rch.is_closed
        safter = sch.is_closed

        self.assertFalse(rbefore)
        self.assertFalse(sbefore)
        self.assertTrue(rafter)
        self.assertTrue(safter)


class TestRecvChannelAttrs(TestBase):

    def test_id_type(self):
        rch, _ = channels.create()
        self.assertIsInstance(rch.id, _channels.ChannelID)

    def test_custom_id(self):
        rch = channels.RecvChannel(1)
        self.assertEqual(rch.id, 1)

        with self.assertRaises(TypeError):
            channels.RecvChannel('1')

    def test_id_readonly(self):
        rch = channels.RecvChannel(1)
        with self.assertRaises(AttributeError):
            rch.id = 2

    def test_equality(self):
        ch1, _ = channels.create()
        ch2, _ = channels.create()
        self.assertEqual(ch1, ch1)
        self.assertNotEqual(ch1, ch2)

    def test_pickle(self):
        ch, _ = channels.create()
        data = pickle.dumps(ch)
        unpickled = pickle.loads(data)
        self.assertEqual(unpickled, ch)


class TestSendChannelAttrs(TestBase):

    def test_id_type(self):
        _, sch = channels.create()
        self.assertIsInstance(sch.id, _channels.ChannelID)

    def test_custom_id(self):
        sch = channels.SendChannel(1)
        self.assertEqual(sch.id, 1)

        with self.assertRaises(TypeError):
            channels.SendChannel('1')

    def test_id_readonly(self):
        sch = channels.SendChannel(1)
        with self.assertRaises(AttributeError):
            sch.id = 2

    def test_equality(self):
        _, ch1 = channels.create()
        _, ch2 = channels.create()
        self.assertEqual(ch1, ch1)
        self.assertNotEqual(ch1, ch2)

    def test_pickle(self):
        _, ch = channels.create()
        data = pickle.dumps(ch)
        unpickled = pickle.loads(data)
        self.assertEqual(unpickled, ch)


class TestSendRecv(TestBase):

    def test_send_recv_main(self):
        r, s = channels.create()
        orig = b'spam'
        s.send_nowait(orig)
        obj = r.recv()

        self.assertEqual(obj, orig)
        self.assertIsNot(obj, orig)

    def test_send_recv_same_interpreter(self):
        interp = interpreters.create()
        interp.exec(dedent("""
            from test.support.interpreters import channels
            r, s = channels.create()
            orig = b'spam'
            s.send_nowait(orig)
            obj = r.recv()
            assert obj == orig, 'expected: obj == orig'
            assert obj is not orig, 'expected: obj is not orig'
            """))

    @unittest.skip('broken (see BPO-...)')
    def test_send_recv_different_interpreters(self):
        r1, s1 = channels.create()
        r2, s2 = channels.create()
        orig1 = b'spam'
        s1.send_nowait(orig1)
        out = _run_output(
            interpreters.create(),
            dedent(f"""
                obj1 = r.recv()
                assert obj1 == b'spam', 'expected: obj1 == orig1'
                # When going to another interpreter we get a copy.
                assert id(obj1) != {id(orig1)}, 'expected: obj1 is not orig1'
                orig2 = b'eggs'
                print(id(orig2))
                s.send_nowait(orig2)
                """),
            channels=dict(r=r1, s=s2),
            )
        obj2 = r2.recv()

        self.assertEqual(obj2, b'eggs')
        self.assertNotEqual(id(obj2), int(out))

    def test_send_recv_different_threads(self):
        r, s = channels.create()

        def f():
            while True:
                try:
                    obj = r.recv()
                    break
                except channels.ChannelEmptyError:
                    time.sleep(0.1)
            s.send(obj)
        t = threading.Thread(target=f)
        t.start()

        orig = b'spam'
        s.send(orig)
        obj = r.recv()
        t.join()

        self.assertEqual(obj, orig)
        self.assertIsNot(obj, orig)

    def test_send_recv_nowait_main(self):
        r, s = channels.create()
        orig = b'spam'
        s.send_nowait(orig)
        obj = r.recv_nowait()

        self.assertEqual(obj, orig)
        self.assertIsNot(obj, orig)

    def test_send_recv_nowait_main_with_default(self):
        r, _ = channels.create()
        obj = r.recv_nowait(None)

        self.assertIsNone(obj)

    def test_send_recv_nowait_same_interpreter(self):
        interp = interpreters.create()
        interp.exec(dedent("""
            from test.support.interpreters import channels
            r, s = channels.create()
            orig = b'spam'
            s.send_nowait(orig)
            obj = r.recv_nowait()
            assert obj == orig, 'expected: obj == orig'
            # When going back to the same interpreter we get the same object.
            assert obj is not orig, 'expected: obj is not orig'
            """))

    @unittest.skip('broken (see BPO-...)')
    def test_send_recv_nowait_different_interpreters(self):
        r1, s1 = channels.create()
        r2, s2 = channels.create()
        orig1 = b'spam'
        s1.send_nowait(orig1)
        out = _run_output(
            interpreters.create(),
            dedent(f"""
                obj1 = r.recv_nowait()
                assert obj1 == b'spam', 'expected: obj1 == orig1'
                # When going to another interpreter we get a copy.
                assert id(obj1) != {id(orig1)}, 'expected: obj1 is not orig1'
                orig2 = b'eggs'
                print(id(orig2))
                s.send_nowait(orig2)
                """),
            channels=dict(r=r1, s=s2),
            )
        obj2 = r2.recv_nowait()

        self.assertEqual(obj2, b'eggs')
        self.assertNotEqual(id(obj2), int(out))

    def test_recv_timeout(self):
        r, _ = channels.create()
        with self.assertRaises(TimeoutError):
            r.recv(timeout=1)

    def test_recv_channel_does_not_exist(self):
        ch = channels.RecvChannel(1_000_000)
        with self.assertRaises(channels.ChannelNotFoundError):
            ch.recv()

    def test_send_channel_does_not_exist(self):
        ch = channels.SendChannel(1_000_000)
        with self.assertRaises(channels.ChannelNotFoundError):
            ch.send(b'spam')

    def test_recv_nowait_channel_does_not_exist(self):
        ch = channels.RecvChannel(1_000_000)
        with self.assertRaises(channels.ChannelNotFoundError):
            ch.recv_nowait()

    def test_send_nowait_channel_does_not_exist(self):
        ch = channels.SendChannel(1_000_000)
        with self.assertRaises(channels.ChannelNotFoundError):
            ch.send_nowait(b'spam')

    def test_recv_nowait_empty(self):
        ch, _ = channels.create()
        with self.assertRaises(channels.ChannelEmptyError):
            ch.recv_nowait()

    def test_recv_nowait_default(self):
        default = object()
        rch, sch = channels.create()
        obj1 = rch.recv_nowait(default)
        sch.send_nowait(None)
        sch.send_nowait(1)
        sch.send_nowait(b'spam')
        sch.send_nowait(b'eggs')
        obj2 = rch.recv_nowait(default)
        obj3 = rch.recv_nowait(default)
        obj4 = rch.recv_nowait()
        obj5 = rch.recv_nowait(default)
        obj6 = rch.recv_nowait(default)

        self.assertIs(obj1, default)
        self.assertIs(obj2, None)
        self.assertEqual(obj3, 1)
        self.assertEqual(obj4, b'spam')
        self.assertEqual(obj5, b'eggs')
        self.assertIs(obj6, default)

    def test_send_buffer(self):
        buf = bytearray(b'spamspamspam')
        obj = None
        rch, sch = channels.create()

        def f():
            nonlocal obj
            while True:
                try:
                    obj = rch.recv()
                    break
                except channels.ChannelEmptyError:
                    time.sleep(0.1)
        t = threading.Thread(target=f)
        t.start()

        sch.send_buffer(buf)
        t.join()

        self.assertIsNot(obj, buf)
        self.assertIsInstance(obj, memoryview)
        self.assertEqual(obj, buf)

        buf[4:8] = b'eggs'
        self.assertEqual(obj, buf)
        obj[4:8] = b'ham.'
        self.assertEqual(obj, buf)

    def test_send_buffer_nowait(self):
        buf = bytearray(b'spamspamspam')
        rch, sch = channels.create()
        sch.send_buffer_nowait(buf)
        obj = rch.recv()

        self.assertIsNot(obj, buf)
        self.assertIsInstance(obj, memoryview)
        self.assertEqual(obj, buf)

        buf[4:8] = b'eggs'
        self.assertEqual(obj, buf)
        obj[4:8] = b'ham.'
        self.assertEqual(obj, buf)

    def test_send_cleared_with_subinterpreter(self):
        def common(rch, sch, unbound=None, presize=0):
            if not unbound:
                extraargs = ''
            elif unbound is channels.UNBOUND:
                extraargs = ', unbound=channels.UNBOUND'
            elif unbound is channels.UNBOUND_ERROR:
                extraargs = ', unbound=channels.UNBOUND_ERROR'
            elif unbound is channels.UNBOUND_REMOVE:
                extraargs = ', unbound=channels.UNBOUND_REMOVE'
            else:
                raise NotImplementedError(repr(unbound))
            interp = interpreters.create()

            _run_output(interp, dedent(f"""
                from test.support.interpreters import channels
                sch = channels.SendChannel({sch.id})
                obj1 = b'spam'
                obj2 = b'eggs'
                sch.send_nowait(obj1{extraargs})
                sch.send_nowait(obj2{extraargs})
                """))
            self.assertEqual(
                _channels.get_count(rch.id),
                presize + 2,
            )

            if presize == 0:
                obj1 = rch.recv()
                self.assertEqual(obj1, b'spam')
                self.assertEqual(
                    _channels.get_count(rch.id),
                    presize + 1,
                )

            return interp

        with self.subTest('default'):  # UNBOUND
            rch, sch = channels.create()
            interp = common(rch, sch)
            del interp
            self.assertEqual(_channels.get_count(rch.id), 1)
            obj1 = rch.recv()
            self.assertEqual(_channels.get_count(rch.id), 0)
            self.assertIs(obj1, channels.UNBOUND)
            self.assertEqual(_channels.get_count(rch.id), 0)
            with self.assertRaises(channels.ChannelEmptyError):
                rch.recv_nowait()

        with self.subTest('UNBOUND'):
            rch, sch = channels.create()
            interp = common(rch, sch, channels.UNBOUND)
            del interp
            self.assertEqual(_channels.get_count(rch.id), 1)
            obj1 = rch.recv()
            self.assertIs(obj1, channels.UNBOUND)
            self.assertEqual(_channels.get_count(rch.id), 0)
            with self.assertRaises(channels.ChannelEmptyError):
                rch.recv_nowait()

        with self.subTest('UNBOUND_ERROR'):
            rch, sch = channels.create()
            interp = common(rch, sch, channels.UNBOUND_ERROR)

            del interp
            self.assertEqual(_channels.get_count(rch.id), 1)
            with self.assertRaises(channels.ItemInterpreterDestroyed):
                rch.recv()

            self.assertEqual(_channels.get_count(rch.id), 0)
            with self.assertRaises(channels.ChannelEmptyError):
                rch.recv_nowait()

        with self.subTest('UNBOUND_REMOVE'):
            rch, sch = channels.create()

            interp = common(rch, sch, channels.UNBOUND_REMOVE)
            del interp
            self.assertEqual(_channels.get_count(rch.id), 0)
            with self.assertRaises(channels.ChannelEmptyError):
                rch.recv_nowait()

            sch.send_nowait(b'ham', unbound=channels.UNBOUND_REMOVE)
            self.assertEqual(_channels.get_count(rch.id), 1)
            interp = common(rch, sch, channels.UNBOUND_REMOVE, 1)
            self.assertEqual(_channels.get_count(rch.id), 3)
            sch.send_nowait(42, unbound=channels.UNBOUND_REMOVE)
            self.assertEqual(_channels.get_count(rch.id), 4)
            del interp
            self.assertEqual(_channels.get_count(rch.id), 2)
            obj1 = rch.recv()
            obj2 = rch.recv()
            self.assertEqual(obj1, b'ham')
            self.assertEqual(obj2, 42)
            self.assertEqual(_channels.get_count(rch.id), 0)
            with self.assertRaises(channels.ChannelEmptyError):
                rch.recv_nowait()

    def test_send_cleared_with_subinterpreter_mixed(self):
        rch, sch = channels.create()
        interp = interpreters.create()

        # If we don't associate the main interpreter with the channel
        # then the channel will be automatically closed when interp
        # is destroyed.
        sch.send_nowait(None)
        rch.recv()
        self.assertEqual(_channels.get_count(rch.id), 0)

        _run_output(interp, dedent(f"""
            from test.support.interpreters import channels
            sch = channels.SendChannel({sch.id})
            sch.send_nowait(1, unbound=channels.UNBOUND)
            sch.send_nowait(2, unbound=channels.UNBOUND_ERROR)
            sch.send_nowait(3)
            sch.send_nowait(4, unbound=channels.UNBOUND_REMOVE)
            sch.send_nowait(5, unbound=channels.UNBOUND)
            """))
        self.assertEqual(_channels.get_count(rch.id), 5)

        del interp
        self.assertEqual(_channels.get_count(rch.id), 4)

        obj1 = rch.recv()
        self.assertIs(obj1, channels.UNBOUND)
        self.assertEqual(_channels.get_count(rch.id), 3)

        with self.assertRaises(channels.ItemInterpreterDestroyed):
            rch.recv()
        self.assertEqual(_channels.get_count(rch.id), 2)

        obj2 = rch.recv()
        self.assertIs(obj2, channels.UNBOUND)
        self.assertEqual(_channels.get_count(rch.id), 1)

        obj3 = rch.recv()
        self.assertIs(obj3, channels.UNBOUND)
        self.assertEqual(_channels.get_count(rch.id), 0)

    def test_send_cleared_with_subinterpreter_multiple(self):
        rch, sch = channels.create()
        interp1 = interpreters.create()
        interp2 = interpreters.create()

        sch.send_nowait(1)
        _run_output(interp1, dedent(f"""
            from test.support.interpreters import channels
            rch = channels.RecvChannel({rch.id})
            sch = channels.SendChannel({sch.id})
            obj1 = rch.recv()
            sch.send_nowait(2, unbound=channels.UNBOUND)
            sch.send_nowait(obj1, unbound=channels.UNBOUND_REMOVE)
            """))
        _run_output(interp2, dedent(f"""
            from test.support.interpreters import channels
            rch = channels.RecvChannel({rch.id})
            sch = channels.SendChannel({sch.id})
            obj2 = rch.recv()
            obj1 = rch.recv()
            """))
        self.assertEqual(_channels.get_count(rch.id), 0)
        sch.send_nowait(3)
        _run_output(interp1, dedent("""
            sch.send_nowait(4, unbound=channels.UNBOUND)
            # interp closed here
            sch.send_nowait(5, unbound=channels.UNBOUND_REMOVE)
            sch.send_nowait(6, unbound=channels.UNBOUND)
            """))
        _run_output(interp2, dedent("""
            sch.send_nowait(7, unbound=channels.UNBOUND_ERROR)
            # interp closed here
            sch.send_nowait(obj1, unbound=channels.UNBOUND_ERROR)
            sch.send_nowait(obj2, unbound=channels.UNBOUND_REMOVE)
            sch.send_nowait(8, unbound=channels.UNBOUND)
            """))
        _run_output(interp1, dedent("""
            sch.send_nowait(9, unbound=channels.UNBOUND_REMOVE)
            sch.send_nowait(10, unbound=channels.UNBOUND)
            """))
        self.assertEqual(_channels.get_count(rch.id), 10)

        obj3 = rch.recv()
        self.assertEqual(obj3, 3)
        self.assertEqual(_channels.get_count(rch.id), 9)

        obj4 = rch.recv()
        self.assertEqual(obj4, 4)
        self.assertEqual(_channels.get_count(rch.id), 8)

        del interp1
        self.assertEqual(_channels.get_count(rch.id), 6)

        # obj5 was removed

        obj6 = rch.recv()
        self.assertIs(obj6, channels.UNBOUND)
        self.assertEqual(_channels.get_count(rch.id), 5)

        obj7 = rch.recv()
        self.assertEqual(obj7, 7)
        self.assertEqual(_channels.get_count(rch.id), 4)

        del interp2
        self.assertEqual(_channels.get_count(rch.id), 3)

        # obj1
        with self.assertRaises(channels.ItemInterpreterDestroyed):
            rch.recv()
        self.assertEqual(_channels.get_count(rch.id), 2)

        # obj2 was removed

        obj8 = rch.recv()
        self.assertIs(obj8, channels.UNBOUND)
        self.assertEqual(_channels.get_count(rch.id), 1)

        # obj9 was removed

        obj10 = rch.recv()
        self.assertIs(obj10, channels.UNBOUND)
        self.assertEqual(_channels.get_count(rch.id), 0)


if __name__ == '__main__':
    # Test needs to be a package, so we can do relative imports.
    unittest.main()