
from contextlib import asynccontextmanager
from pytest import mark
from subprocess import PIPE
from traitlets.config.loader import Config
from tornado.testing import gen_test

from .. import MaximumKernelsException
from ..mapping import LimitedPooledMappingKernelManager


from .utils import shutdown_all_direct, TestAsyncKernelManager

# Test that it works as normal with default config
class TestMappingKernelManagerUnused(TestAsyncKernelManager):
    __test__ = True

    @asynccontextmanager
    async def _get_tcp_km(self):
        c = Config()
        km = LimitedPooledMappingKernelManager(config=c)
        try:
            yield km
        finally:
            km.shutdown_all()

    # Mapping manager doesn't handle this:
    @mark.skip()
    @gen_test
    async def test_tcp_lifecycle_with_kernel_id(self):
        pass



# Test that it works with a max that is larger than pool size
class TestMappingKernelManagerApplied(TestAsyncKernelManager):
    __test__ = True

    @asynccontextmanager
    async def _get_tcp_km(self):
        c = Config()
        c.LimitedKernelManager.max_kernels = 4
        c.LimitedPooledMappingKernelManager.kernel_pool_size = 2
        c.LimitedPooledMappingKernelManager.pool_kwargs = dict(stdout=PIPE, stderr=PIPE)
        km = LimitedPooledMappingKernelManager(config=c)
        try:
            yield km
        finally:
            km.shutdown_all()

    # Mapping manager doesn't handle this:
    @mark.skip()
    @gen_test
    async def test_tcp_lifecycle_with_kernel_id(self):
        pass

    @gen_test
    async def test_exceed_pool_size(self):
        async with self._get_tcp_km() as km:
            self.assertEqual(len(km._pool), 2)
            kids = []
            for i in range(4):
                kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
                self.assertIn(kid, km)
                kids.append(kid)
                self.assertEqual(len(km._pool), 2)

            shutdown_all_direct(km)
            for kid in kids:
                self.assertNotIn(kid, km)

            # Cycle again to assure the pool survives that
            kids = []
            for i in range(4):
                kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
                self.assertIn(kid, km)
                kids.append(kid)
                self.assertEqual(len(km._pool), 2)

            km.shutdown_all()
            for kid in kids:
                self.assertNotIn(kid, km)

    @gen_test
    async def test_breach_max(self):
        async with self._get_tcp_km() as km:
            kids = []
            for i in range(4):
                kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
                self.assertIn(kid, km)
                kids.append(kid)

            with self.assertRaises(MaximumKernelsException):
                await km.start_kernel(stdout=PIPE, stderr=PIPE)

            # Remove and add one to make sure we correctly recovered
            km.shutdown_kernel(kid)
            self.assertNotIn(kid, km)
            kids.pop()

            kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
            self.assertIn(kid, km)
            kids.append(kid)

            km.shutdown_all()
            for kid in kids:
                self.assertNotIn(kid, km)


from contextlib import asynccontextmanager
from pytest import mark
from subprocess import PIPE
from traitlets.config.loader import Config
from tornado.testing import gen_test

from .. import MaximumKernelsException
from ..mapping import LimitedPooledMappingKernelManager


from .utils import shutdown_all_direct, TestAsyncKernelManager

# Test that it works as normal with default config
class TestMappingKernelManagerUnused(TestAsyncKernelManager):
    __test__ = True

    @asynccontextmanager
    async def _get_tcp_km(self):
        c = Config()
        km = LimitedPooledMappingKernelManager(config=c)
        try:
            yield km
        finally:
            km.shutdown_all()

    # Mapping manager doesn't handle this:
    @mark.skip()
    @gen_test
    async def test_tcp_lifecycle_with_kernel_id(self):
        pass



# Test that it works with a max that is larger than pool size
class TestMappingKernelManagerApplied(TestAsyncKernelManager):
    __test__ = True

    @asynccontextmanager
    async def _get_tcp_km(self):
        c = Config()
        c.LimitedKernelManager.max_kernels = 4
        c.LimitedPooledMappingKernelManager.kernel_pool_size = 2
        c.LimitedPooledMappingKernelManager.pool_kwargs = dict(stdout=PIPE, stderr=PIPE)
        km = LimitedPooledMappingKernelManager(config=c)
        try:
            yield km
        finally:
            km.shutdown_all()

    # Mapping manager doesn't handle this:
    @mark.skip()
    @gen_test
    async def test_tcp_lifecycle_with_kernel_id(self):
        pass

    @gen_test
    async def test_exceed_pool_size(self):
        async with self._get_tcp_km() as km:
            self.assertEqual(len(km._pool), 2)
            kids = []
            for i in range(4):
                kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
                self.assertIn(kid, km)
                kids.append(kid)
                self.assertEqual(len(km._pool), 2)

            shutdown_all_direct(km)
            for kid in kids:
                self.assertNotIn(kid, km)

            # Cycle again to assure the pool survives that
            kids = []
            for i in range(4):
                kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
                self.assertIn(kid, km)
                kids.append(kid)
                self.assertEqual(len(km._pool), 2)

            km.shutdown_all()
            for kid in kids:
                self.assertNotIn(kid, km)

    @gen_test
    async def test_breach_max(self):
        async with self._get_tcp_km() as km:
            kids = []
            for i in range(4):
                kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
                self.assertIn(kid, km)
                kids.append(kid)

            with self.assertRaises(MaximumKernelsException):
                await km.start_kernel(stdout=PIPE, stderr=PIPE)

            # Remove and add one to make sure we correctly recovered
            km.shutdown_kernel(kid)
            self.assertNotIn(kid, km)
            kids.pop()

            kid = await km.start_kernel(stdout=PIPE, stderr=PIPE)
            self.assertIn(kid, km)
            kids.append(kid)

            km.shutdown_all()
            for kid in kids:
                self.assertNotIn(kid, km)

# Test that it works as normal with default config # Mapping manager doesn't handle this: # Test that it works with a max that is larger than pool size # Mapping manager doesn't handle this: # Cycle again to assure the pool survives that # Remove and add one to make sure we correctly recovered