import weakref
import gc
import threading
import unittest

from succession import _Chain, _SuccessionIterator, Succession, TimeoutError


class TestSuccession(unittest.TestCase):
    def test_chain(self):
        chain = _Chain()
        chain.push(2)
        self.assertEqual(chain.wait_result(), 2)

    def test_chain_iter(self):
        head = _Chain()
        chain = head

        for i in [1, 2, 3, 4, 5]:
            chain = chain.push(i)
        chain.close()

        self.assertEqual(list(_SuccessionIterator(head)), [1, 2, 3, 4, 5])

    def test_memory(self):
        # Make sure that chains don't hold references to previous links
        chain = _Chain()
        head = weakref.ref(chain)
        for i in range(1000):
            chain = chain.push(i)
        gc.collect()
        self.assertIsNone(head())

    def test_iter_memory(self):
        # Make sure that chain iterators do not hold a reference to the head
        chain = _Chain()

        def push_1000(chain):
            for i in range(1000):
                chain = chain.push(i)
        t = threading.Thread(target=push_1000, args=(chain,), daemon=True)

        iterator = _SuccessionIterator(chain)
        chain = weakref.ref(chain)

        t.start()
        for i in range(1000):
            next(iterator)

        t.join()

        gc.collect()
        self.assertIsNone(chain())

    def test_succession(self):
        succession = Succession()

        for i in [1, 2, 3, 4, 5]:
            succession.push(i)
        succession.close()

        self.assertEqual(list(succession), [1, 2, 3, 4, 5])

    def test_zero_timeout(self):
        succession = Succession()

        for i in [1, 2, 3, 4, 5]:
            succession.push(i)

        self.assertEqual(list(succession.iter(timeout=0)), [1, 2, 3, 4, 5])

    def test_nonzero_timeout(self):
        succession = Succession()

        for i in [1, 2, 3, 4, 5]:
            succession.push(i)

        result = []
        try:
            for item in succession.iter(timeout=0.01):
                result.append(item)
        except TimeoutError:
            self.assertEqual(result, [1, 2, 3, 4, 5])
        else:
            self.fail()

    def test_release_iter(self):
        succession = Succession()
        root = weakref.ref(succession._root)
        iterator = weakref.ref(iter(succession))

        for i in [1, 2, 3, 4, 5]:
            succession.push(i)
        succession.drop()

        gc.collect()
        self.assertIsNone(root())
        self.assertIsNone(iterator())

    def test_compress(self):
        succession = Succession(compress=lambda items: [sum(items)])

        from_start = succession.iter()

        for i in [1, 2, 3, 4, 5]:
            succession.push(i)
        succession.close()

        from_end = succession.iter()

        self.assertEqual(list(from_start), [1, 2, 3, 4, 5])
        self.assertEqual(list(from_end), [15])

    def test_head(self):
        succession = Succession()

        self.assertEqual(list(succession.head()), [])

        for i in [1, 2, 3, 4, 5]:
            succession.push(i)

        self.assertEqual(list(succession.head()), [1, 2, 3, 4, 5])

        succession.close()

    def test_drop(self):
        succession = Succession()
        root = weakref.ref(succession._root)

        for i in [1, 2, 3, 4, 5]:
            succession.push(i)
        succession.drop()
        gc.collect()
        self.assertIsNone(root())

        for i in [6, 7, 8, 9, 10]:
            succession.push(i)
        succession.close()

        self.assertEqual(list(succession), [6, 7, 8, 9, 10])

    def test_echo(self):
        req = Succession()
        res = Succession()

        def echo():
            for m in req:
                res.push(m)
            res.close()

        t = threading.Thread(target=echo)
        t.start()

        res_iter = iter(res)

        req.push(1)
        self.assertEqual(next(res_iter), 1)
        req.push(2)
        self.assertEqual(next(res_iter), 2)
        req.push(3)
        self.assertEqual(next(res_iter), 3)
        req.close()

        t.join()


loader = unittest.TestLoader()
suite = unittest.TestSuite((
    loader.loadTestsFromTestCase(TestSuccession),
))

@@ -2899,24 +2899,536 @@
 nd), %5B15%5D)%0A%0A
+    def test_drop_after_push(self):%0A        succession = Succession(drop=True)%0A%0A        from_start = succession.iter()%0A%0A        for i in %5B1, 2, 3%5D:%0A            succession.push(i)%0A%0A        from_middle = succession.iter()%0A%0A        for i in %5B4, 5, 6%5D:%0A            succession.push(i)%0A%0A        succession.close()%0A%0A        from_end = succession.iter()%0A%0A        self.assertEqual(list(from_start), %5B1, 2, 3, 4, 5, 6%5D)%0A        self.assertEqual(list(from_middle), %5B4, 5, 6%5D)%0A        self.assertEqual(list(from_end), %5B%5D)%0A%0A%0A
     def test

add test for drop after push