# Path: mapchete/_executor.py
# class Executor:
#     """
#     Executor factory for dask and concurrent.futures executor
# 
#     Will move into the mapchete core package.
#     """
# 
#     def __new__(cls, *args, concurrency=None, **kwargs):
#         if concurrency == "dask":
#             try:
#                 return DaskExecutor(*args, **kwargs)
#             except ImportError as exc:  # pragma: no cover
#                 raise ImportError(
#                     "this feature requires the mapchete[dask] extra"
#                 ) from exc
# 
#         elif concurrency is None or kwargs.get("max_workers") == 1:
#             return SequentialExecutor(*args, **kwargs)
# 
#         elif concurrency in ["processes", "threads"]:
#             return ConcurrentFuturesExecutor(*args, concurrency=concurrency, **kwargs)
# 
#         else:  # pragma: no cover
#             raise ValueError(
#                 f"concurrency must be one of None, 'processes', 'threads' or 'dask', not {concurrency}"
#             )
# 
# class SkippedFuture:
#     """Wrapper class to mimick future interface for empty tasks."""
# 
#     def __init__(self, result=None, skip_info=None, **kwargs):
#         self._result = result
#         self.skip_info = skip_info
# 
#     def result(self):
#         """Only return initial result value."""
#         return self._result
# 
#     def exception(self):  # pragma: no cover
#         """Nothing to raise here."""
#         return
# 
#     def cancelled(self):  # pragma: no cover
#         """Nothing to cancel here."""
#         return False
# 
#     def __repr__(self):  # pragma: no cover
#         """Return string representation."""
#         return f"<SkippedFuture: type: {type(self._result)}, exception: {type(self._exception)})"
# 
# Path: mapchete/_executor.py
# class FakeFuture:
#     """Wrapper class to mimick future interface."""
# 
#     def __init__(self, func, fargs=None, fkwargs=None):
#         """Set attributes."""
#         fargs = fargs or []
#         fkwargs = fkwargs or {}
#         try:
#             self._result, self._exception = func(*fargs, **fkwargs), None
#         except Exception as e:  # pragma: no cover
#             self._result, self._exception = None, e
# 
#     def result(self):
#         """Return task result."""
#         if self._exception:
#             logger.exception(self._exception)
#             raise self._exception
#         return self._result
# 
#     def exception(self):
#         """Raise task exception if any."""
#         return self._exception
# 
#     def cancelled(self):  # pragma: no cover
#         """Sequential futures cannot be cancelled."""
#         return False
# 
#     def __repr__(self):  # pragma: no cover
#         """Return string representation."""
#         return f"<FakeFuture: type: {type(self._result)}, exception: {type(self._exception)})"


import pytest
import time
from mapchete import Executor, SkippedFuture
from mapchete._executor import FakeFuture

        assert not executor.running_futures


def test_dask_executor_as_completed_unlimited_max_tasks():
    items = 100
    with Executor(concurrency="dask") as executor:
        # abort
        for future in executor.as_completed(
            _dummy_process, range(items), max_submitted_tasks=None
        ):
            assert future.result()

        assert not executor.running_futures


def test_concurrent_futures_dask_executor_map():
    items = list(range(10))
    with Executor(concurrency="dask") as executor:
        result = executor.map(_dummy_process, items)
        assert [i + 1 for i in items] == result


def test_fake_future():
    def task(*args, **kwargs):
        return True

    def failing_task(*args, **kwargs):
        raise RuntimeError()


# Path: mapchete/_executor.py
# class Executor:
#     """
#     Executor factory for dask and concurrent.futures executor
# 
#     Will move into the mapchete core package.
#     """
# 
#     def __new__(cls, *args, concurrency=None, **kwargs):
#         if concurrency == "dask":
#             try:
#                 return DaskExecutor(*args, **kwargs)
#             except ImportError as exc:  # pragma: no cover
#                 raise ImportError(
#                     "this feature requires the mapchete[dask] extra"
#                 ) from exc
# 
#         elif concurrency is None or kwargs.get("max_workers") == 1:
#             return SequentialExecutor(*args, **kwargs)
# 
#         elif concurrency in ["processes", "threads"]:
#             return ConcurrentFuturesExecutor(*args, concurrency=concurrency, **kwargs)
# 
#         else:  # pragma: no cover
#             raise ValueError(
#                 f"concurrency must be one of None, 'processes', 'threads' or 'dask', not {concurrency}"
#             )
# 
# class SkippedFuture:
#     """Wrapper class to mimick future interface for empty tasks."""
# 
#     def __init__(self, result=None, skip_info=None, **kwargs):
#         self._result = result
#         self.skip_info = skip_info
# 
#     def result(self):
#         """Only return initial result value."""
#         return self._result
# 
#     def exception(self):  # pragma: no cover
#         """Nothing to raise here."""
#         return
# 
#     def cancelled(self):  # pragma: no cover
#         """Nothing to cancel here."""
#         return False
# 
#     def __repr__(self):  # pragma: no cover
#         """Return string representation."""
#         return f"<SkippedFuture: type: {type(self._result)}, exception: {type(self._exception)})"
# 
# Path: mapchete/_executor.py
# class FakeFuture:
#     """Wrapper class to mimick future interface."""
# 
#     def __init__(self, func, fargs=None, fkwargs=None):
#         """Set attributes."""
#         fargs = fargs or []
#         fkwargs = fkwargs or {}
#         try:
#             self._result, self._exception = func(*fargs, **fkwargs), None
#         except Exception as e:  # pragma: no cover
#             self._result, self._exception = None, e
# 
#     def result(self):
#         """Return task result."""
#         if self._exception:
#             logger.exception(self._exception)
#             raise self._exception
#         return self._result
# 
#     def exception(self):
#         """Raise task exception if any."""
#         return self._exception
# 
#     def cancelled(self):  # pragma: no cover
#         """Sequential futures cannot be cancelled."""
#         return False
# 
#     def __repr__(self):  # pragma: no cover
#         """Return string representation."""
#         return f"<FakeFuture: type: {type(self._result)}, exception: {type(self._exception)})"


# Path: test/test_executor.py
import pytest
import time
from mapchete import Executor, SkippedFuture
from mapchete._executor import FakeFuture


        assert not executor.running_futures


def test_dask_executor_as_completed_unlimited_max_tasks():
    items = 100
    with Executor(concurrency="dask") as executor:
        # abort
        for future in executor.as_completed(
            _dummy_process, range(items), max_submitted_tasks=None
        ):
            assert future.result()

        assert not executor.running_futures


def test_concurrent_futures_dask_executor_map():
    items = list(range(10))
    with Executor(concurrency="dask") as executor:
        result = executor.map(_dummy_process, items)
        assert [i + 1 for i in items] == result


def test_fake_future():
    def task(*args, **kwargs):
        return True

    def failing_task(*args, **kwargs):
        raise RuntimeError()


    future = FakeFuture(task, fargs=[1, True], fkwargs=dict(foo="bar"))