Moves cyanide int tests to t/integration

t/integration/app.py

@@ -0,0 +1,6 @@
+from __future__ import absolute_import, unicode_literals
+
+from cyanide.app import app
+from cyanide import tasks
+
+__all__ = ['app', 'tasks']

t/integration/conftest.py

@@ -0,0 +1,54 @@
+from __future__ import absolute_import, unicode_literals
+
+import pytest
+import socket
+import sys
+
+from itertools import count
+
+from celery.exceptions import TimeoutError
+from celery.utils.text import truncate
+from cyanide.suite import ManagerMixin
+from cyanide.tasks import marker
+
+
+def _celerymark(app, redis_results=None, **kwargs):
+    if redis_results and not app.conf.result_backend.startswith('redis'):
+        pytest.skip('Test needs Redis result backend.')
+
+
+@pytest.fixture
+def app(request):
+    from .app import app
+    app.finalize()
+    app.set_current()
+    mark = request.node.get_marker('celery')
+    mark = mark and mark.kwargs or {}
+    _celerymark(app, **mark)
+    yield app
+
+
+@pytest.fixture
+def manager(app):
+    with CeleryManager(app) as manager:
+        yield manager
+
+
+class CeleryManager(ManagerMixin):
+
+    # we don't stop full suite when a task result is missing.
+    TaskPredicate = AssertionError
+
+    def __init__(self, app, no_join=False, **kwargs):
+        self.app = app
+        self.no_join = no_join
+        self._init_manager(app, **kwargs)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc_info):
+        self.close()
+
+    def close(self):
+        pass

t/integration/test_canvas.py

@@ -0,0 +1,106 @@
+from __future__ import absolute_import, unicode_literals
+
+import pytest
+
+from celery import chain, group, uuid
+
+from cyanide.tasks import add, collect_ids, ids
+
+
+class test_chain:
+
+    def test_simple_chain(self, manager):
+        c = add.s(4, 4) | add.s(8) | add.s(16)
+        assert manager.join(c()) == 32
+
+    def test_complex_chain(self, manager):
+        c = (
+            add.s(2, 2) | (
+                add.s(4) | add.s(8) | add.s(16)
+            ) |
+            group(add.s(i) for i in range(4))
+        )
+        res = c()
+        assert res.get() == [32, 33, 34, 35]
+
+    def test_parent_ids(self, manager, num=10):
+        c = chain(ids.si(i) for i in range(num))
+        c.freeze()
+        res = c()
+        res.get(timeout=5)
+        self.assert_ids(res, num - 1)
+
+    def assert_ids(self, res, size):
+        i, root = size, res
+        while root.parent:
+            root = root.parent
+        node = res
+        while node:
+            root_id, parent_id, value = node.get(timeout=5)
+            assert value == i
+            assert root_id == root.id
+            if node.parent:
+                assert parent_id == node.parent.id
+            node = node.parent
+            i -= 1
+
+
+class test_group:
+
+    def test_parent_ids(self):
+        g = ids.si(1) | ids.si(2) | group(ids.si(i) for i in range(2, 50))
+        res = g()
+        expected_root_id = res.parent.parent.id
+        expected_parent_id = res.parent.id
+        values = res.get(timeout=5)
+
+        for i, r in enumerate(values):
+            root_id, parent_id, value = r
+            assert root_id == expected_root_id
+            assert parent_id == expected_parent_id
+            assert value == i + 2
+
+
+class xxx_chord:
+
+    @pytest.mark.celery(redis_results=1)
+    def test_parent_ids(self, manager):
+        self.assert_parentids_chord()
+        self.assert_parentids_chord(uuid(), uuid())
+
+    def assert_parentids_chord(self, base_root=None, base_parent=None):
+        g = (
+            ids.si(1) |
+            ids.si(2) |
+            group(ids.si(i) for i in range(3, 50)) |
+            collect_ids.s(i=50) |
+            ids.si(51)
+        )
+        g.freeze(root_id=base_root, parent_id=base_parent)
+        res = g.apply_async(root_id=base_root, parent_id=base_parent)
+        expected_root_id = base_root or res.parent.parent.parent.id
+
+        root_id, parent_id, value = res.get(timeout=5)
+        assert value == 51
+        assert root_id == expected_root_id
+        assert parent_id == res.parent.id
+
+        prev, (root_id, parent_id, value) = res.parent.get(timeout=5)
+        assert value == 50
+        assert root_id == expected_root_id
+        assert parent_id == res.parent.parent.id
+
+        for i, p in enumerate(prev):
+            root_id, parent_id, value = p
+            assert root_id == expected_root_id
+            assert parent_id == res.parent.parent.id
+
+        root_id, parent_id, value = res.parent.parent.get(timeout=5)
+        assert value == 2
+        assert parent_id == res.parent.parent.parent.id
+        assert root_id == expected_root_id
+
+        root_id, parent_id, value = res.parent.parent.parent.get(timeout=5)
+        assert value == 1
+        assert root_id == expected_root_id
+        assert parent_id == base_parent

t/integration/test_tasks.py

@@ -0,0 +1,18 @@
+from __future__ import absolute_import, unicode_literals
+from celery import group
+from cyanide.tasks import print_unicode, sleeping
+
+
+class test_tasks:
+
+    def test_task_accepted(self, manager, sleep=1):
+        r1 = sleeping.delay(sleep)
+        sleeping.delay(sleep)
+        manager.assert_accepted([r1.id])
+
+    def test_unicode_task(self, manager):
+        manager.join(
+            group(print_unicode.s() for _ in range(5))(),
+            timeout=1, propagate=True,
+        )
+