Function run_computation

python/pathway/tests/test_deduplicate.py:375–408

(nb_rows: int, offset: int, expected: list[int])

    )

    def run_computation(nb_rows: int, offset: int, expected: list[int]):
        G.clear()

        def acceptor(new_value, old_value) -> bool:
            return new_value >= old_value + 2

        class InputSchema(pw.Schema):
            value: float

        class StreamSubject(pw.io.python.ConnectorSubject):
            def run(self):
                for index in range(offset, offset + nb_rows):
                    self.next_json({"value": index})
                    # Make sure that the data can't be batched
                    time.sleep(0.01)
                    self.commit()

        table = pw.io.python.read(
            StreamSubject(),
            schema=InputSchema,
            format="json",
        )
        result = table.deduplicate(value=pw.this.value, acceptor=acceptor)

        emit = mock.Mock()

        def on_change(key: pw.Pointer, row: dict, time: int, is_addition: bool):
            if is_addition:
                emit(row["value"])

        pw.io.subscribe(result, on_change)
        run(persistence_config=persistence_config)

        emit.assert_has_calls([mock.call(i) for i in expected])

    run_computation(6, 0, [0, 2, 4])
    run_computation(5, 6, [6, 8, 10])