Function test_serialization

python/pathway/tests/test_py_object_wrapper.py:221–300

(tmp_path: Path, serialization: str, port: int)

)
@needs_multiprocessing_fork
def test_serialization(tmp_path: Path, serialization: str, port: int) -> None:
    input_path = tmp_path / "in.csv"
    output_path = tmp_path / "out.csv"

    with open(input_path, "w") as f:
        f.write(
            """instance,a
0,ab
2,def
2,x
"""
        )

    class InputSchema(pw.Schema):
        instance: int
        a: str

    class OutputSchema(pw.Schema):
        res: str

    def target():
        t = pw.io.csv.read(input_path, schema=InputSchema, mode="static")
        z = t.copy()

        serializer = get_serializer(serialization)

        @pw.udf
        def create_simple(a: str) -> pw.PyObjectWrapper[SimpleStr]:
            return pw.wrap_py_object(SimpleStr(a), serializer=serializer)

        @pw.udf
        def use_python_object(a: pw.PyObjectWrapper[SimpleStr], x: str) -> str:
            return a.value.concat(x)

        res = (
            t.with_columns(s=create_simple(pw.this.a))
            .join(z, left_instance=pw.left.instance, right_instance=pw.right.instance)
            .select(res=use_python_object(pw.left.s, pw.right.a))
        )
        pw.io.csv.write(res, output_path)
        run()

    @dataclass
    class Checker:
        error: AssertionError | None = None

        def __call__(self) -> bool:
            if not output_path.exists():
                return False
            try:
                G.clear()
                result = pw.io.csv.read(output_path, schema=OutputSchema, mode="static")
                expected = pw.debug.table_from_markdown(
                    """
                    res
                    abab
                    defx
                    defdef