Function test_subscribe

python/pathway/tests/test_io.py:612–638

()

def test_subscribe():
    class TestSubject(pw.io.python.ConnectorSubject):
        def run(self):
            self.next(data="foo")

    table = pw.io.python.read(TestSubject(), format="raw")

    root = mock.Mock()

    pw.io.subscribe(
        table,
        on_change=root.on_change,
        on_time_end=root.on_time_end,
        on_end=root.on_end,
    )

    run()

    root.assert_has_calls(
        [
            mock.call.on_change(
                key=mock.ANY, row={"data": "foo"}, time=mock.ANY, is_addition=True
            ),
            mock.call.on_time_end(mock.ANY),
            mock.call.on_end(),
        ]
    )


def test_python_write():