Function test_simple

python/pathway/tests/test_async_transformer.py:34–68

(monkeypatch)

def test_simple(monkeypatch):
    monkeypatch.delenv("PATHWAY_PERSISTENT_STORAGE", raising=False)

    class OutputSchema(pw.Schema):
        ret: int

    class TestAsyncTransformer(pw.AsyncTransformer, output_schema=OutputSchema):
        async def invoke(self, value: int) -> dict[str, Any]:
            await asyncio.sleep(random.uniform(0, 0.1))
            return dict(ret=value + 1)

    input_table = T(
        """
            | value
        1   | 1
        2   | 2
        3   | 3
        """
    )

    result = TestAsyncTransformer(input_table=input_table).successful

    assert result._universe.is_subset_of(input_table._universe)

    assert_table_equality(
        result,
        T(
            """
            | ret
        1   | 2
        2   | 3
        3   | 4
        """
        ),
    )


def test_file_io(monkeypatch, tmp_path: pathlib.Path):