Function do_run

python/pathway/tests/test_persistence_iterate.py:100–120

(changes, expected_diffs)

    events: dict = {}

    def do_run(changes, expected_diffs):
        for eid, edata in changes.items():
            events[eid] = edata
            et, flag, data = edata
            write_lines(
                input_path / f"{eid}.csv",
                ["event_time,flag,data", f"{et},{flag},{data}"],
            )

        G.clear()
        t = pw.io.csv.read(input_path, schema=EventSchema, mode="static")
        res = pipeline_fn(t)
        pw.io.csv.write(res, output_path)
        run(
            persistence_config=pw.persistence.Config(
                pw.persistence.Backend.filesystem(pstorage_path),
                persistence_mode=mode,
            )
        )
        _assert_diffs_are_unit(output_path)
        _assert_diffs_match(output_path, expected_diffs)

    return events, do_run