Function test_stream_generator_from_markdown

python/pathway/tests/test_io.py:2289–2327

()

@pytest.mark.filterwarnings("ignore:timestamps are required to be even")
def test_stream_generator_from_markdown():
    stream_generator = pw.debug.StreamGenerator()
    t = stream_generator.table_from_markdown(
        """
       | colA | colB | _time
    1  | 1    | 2    | 1
    5  | 2    | 3    | 1
    10 | 5    | 1    | 2
    """
    )
    on_change = mock.Mock()
    pw.io.subscribe(t, on_change=on_change)

    run(persistence_config=stream_generator.persistence_config())

    on_change.assert_has_calls(
        [
            mock.call.on_change(
                key=api.ref_scalar(1),
                row={"colA": 1, "colB": 2},
                time=2,
                is_addition=True,
            ),
            mock.call.on_change(
                key=api.ref_scalar(5),
                row={"colA": 2, "colB": 3},
                time=2,
                is_addition=True,
            ),
            mock.call.on_change(
                key=api.ref_scalar(10),
                row={"colA": 5, "colB": 1},
                time=4,
                is_addition=True,
            ),
        ],
        any_order=True,
    )
    assert on_change.call_count == 3


def test_stream_generator_from_markdown_with_diffs():