Repository: pathwaycom/pathway
File: integration_tests/webserver/test_rest_connector.py (Python)

Code before:
            """,
            output_path,
            usecols=["sum", "diff"],
            index_col=["key"],
        ),
        10,
    )

    t.join()


def test_server_fail_on_duplicate_port(tmp_path: pathlib.Path, port: int) -> None:
    output_path = tmp_path / "output.csv"

    class InputSchema(pw.Schema):
        k: int
        v: int

    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
    queries, response_writer = pw.io.http.rest_connector(
        webserver=webserver, schema=InputSchema, delete_completed_queries=False
    )
    response_writer(queries.select(query_id=queries.id, result=pw.this.v))

    queries_dup, response_writer_dup = pw.io.http.rest_connector(
        webserver=webserver, schema=InputSchema, delete_completed_queries=False
    )
    response_writer_dup(queries_dup.select(query_id=queries_dup.id, result=pw.this.v))

    sum = queries.groupby(pw.this.k).reduce(
        key=pw.this.k, sum=pw.reducers.sum(pw.this.v)
    )
    sum_dup = queries_dup.groupby(pw.this.k).reduce(
        key=pw.this.k, sum=pw.reducers.sum(pw.this.v)
    )

    pw.io.csv.write(sum, output_path)
    pw.io.csv.write(sum_dup, output_path)

    with pytest.raises(OSError) as exc_info:
        pw.run()
    error_message = str(exc_info.value)
    assert (
        "error while attempting to bind on address" in error_message
        or "Address already in use" in error_message
    )


def _test_server_two_endpoints(
    tmp_path: pathlib.Path, port: int, with_cors: bool
) -> None:

Review comment (refactor):
This part is a little bit trickier.

When one uses `PathwayWebserver`, the error message is actually different: `"Added route will never be executed, method POST is already registered"`. So it wouldn't match the `assert` at the end of the test.

The second tricky part is that the error will be spawned not at the time of `pw.run`, but earlier, at the moment the duplicate endpoint is added via `pw.io.http.rest_connector`. So you need to catch an exception there and check it against the new error pattern.

As a consequence (a good one), this test no longer needs to run to find out that there's an error. Perhaps we can move it to the unit tests that run more often? Could you please try to place it in `python/pathway/tests/test_io.py` and check if it runs normally? If so, let's move it there.

Code after:

    wait_result_with_checker(
        expect_csv_checker(
            """
            key | sum    | diff
            1   | 1      | 1
            1   | 1      | -1
            1   | 3      | 1
            """,
            output_path,
            usecols=["sum", "diff"],
            index_col=["key"],
        ),
        10,
    )

    t.join()


def test_server_fail_on_duplicate_port(tmp_path: pathlib.Path, port: int) -> None:
    class InputSchema(pw.Schema):
        k: int
        v: int

    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
    queries, response_writer = pw.io.http.rest_connector(
        webserver=webserver, schema=InputSchema, delete_completed_queries=False
    )
    response_writer(queries.select(query_id=queries.id, result=pw.this.v))

    with pytest.raises(RuntimeError, match="Added route will never be executed"):
        queries_dup, response_writer_dup = pw.io.http.rest_connector(
            webserver=webserver, schema=InputSchema, delete_completed_queries=False
        )


def _test_server_two_endpoints(
    tmp_path: pathlib.Path, port: int, with_cors: bool
) -> None:
    output_path = tmp_path / "output.csv"

    class InputSchema(pw.Schema):
        query: str
        user: str

    def uppercase_logic(queries: pw.Table) -> pw.Table:
        return queries.select(
            query_id=queries.id, result=pw.apply(lambda x: x.upper(), pw.this.query)
        )

    def duplicate_logic(queries: pw.Table) -> pw.Table: