test futures ascompleted timing out before completing it s final future alreadycompleted cancelledandnotifiedfuture exceptionfuture successfulfuture windows clock resolution is around 15 6 ms shorttimeout 0 100 for timeout in 0 shorttimeout with self subtesttimeout completedfutures set future self executor submittime sleep shorttimeout 10 try for f in futures ascompleted alreadycompleted future timeout completedfutures addf except futures timeouterror pass check that future wasn t completed self assertequalcompletedfutures alreadycompleted def testduplicatefuturesself issue 20367 duplicate futures should not raise exceptions or give duplicate responses issue 31641 accept arbitrary iterables future1 self executor submittime sleep 2 completed f for f in futures ascompleteditertools repeatfuture1 3 self assertequallencompleted 1 def testfreereferenceyieldedfutureself issue 14406 generator should not keep references to finished futures futureslist future for in range8 futureslist appendcreatefuturestatecancelledandnotified futureslist appendcreatefuturestatefinished result42 with self assertraisesfutures timeouterror for future in futures ascompletedfutureslist timeout0 futureslist removefuture wr weakref reffuture del future support gccollect for pypy or other gcs self assertisnonewr futureslist0 setresulttest for future in futures ascompletedfutureslist futureslist removefuture wr weakref reffuture del future support gccollect for pypy or other gcs self assertisnonewr if futureslist futureslist0 setresulttest def testcorrecttimeoutexceptionmsgself futureslist cancelledandnotifiedfuture pendingfuture runningfuture successfulfuture with self assertraisesfutures timeouterror as cm listfutures ascompletedfutureslist timeout0 self assertequalstrcm exception 2 of 4 futures unfinished createexecutortestsglobals ascompletedtests def setupmodule setupmodule if name main unittest main test futures as_completed timing out before completing it s final future windows clock resolution is around 15 6 ms check that future wasn t completed issue 20367 duplicate futures should not raise exceptions or give duplicate responses issue 31641 accept arbitrary iterables issue 14406 generator should not keep references to finished futures for pypy or other gcs for pypy or other gcs
import itertools
import time
import unittest
import weakref
from concurrent import futures
from concurrent.futures._base import (
    CANCELLED_AND_NOTIFIED, FINISHED, Future)

from test import support

from .util import (
    PENDING_FUTURE, RUNNING_FUTURE,
    CANCELLED_AND_NOTIFIED_FUTURE, EXCEPTION_FUTURE, SUCCESSFUL_FUTURE,
    create_future, create_executor_tests, setup_module)


def mul(x, y):
    return x * y


class AsCompletedTests:
    def test_no_timeout(self):
        future1 = self.executor.submit(mul, 2, 21)
        future2 = self.executor.submit(mul, 7, 6)

        completed = set(futures.as_completed(
                [CANCELLED_AND_NOTIFIED_FUTURE,
                 EXCEPTION_FUTURE,
                 SUCCESSFUL_FUTURE,
                 future1, future2]))
        self.assertEqual(set(
                [CANCELLED_AND_NOTIFIED_FUTURE,
                 EXCEPTION_FUTURE,
                 SUCCESSFUL_FUTURE,
                 future1, future2]),
                completed)

    def test_future_times_out(self):
        
        already_completed = {CANCELLED_AND_NOTIFIED_FUTURE,
                             EXCEPTION_FUTURE,
                             SUCCESSFUL_FUTURE}

        
        short_timeout = 0.100
        for timeout in (0, short_timeout):
            with self.subTest(timeout):

                completed_futures = set()
                future = self.executor.submit(time.sleep, short_timeout * 10)

                try:
                    for f in futures.as_completed(
                        already_completed | {future},
                        timeout
                    ):
                        completed_futures.add(f)
                except futures.TimeoutError:
                    pass

                
                self.assertEqual(completed_futures, already_completed)

    def test_duplicate_futures(self):
        
        
        
        future1 = self.executor.submit(time.sleep, 2)
        completed = [
            f for f in futures.as_completed(itertools.repeat(future1, 3))
        ]
        self.assertEqual(len(completed), 1)

    def test_free_reference_yielded_future(self):
        
        
        futures_list = [Future() for _ in range(8)]
        futures_list.append(create_future(state=CANCELLED_AND_NOTIFIED))
        futures_list.append(create_future(state=FINISHED, result=42))

        with self.assertRaises(futures.TimeoutError):
            for future in futures.as_completed(futures_list, timeout=0):
                futures_list.remove(future)
                wr = weakref.ref(future)
                del future
                support.gc_collect()  
                self.assertIsNone(wr())

        futures_list[0].set_result("test")
        for future in futures.as_completed(futures_list):
            futures_list.remove(future)
            wr = weakref.ref(future)
            del future
            support.gc_collect()  
            self.assertIsNone(wr())
            if futures_list:
                futures_list[0].set_result("test")

    def test_correct_timeout_exception_msg(self):
        futures_list = [CANCELLED_AND_NOTIFIED_FUTURE, PENDING_FUTURE,
                        RUNNING_FUTURE, SUCCESSFUL_FUTURE]

        with self.assertRaises(futures.TimeoutError) as cm:
            list(futures.as_completed(futures_list, timeout=0))

        self.assertEqual(str(cm.exception), '2 (of 4) futures unfinished')


create_executor_tests(globals(), AsCompletedTests)


def setUpModule():
    setup_module()


if __name__ == "__main__":
    unittest.main()