Function test_skips

python/pathway/tests/test_transformers.py:114–164

()

def test_skips():
    @transformer
    class list_traversal:
        class nodes(ClassArg):
            next = input_attribute()
            val = input_attribute()

        class requests(ClassArg):
            node = input_attribute()
            steps = input_attribute()

            @output_attribute
            def reached_node(self) -> int:
                node = self.transformer.nodes[self.node]
                for _ in range(self.steps):
                    node = self.transformer.nodes[node.next]
                return node.id

            @output_attribute
            def reached_value(self) -> int:
                node = self.transformer.nodes[self.reached_node]
                return node.val

    nodes = T(
        """
            | next | val
        1   | 2    | 11
        2   | 3    | 12
        3   |      | 13
        """
    ).with_columns(next=this.pointer_from(this.next))

    requests = T(
        """
            | node | steps
        10  | 1    | 1
        20  | 3    | 0
        """
    ).with_columns(node=nodes.pointer_from(this.node))

    replies = list_traversal(nodes, requests).requests
    assert_table_equality_wo_types(  # XXX: no Pointer in schema
        replies,
        T(
            """
                | reached_node  | reached_value
            10  | 2             | 12
            20  | 3             | 13
            """
        ).with_columns(reached_node=nodes.pointer_from(this.reached_node)),
    )


def test_output_attribute_rename():