Method test_stamping_workflow

t/integration/test_canvas.py:3373–3442

(self, manager, subtests)

class test_stamping_mechanism:
    def test_stamping_workflow(self, manager, subtests):
        try:
            manager.app.backend.ensure_chords_allowed()
        except NotImplementedError as e:
            raise pytest.skip(e.args[0])

        workflow = group(
            add.s(1, 2) | add.s(3),
            add.s(4, 5) | add.s(6),
            identity.si(21),
        ) | group(
            xsum.s(),
            xsum.s(),
        )

        @task_received.connect
        def task_received_handler(request=None, **kwargs):
            nonlocal assertion_result
            link = None
            if request._Request__payload[2]["callbacks"]:
                link = signature(request._Request__payload[2]["callbacks"][0])
            link_error = None
            if request._Request__payload[2]["errbacks"]:
                link_error = signature(request._Request__payload[2]["errbacks"][0])

            assertion_result = all(
                [
                    assertion_result,
                    [stamped_header in request.stamps for stamped_header in request.stamped_headers],
                    [
                        stamped_header in link.options
                        for stamped_header in link.options["stamped_headers"]
                        if link  # the link itself doesn't have a link
                    ],
                    [
                        stamped_header in link_error.options
                        for stamped_header in link_error.options["stamped_headers"]
                        if link_error  # the link_error itself doesn't have a link_error
                    ],
                ]
            )

        @before_task_publish.connect
        def before_task_publish_handler(
            body=None,
            headers=None,
            **kwargs,
        ):
            nonlocal assertion_result

            assertion_result = all(
                [stamped_header in headers["stamps"] for stamped_header in headers["stamped_headers"]]
            )

        class CustomStampingVisitor(StampingVisitor):
            def on_signature(self, sig, **headers) -> dict:
                return {"on_signature": 42}